44 lines
1.3 KiB
Python
44 lines
1.3 KiB
Python
import logging
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from wcs_olap.feeder import WcsOlapFeeder
|
|
|
|
|
|
def test_constructor():
|
|
feeder = WcsOlapFeeder(api=None, pg_dsn='', schema='x' * 63)
|
|
schema_temp = feeder.ctx.as_dict()['schema_temp']
|
|
assert len(schema_temp) < 64
|
|
assert schema_temp == 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxedd110xxxxxxxxxxxxxxxxxxxxxxxx_temp'
|
|
|
|
with pytest.raises(ValueError):
|
|
feeder = WcsOlapFeeder(api=None, pg_dsn='', schema='x' * 64)
|
|
|
|
|
|
def test_sql_error_logging(caplog):
|
|
feeder = WcsOlapFeeder(api=None, pg_dsn='', schema='x' * 63, logger=logging.getLogger())
|
|
with pytest.raises(Exception):
|
|
feeder.ex('COIN')
|
|
assert 'Failed to execute' in caplog.text
|
|
|
|
|
|
def test_post_sync_commands(mock_cursor_execute, wcs, postgres_db, olap_cmd):
|
|
with olap_cmd.config() as config:
|
|
config.set(
|
|
'wcs-olap',
|
|
'post-sync-commands',
|
|
"NOTIFY coucou, '{schema}';\nSELECT * FROM information_schema.tables")
|
|
|
|
queries = []
|
|
|
|
def side_effect(query, *args, **kwargs):
|
|
queries.append(query)
|
|
return mock.DEFAULT
|
|
|
|
with mock_cursor_execute(side_effect=side_effect):
|
|
assert olap_cmd() == 0
|
|
|
|
# verify post-sync-commands are in the last executed queries
|
|
assert queries[-1] == "NOTIFY coucou, 'olap';\nSELECT * FROM information_schema.tables"
|