How to test a flask_tryton route

Hello,

I’d like to integrate some tests in a Tryton Flask application, using flask_tryton.

I can’t quite find how to manage having a “test” transaction and the “route” transaction… since the route transaction asserts on None values… Has someone integrated any test like this or could give any pointers? Thank you

❯ python -m unittest discover -s tests
/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/passlib/utils/__init__.py:854: DeprecationWarning: 'crypt' is deprecated and slated for removal in Python 3.13
  from crypt import crypt as _crypt
<frozen importlib._bootstrap>:1049: ImportWarning: PluginImportFixer.find_spec() not found; falling back to find_module()
<frozen importlib._bootstrap>:283: DeprecationWarning: the load_module() method is deprecated and slated for removal in Python 3.12; use exec_module() instead
.F........................
======================================================================
FAIL: test_login (test_application.B2BApplicationTestCase.test_login)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/trytond/tests/test_tryton.py", line 209, in wrapper
    result = func(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/Developer/ack-eus/tryton/applications/b2b/tests/test_application.py", line 92, in test_login
    response = self.client.post('/login', data={
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/werkzeug/test.py", line 1145, in post
    return self.open(*args, **kw)
           ^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask/testing.py", line 223, in open
    response = super().open(
               ^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/werkzeug/test.py", line 1094, in open
    response = self.run_wsgi_app(request.environ, buffered=buffered)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/werkzeug/test.py", line 961, in run_wsgi_app
    rv = run_wsgi_app(self.application, environ, buffered=buffered)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/werkzeug/test.py", line 1242, in run_wsgi_app
    app_rv = app(environ, start_response)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask/app.py", line 2548, in __call__
    return self.wsgi_app(environ, start_response)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask/app.py", line 2528, in wsgi_app
    response = self.handle_exception(e)
               ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask/app.py", line 2525, in wsgi_app
    response = self.full_dispatch_request()
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask/app.py", line 1822, in full_dispatch_request
    rv = self.handle_user_exception(e)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask/app.py", line 1820, in full_dispatch_request
    rv = self.dispatch_request()
         ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask/app.py", line 1796, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask_tryton.py", line 37, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/flask_tryton.py", line 196, in wrapper
    with Transaction().start(database, transaction_user,
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hodeinavarro/.local/share/ack-eus/environments/zafra/lib/python3.11/site-packages/trytond/transaction.py", line 112, in start
    assert self.user is None
           ^^^^^^^^^^^^^^^^^
AssertionError

----------------------------------------------------------------------
Ran 26 tests in 77.532s

FAILED (failures=1)
b2b/application/__init__.py
from flask import Flask

def create_app(config=None):
    app = Flask(__name__)
    if config is not None:
        app.config.from_pyfile(config)

    from .tryton import tryton
    tryton.init_app(app)

    from .views import view

    app.register_blueprint(view)

    return app
b2b/application/views.py
from functools import wraps

from flask import Blueprint, redirect, render_template, request, url_for
from werkzeug.wrappers import Response as BaseResponse

from .tryton import tryton

view = Blueprint('view', __name__, 'static', '/', 'templates')


def protected(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        LOGIN_RESPONSE = redirect(url_for('view.login', r=request.path))

        key = request.cookies.get('session')
        if key is None:
            return LOGIN_RESPONSE
        return validate_session(key, f(*args, **kwargs), LOGIN_RESPONSE)
    return wrapper


@tryton.transaction()
def validate_session(key: str, success_response: BaseResponse,
        LOGIN_RESPONSE: BaseResponse) -> BaseResponse:
    WebUserSession = tryton.pool.get('web.user.session')
    user = WebUserSession.get_user(key)
    if user is None:
        response = LOGIN_RESPONSE
        response.delete_cookie('session')
        return response
    return success_response


def authenticate(username: str, password: str) -> BaseResponse:
    WebUser = tryton.pool.get('web.user')
    WebUserSession = tryton.pool.get('web.user.session')

    user = WebUser.authenticate(username, password)

    if not user:
        # TODO: Add flash message
        return redirect(url_for('view.login', r=request.args.get('r')))

    session = user.new_session()
    response = redirect(request.args.get('r') or url_for('view.index'))
    response.set_cookie('session', session, max_age=WebUserSession.timeout())
    return response


@view.route('/login', methods={'GET', 'POST'})
@tryton.transaction()
def login():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')

        if username and password:
            return authenticate(username, password)
        else:
            # TODO: Add flash message
            return redirect(url_for('view.login', r=request.args.get('r')))

    return render_template('login.html')


@view.route('/')
@protected
def index():
    return 'Hello, World!'
b2b/application/config.test.py
from application.config import get_secret_key, get_tryton_database

TESTING = True
SECRET_KEY = get_secret_key()
TRYTON_DATABASE = get_tryton_database()

b2b/tests/test_application.py

import unittest

from application import create_app
from trytond.config import config
from trytond.pool import Pool
from trytond.tests.test_tryton import (activate_module, drop_db,
    with_transaction)


class B2BApplicationTestCase(unittest.TestCase):
    "Test B2B application"
    module = 'b2b'
    extras = []
    language = 'es'

    @classmethod
    def setUpClass(cls):
        drop_db()
        modules = [cls.module]
        if cls.extras:
            modules.extend(cls.extras)
        activate_module(modules, lang=cls.language)
        super().setUpClass()

    def setUp(self):
        self.app = create_app('config.test.py')
        self.client = self.app.test_client()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        drop_db()

    @with_transaction()
    def test_login(self):
        user = 'test'
        password = 'test'.zfill(int(config.get('password', 'length')))

        pool = Pool()
        WebUser = pool.get('web.user')

        WebUser.create([{
            'email': user,
            'password': password
        }])

        response = self.client.post('/login', data={
            'username': user,
            'password': password
        })

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.location, '/')

I guess the flask application must be configured to have a TRYTON_DATABASE and TRYTON_USER set.

TRYTON_DATABASE is set on create_app function.
TRYTON_USER is set to 0 as default correctly when initialising a Tryton app.

When testing those are also set alongside TESTING = True.

In fact, the problem comes because there is a TRYTON_USER set on the Flask application, since the transaction was started on test_application.py with @with_transaction() and then again tries to start a new one on login route with @tryton.transaction().

So my problem exact problem is that I want to create a web user on test_application.py and then authenticate it on the application, being able to rollback the creation because subsequent tests will not allow me to have a WebUser with same email.

I’m thinking on creating a custom decorator to set up all necessary Tryton information before tests, commit, do the test and then delete all data created on the test db, but wanted to hear if there could be a nicer way to do it.

A possible solution would be to make the transaction decorator re-entrant as long as it is for the same database. In case a transaction has already started, it could just ensure to set the proper parameter like the user and the context.

I’m kind of lost with transactions when working outside my usual workflow…

Do you mean that the Transaction() should join instead of start if the Transaction().database.name equals the TRYTON_DATABASE env var in flask_tryton?

This should be handled in flask_tryton as a feature request, right?
What could I try to test in order to submit a change?

Yes if the transaction has already started, then we should not start a new one.

Yes.

Your test case for example. Indeed this should allow to implement tests for flask-tryton.

By the way we are in the process to also migrate to Heptapod but there is not dead line.

Was there ever an agreed-upon pattern for testing a flask-tryon route?

I do not see in the repo history a change set for the example test mentioned in the most recent response, above.

Personally I put a minimal effort on just maintaining flask-tryton because for me the way to go is REST API for user application

But nobody has yet submitted a change to make transaction decorator re-entrant.

This statement, make me wonder to ask: Do you expect to deprecate someday flask-tryton?

For me “deprecate” has no meaning in the context of “contrib” because there is already no maintenance guarantee.

1 Like

Thank you for the information!

@hodeinavarro Did you find a solution for this?

It should be solved with:

:100: Lucky me. This was jsut fixed 2 weeks ago and merged 1 week ago

:frowning_face: This update did not fix this issue (but another one).

Here is my traceback, which is a bit longer than the original one.

======================================================================
ERROR: test_login (test_application.B2BApplicationTestCase)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/tmp/_venv/lib64/python3.10/site-packages/trytond/tests/test_tryton.py", line 276, in wrapper
    result = func(*args, **kwargs)
  File "/tmp/b2b/tests/test_application.py", line 47, in test_login
    response = self.client.post('/login', data={
  File "/tmp/_venv/lib64/python3.10/site-packages/werkzeug/test.py", line 1167, in post
    return self.open(*args, **kw)
  File "/tmp/_venv/lib64/python3.10/site-packages/flask/testing.py", line 234, in open
    response = super().open(
  File "/tmp/_venv/lib64/python3.10/site-packages/werkzeug/test.py", line 1116, in open
    response_parts = self.run_wsgi_app(request.environ, buffered=buffered)
  File "/tmp/_venv/lib64/python3.10/site-packages/werkzeug/test.py", line 988, in run_wsgi_app
    rv = run_wsgi_app(self.application, environ, buffered=buffered)
  File "/tmp/_venv/lib64/python3.10/site-packages/werkzeug/test.py", line 1264, in run_wsgi_app
    app_rv = app(environ, start_response)
  File "/tmp/_venv/lib64/python3.10/site-packages/flask/app.py", line 1536, in __call__
    return self.wsgi_app(environ, start_response)
  File "/tmp/_venv/lib64/python3.10/site-packages/flask/app.py", line 1514, in wsgi_app
    response = self.handle_exception(e)
  File "/tmp/_venv/lib64/python3.10/site-packages/flask/app.py", line 1511, in wsgi_app
    response = self.full_dispatch_request()
  File "/tmp/_venv/lib64/python3.10/site-packages/flask/app.py", line 919, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/tmp/_venv/lib64/python3.10/site-packages/flask/app.py", line 917, in full_dispatch_request
    rv = self.dispatch_request()
  File "/tmp/_venv/lib64/python3.10/site-packages/flask/app.py", line 902, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)  # type: ignore[no-any-return]
  File "/tmp/_venv/lib64/python3.10/site-packages/flask_tryton.py", line 177, in wrapper
    with Transaction().start(
  File "/tmp/_venv/lib64/python3.10/site-packages/trytond/transaction.py", line 168, in start
    assert self.user is None
AssertionError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/tmp/_venv/lib64/python3.10/site-packages/trytond/tests/test_tryton.py", line 283, in wrapper
    transaction.rollback()
  File "/tmp/_venv/lib64/python3.10/site-packages/trytond/transaction.py", line 358, in rollback
    self.connection.rollback()
AttributeError: 'NoneType' object has no attribute 'rollback'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/tmp/_venv/lib64/python3.10/site-packages/trytond/tests/test_tryton.py", line 272, in wrapper
    with Transaction().start(
  File "/tmp/_venv/lib64/python3.10/site-packages/trytond/transaction.py", line 226, in __exit__
    self.stop(type is None)
  File "/tmp/_venv/lib64/python3.10/site-packages/trytond/transaction.py", line 260, in stop
    transactions.remove(self)
ValueError: list.remove(x): x not in list

----------------------------------------------------------------------
Ran 1 test in 0.428s

FAILED (errors=1)

I was able to work around this by using Transaction(new=True) in flask_tryton.Tryton.transaction(). I made it depending on current_app.config['TESTING'].

If this is an acceptable solution, I will submit a merge request.

I think the problem is because in your test you are creating a tryton transaction and latter the flask application tries to recreate another inside the same transaction and that causes the error. Removing the with_transaction decorator should fix such problem.

Thanks.

Removing the with_transaction decorator requires to create Transactions within each test-case like this:

    #@with_transaction()
    def test_login(self):
        with Transaction().start(None, 1):
            pool = Pool()
            WebUser = pool.get('web.user')
            WebUser.create([{
                'email': 'user@test.example.com',
                'password': 'password1234',
            }])
       # perform some requests, eg. DELETE the user
       with Transaction().start(None, 1):
            pool = Pool()
            WebUser = pool.get('web.user')
            users = WebUser.search(['email', '=', 'user@test.example.com'])
       assert users == []

Which is not quite elegant - and required to change every test-case copied over from other applications not using flask_tryton.

Also, the with_transaction decorator rolls back any transactions and thus allows to reuse the database for the next test.

Any more ideas?

Basically what is needed here are nested transactions. sqlite supports these using SAVEPOINT. Anyhow, Python’s sqlite3 module does not and uses BEGIN DEFERRED - which is not actullay compatible with Savepoints (or maybe is, I didn’t figure out). I tried several hours whether it’s still possible to get this hacked(!) into Tryton (just for use in the test suite) with some respectable intermediate results but no success at last.

So I ended up with dropping and recreating the database for every single test-case and creating Transactions for setup and eventually testing afterwards, see below. This has the same effect (a clean database for every test-case), but is not efficient. Anyhow, when using DB_CACHE in a RAM-based filesystem (e.g. a tmpfs mounted at /tmp) its still “fast”. For test-suites with only a “few” test-cases and not to be run “all the time”, this is quite acceptable IMHO.

class B2BApplicationTestCase(unittest.TestCase):
    "Test B2B application"
    module = 'web_user'
    extras = []
    language = 'es'

    @classmethod
    def setUpClass(cls):
        drop_db()
        cls._modules = [cls.module]
        if cls.extras:
            cls._modules.extend(cls.extras)

    def setUp(self):
        activate_module(self._modules, lang=self.language)
        self.app = create_app('config.test.py')
        self.client = self.app.test_client()

    def tearDown(self):
        drop_db()

    def test_login1(self):
        user = 'test'
        password = 'test'.zfill(int(config.get('password', 'length')))
        with Transaction().start(None, 1):
            pool = Pool()
            WebUser = pool.get('web.user')
            WebUser.create([{
                'email': user,
                'password': password
            }])

        response = self.client.post('/login', data={
            'username': user,
            'password': password
        })

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.location, '/')

As I already said, the proper solution is to make transaction decorator re-entrent.

1 Like

Sure, and I’d be happy is anybody would implement it. For me this is beyond my skills. Esp. since it requires some means of nesting read-only and read-write transactions.