6

I have a mocking problem I wasn't able to solve for hours when used decorator @mock.patch, but solved immediately when tried using context manager. What I'd like to do is to ask are there any more differences in how it's working under the hood? Speaking more I mean more than just setting the mock context.

My example is about mocking database connection for aiohttp server. Test looks as follows:

backend/tests/test_server.py

# source: https://docs.aiohttp.org/en/stable/testing.html#unittest

from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
from alchemy_mock.mocking import AlchemyMagicMock
from backend.app import create_app
from datetime import datetime
from unittest import mock

import aiohttp
import asyncio
import json


class MyAppTestCase(AioHTTPTestCase):
    """
    Class for testing aiohttp server

    Methods
    -------
    ...
    test_get_version()
        Test response for GET /version
    """

    @mock.patch('backend.app.SessionProvider')
    async def get_application(self, mock_sess_prov):
        mock_sess_prov.session = AlchemyMagicMock()
        mock_sess_prov.start_session = lambda: None
        return create_app()

...


    @unittest_run_loop
    async def test_get_version(self):
        """Test response for GET /version

            ------------
            Verification
            Answer comparison
        """
        resp: aiohttp.ClientResponse =\
            await self.client.request("GET", "/version")
        assert resp.status == 200
        text = await resp.text()
        assert "Project version is 0.1" in text

So the problem is with get_application() method. When I mock it as above, I'm getting error because target class SessionProvider isn't mocked and tries to create SqlAlchemy session object. When I rebuild mentioned function as follows:

async def get_application(self):
    with mock.patch('backend.app.SessionProvider') as mock_sess_prov:
        mock_sess_prov.session = AlchemyMagicMock()
        mock_sess_prov.start_session = lambda: None
        return create_app()

everything is okay. To give more view on how my app is working, I pasted my app code and Session Manager:

backend/backend/app.py:

from aiohttp import web
from backend.schema import ReservationSchema
import marshmallow.exceptions
from backend.models import Reservation, ValidationError
from backend.session_provider import SessionProvider
from sqlalchemy.exc import IntegrityError

import json
import logging


async def version(request) -> web.Response:
    """GET /version endpoint

    Returns API version

    Returns
    -------
    web.Response
        Http response object
    """
    text: str = "Project version is 0.1"
    return web.Response(text=text)


async def new_reservation(request) -> web.Response:
    """POST /new-reservation endpoint

    Create new reservation

    Returns
    -------
    web.Response
        Http response object
    """

    # TODO: How to check that object?
    body = await request.json()
    # json_obj = json.loads(body)
    if type(body) is str:
        body = json.loads(body)
    try:
        reservation: Reservation = ReservationSchema().load(body)
        SessionProvider.session.add(reservation)
        SessionProvider.session.commit()
        return web.Response(text='OK')
    except IntegrityError:
        return web.Response(text='ERR: No restaurant has given id')
    except marshmallow.exceptions.ValidationError as e:
        return web.Response(text='ERR: Failed validation:\n' + str(e))
    except ValidationError as e:
        return web.Response(text='ERR: Failed validation:\n' + str(e))


def create_app() -> web.Application:
    """Creates application object with routing

    Returns
    -------
    web.Application
        Application object
    """
    SessionProvider.start_session()
    app = web.Application()
    app.add_routes([web.get('/version', version),
                    web.post('/new-reservation', new_reservation)])
    return app


if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    app = create_app()
    web.run_app(app, port=8000)

backend/backend/session_provider.py:

from backend.config import DATABASE_URI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


class SessionProvider:

    engine = None
    sess_maker = None
    session = None

    @classmethod
    def start_session(cls):
        # DATABASE_URI = f"postgres+psycopg2://postgres:example@localhost:5433/restro"
        cls._engine = create_engine(DATABASE_URI)
        cls.sess_maker = sessionmaker(bind=cls._engine)
        # create a Session
        cls.session = cls.sess_maker()

So my first guess was maybe it's related to paths, but when I noticed exactly same mock.patch('backend.app.SessionProvider') used with context manager works properly i started to doubt it. Do you have idea why one method isn't working while second does is? In the end, I wanna share my project tree so maybe this will help us find the answer:

project tree

1
  • 1
    Do you know that using synchronous SQLAlchemy in the same thread with aiohttp is an anti-pattern? It kills the concurrency, the final performance is terrible. Commented Dec 9, 2019 at 15:45

0

Browse other questions tagged or ask your own question.