116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
import requests
|
|
import pathlib2
|
|
import mock
|
|
|
|
|
|
def test_wcs_fixture(wcs, postgres_db, tmpdir, olap_cmd, caplog):
|
|
olap_cmd()
|
|
|
|
expected_schema = [
|
|
('agent', 'id'),
|
|
('agent', 'label'),
|
|
('category', 'id'),
|
|
('category', 'label'),
|
|
('channel', 'id'),
|
|
('channel', 'label'),
|
|
('evolution', 'id'),
|
|
('evolution', 'generic_status_id'),
|
|
('evolution', 'formdata_id'),
|
|
('evolution', 'time'),
|
|
('evolution', 'date'),
|
|
('evolution', 'hour_id'),
|
|
('evolution_demande', 'id'),
|
|
('evolution_demande', 'status_id'),
|
|
('evolution_demande', 'formdata_id'),
|
|
('evolution_demande', 'time'),
|
|
('evolution_demande', 'date'),
|
|
('evolution_demande', 'hour_id'),
|
|
('formdata', 'id'),
|
|
('formdata', 'formdef_id'),
|
|
('formdata', 'receipt_time'),
|
|
('formdata', 'hour_id'),
|
|
('formdata', 'channel_id'),
|
|
('formdata', 'backoffice'),
|
|
('formdata', 'generic_status_id'),
|
|
('formdata', 'endpoint_delay'),
|
|
('formdata', 'first_agent_id'),
|
|
('formdata', 'geolocation_base'),
|
|
('formdata', 'json_data'),
|
|
('formdata_demande', 'id'),
|
|
('formdata_demande', 'formdef_id'),
|
|
('formdata_demande', 'receipt_time'),
|
|
('formdata_demande', 'hour_id'),
|
|
('formdata_demande', 'channel_id'),
|
|
('formdata_demande', 'backoffice'),
|
|
('formdata_demande', 'generic_status_id'),
|
|
('formdata_demande', 'endpoint_delay'),
|
|
('formdata_demande', 'first_agent_id'),
|
|
('formdata_demande', 'geolocation_base'),
|
|
('formdata_demande', 'json_data'),
|
|
('formdata_demande', 'status_id'),
|
|
('formdata_demande', 'field_field_string'),
|
|
('formdata_demande', 'field_field_item'),
|
|
('formdata_demande', 'field_field_bool'),
|
|
('formdata_demande', 'function__receiver'),
|
|
('formdata_demande_field_field_item', 'id'),
|
|
('formdata_demande_field_field_item', 'label'),
|
|
('formdef', 'id'),
|
|
('formdef', 'category_id'),
|
|
('formdef', 'label'),
|
|
('hour', 'id'),
|
|
('hour', 'label'),
|
|
('role', 'id'),
|
|
('role', 'label'),
|
|
('status', 'id'),
|
|
('status', 'label'),
|
|
('status_demande', 'id'),
|
|
('status_demande', 'label')
|
|
]
|
|
|
|
# verify SQL schema
|
|
with postgres_db.conn() as conn:
|
|
with conn.cursor() as c:
|
|
c.execute('SELECT table_name, column_name '
|
|
'FROM information_schema.columns '
|
|
'WHERE table_schema = \'olap\' ORDER BY table_name, ordinal_position')
|
|
|
|
assert list(c.fetchall()) == expected_schema
|
|
|
|
# verify JSON schema
|
|
with (olap_cmd.model_dir / 'olap.model').open() as fd, \
|
|
(pathlib2.Path(__file__).parent / 'olap.model').open() as fd2:
|
|
json_schema = json.load(fd)
|
|
expected_json_schema = json.load(fd2)
|
|
expected_json_schema['pg_dsn'] = postgres_db.dsn
|
|
assert json_schema == expected_json_schema
|
|
|
|
|
|
def test_requests_exception(wcs, postgres_db, tmpdir, olap_cmd, caplog):
|
|
with mock.patch('requests.get', side_effect=requests.RequestException('wat!')):
|
|
with pytest.raises(SystemExit):
|
|
olap_cmd(no_log_errors=False)
|
|
assert 'wat!' in caplog.text
|
|
|
|
|
|
def test_requests_not_ok(wcs, postgres_db, tmpdir, olap_cmd, caplog):
|
|
with mock.patch('requests.get') as mocked_get:
|
|
mocked_get.return_value.ok = False
|
|
mocked_get.return_value.status_code = 401
|
|
mocked_get.return_value.text = '{"err": 1, "err_desc": "invalid signature"}'
|
|
with pytest.raises(SystemExit):
|
|
olap_cmd(no_log_errors=False)
|
|
assert 'invalid signature' in caplog.text
|
|
|
|
|
|
def test_requests_not_json(wcs, postgres_db, tmpdir, olap_cmd, caplog):
|
|
with mock.patch('requests.get') as mocked_get:
|
|
mocked_get.return_value.ok = True
|
|
mocked_get.return_value.json.side_effect = ValueError('invalid JSON')
|
|
with pytest.raises(SystemExit):
|
|
olap_cmd(no_log_errors=False)
|
|
assert 'Invalid JSON content' in caplog.text
|