|
1
|
|
|
import pytest |
|
2
|
|
|
|
|
3
|
|
|
|
|
4
|
|
|
@pytest.fixture |
|
5
|
|
|
def datapoints_manager_infra(): |
|
6
|
|
|
from so_magic.data.datapoints_manager import DatapointsManager, NonExistantDatapointsError |
|
7
|
|
|
return type('DatapointsManagerInfra', (object,), { |
|
8
|
|
|
'DatapointsManager': DatapointsManager(), |
|
9
|
|
|
'exceptions': type('E', (object,), { |
|
10
|
|
|
'store_with_invalid_key': Exception, |
|
11
|
|
|
'store_with_existing_key': Exception, |
|
12
|
|
|
'retrieve_non_existing_key': NonExistantDatapointsError, |
|
13
|
|
|
}), |
|
14
|
|
|
}) |
|
15
|
|
|
|
|
16
|
|
|
|
|
17
|
|
|
@pytest.fixture |
|
18
|
|
|
def test_subjects(): |
|
19
|
|
|
return type('S', (object,), {'empty_key': type('DynamicSubject', (object,), {'state': [1], 'name': ''}), |
|
20
|
|
|
'k1': type('DynamicSubject', (object,), {'state': [1], 'name': 'k1'})}) |
|
21
|
|
|
|
|
22
|
|
|
|
|
23
|
|
|
def test_registering_datapoints_with_invalid_key(datapoints_manager_infra, test_subjects): |
|
24
|
|
|
with pytest.raises(datapoints_manager_infra.exceptions.store_with_invalid_key, |
|
25
|
|
|
match=rf'Subject {test_subjects.empty_key} with state \[1\] resulted in an empty string as key. ' |
|
26
|
|
|
r'We reject the key, since it is going to "query" a dict/hash\.'): |
|
27
|
|
|
datapoints_manager_infra.DatapointsManager.update(test_subjects.empty_key) |
|
28
|
|
|
|
|
29
|
|
|
def test_registering_datapoints_with_existing_key(datapoints_manager_infra, test_subjects): |
|
30
|
|
|
datapoints_manager_infra.DatapointsManager.update(test_subjects.k1) |
|
31
|
|
|
with pytest.raises(datapoints_manager_infra.exceptions.store_with_existing_key): |
|
32
|
|
|
datapoints_manager_infra.DatapointsManager.update(test_subjects.k1) |
|
33
|
|
|
|
|
34
|
|
|
|
|
35
|
|
|
def test_retrieving_non_registered_datapoints(caplog, datapoints_manager_infra): |
|
36
|
|
|
with pytest.raises(datapoints_manager_infra.exceptions.retrieve_non_existing_key, |
|
37
|
|
|
match=r'Requested non existant Datapoints instance. Probable cause is that this \(self\) ' |
|
38
|
|
|
'DatapointsManager instance has not been notified by a DatapointsFactory'): |
|
39
|
|
|
_ = datapoints_manager_infra.DatapointsManager.datapoints |
|
40
|
|
|
assert 'Non existant Datapoints: {\n' \ |
|
41
|
|
|
' "last-key-used-to-register-datapoints": "",\n' \ |
|
42
|
|
|
' "datapoints-registry-keys": "[]"\n}' in caplog.text |
|
43
|
|
|
|
boromir674 /
so-magic
