53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
import importlib
|
|
import pickle
|
|
import sys
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from combo.utils import spooler
|
|
|
|
|
|
@pytest.fixture
|
|
def uwsgi(monkeypatch):
|
|
uwsgi = mock.Mock()
|
|
uwsgi.opt = ['spooler']
|
|
uwsgi.SPOOL_OK = 1
|
|
monkeypatch.setitem(sys.modules, 'uwsgi', uwsgi)
|
|
# reload with uwsgi impl
|
|
importlib.reload(spooler)
|
|
yield uwsgi
|
|
monkeypatch.delitem(sys.modules, 'uwsgi')
|
|
importlib.reload(spooler)
|
|
|
|
|
|
def test_tenantspool(uwsgi):
|
|
import uwsgidecorators
|
|
|
|
assert spooler.USE_UWSGI
|
|
|
|
mocked_f = mock.Mock()
|
|
|
|
@spooler.tenantspool
|
|
def function_to_spool(*args, **kwargs):
|
|
mocked_f(*args, **kwargs)
|
|
|
|
# check function is registered under its name, and not the name of the wrapper function
|
|
assert 'function_to_spool' in uwsgidecorators.spooler_functions
|
|
|
|
# check the spool request is properly serialized with tenant information
|
|
with mock.patch('combo.utils.spooler.connection') as connection:
|
|
connection.tenant.domain_url = 'example.com'
|
|
function_to_spool(1, b=3)
|
|
assert pickle.loads(uwsgi.spool.call_args[0][0][b'args']) == (1,)
|
|
assert pickle.loads(uwsgi.spool.call_args[0][0][b'kwargs']) == {'b': 3, 'domain': 'example.com'}
|
|
|
|
# check db is initialized, tenant restored and function called
|
|
with mock.patch('combo.utils.spooler.close_old_connections') as close_old_connections:
|
|
with mock.patch('combo.utils.spooler.tenant_context') as tenant_context:
|
|
uwsgidecorators.manage_spool_request(uwsgi.spool.call_args[0][0])
|
|
assert close_old_connections.call_count == 2
|
|
assert tenant_context.call_args[0][0] == 'example.com'
|
|
assert mocked_f.call_args[0] == (1,)
|
|
assert mocked_f.call_args[1] == {'b': 3}
|