1
|
|
|
import pytest |
2
|
|
|
|
3
|
|
|
|
4
|
|
|
@pytest.mark.parametrize('train_args', [ |
5
|
|
|
([6, 8, 'toroid', 'hexagonal']), |
6
|
|
|
# ([12, 12, 'toroid', 'rectangular']) |
7
|
|
|
]) |
8
|
|
|
def test_somagic_scenario(train_args, somagic, test_dataset, sample_collaped_json): |
9
|
|
|
attrs = ('width', 'height', 'type', 'grid_type') |
10
|
|
|
som = somagic.map.train(*train_args[:2], maptype=train_args[2], gridtype=train_args[3]) |
11
|
|
|
assert som.dataset_name == sample_collaped_json |
12
|
|
|
assert all(parameter == getattr(som, attribute) for attribute, parameter in zip(attrs, train_args)) |
13
|
|
|
|
14
|
|
|
|
15
|
|
|
@pytest.fixture |
16
|
|
|
def objects_to_test(): |
17
|
|
|
def get_objects_to_test(so_master): |
18
|
|
|
return { |
19
|
|
|
'dtps_fct': so_master._data_manager.engine.backend.datapoints_factory, |
20
|
|
|
'cmd_fct': so_master._data_manager.engine.backend.command_factory, |
21
|
|
|
'phi_class': so_master._data_manager.phi_class, |
22
|
|
|
'rest': ( |
23
|
|
|
so_master, |
24
|
|
|
so_master._data_manager, |
25
|
|
|
so_master._data_manager.engine, |
26
|
|
|
so_master._data_manager.engine.backend, |
27
|
|
|
so_master._data_manager.engine.datapoints_manager, |
28
|
|
|
so_master._data_manager.engine.backend.datapoints_factory.subject, |
29
|
|
|
so_master._data_manager.engine.backend.datapoints_factory.subject._observers, |
30
|
|
|
so_master._data_manager.phi_class.subject, |
31
|
|
|
so_master._data_manager.phi_class.subject._observers, |
32
|
|
|
) |
33
|
|
|
} |
34
|
|
|
return get_objects_to_test |
35
|
|
|
|
36
|
|
|
|
37
|
|
|
@pytest.fixture(params=[[2]]) |
38
|
|
|
def so_magic_instances(request, objects_to_test): |
39
|
|
|
from so_magic import init_so_magic |
40
|
|
|
return [(i, objects_to_test(i)) for i in iter([init_so_magic() for _ in range(request.param[0])])] |
41
|
|
|
|
42
|
|
|
|
43
|
|
|
def test_somagic_objects(so_magic_instances, assert_different_objects): |
44
|
|
|
from functools import reduce |
45
|
|
|
assert_different_objects(reduce(lambda i, j: i + j, |
46
|
|
|
([x[1]['dtps_fct'], x[1]['cmd_fct'], x[1]['phi_class']] + list(x[1]['rest']) |
|
|
|
|
47
|
|
|
for x in so_magic_instances))) |
48
|
|
|
assert so_magic_instances[0][1]['phi_class'] != so_magic_instances[1][1]['phi_class'] |
49
|
|
|
|
50
|
|
|
|
51
|
|
|
def test_subscriptions(so_magic_instances): |
52
|
|
|
s = so_magic_instances[0] |
53
|
|
|
nb_observers = (1, 1, 1) |
54
|
|
|
|
55
|
|
|
assert s[1]['dtps_fct'].subject._observers[0] == s[0]._data_manager.engine.datapoints_manager |
56
|
|
|
assert s[1]['cmd_fct'].subject._observers[0] == s[0]._data_manager.commands_manager.command.accumulator |
57
|
|
|
assert s[1]['phi_class'].subject._observers[0] == s[0]._data_manager.built_phis |
58
|
|
|
assert all([len(subject._observers) == obs for subject, obs in zip( |
59
|
|
|
(s[1]['dtps_fct'].subject, s[1]['cmd_fct'].subject, s[1]['phi_class'].subject), nb_observers)]) |
60
|
|
|
|