test_data_manager.test_data_manager() - Code Metrics - Inspection of "wip" - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — dev ( 5faf7b...5a9f48 )

by Konstantinos

created 2020-08-16 23:53 UTC

test_data_manager.test_data_manager() B

↳ Parent: test_data_manager

Complexity

Conditions

Size

Total Lines	75
Code Lines	52

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	52
dl	0
loc	75
rs	8.5709
c	0
b	0
f	0
cc	1
nop	1

How to fix Long Method

import pytest
from green_magic.data.commands_manager import CommandsManager
from green_magic.data.backend import Backend, DataEngine
from green_magic.data.data_manager import DataManager
from green_magic.data.backend import panda_handling


@pytest.fixture()
def test_json_data(sample_json):
    return {
        'file_path': sample_json,
        'nb_lines': 100,
        'attributes': {'flavors', 'name', 'medical', 'description', 'image_urls', 'parents', 'negatives', 'grow_info', '_id', 'type', 'image_paths', 'effects'},
    }

def test_engine_registration():
    from green_magic.data.backend import DataEngine
    from green_magic.data.backend.engine import EngineType
    assert type(DataEngine) == EngineType
    from green_magic.data.commands_manager import CommandRegistrator

    # type(type(DataEngine))
    ATTRS = ('registry', '_commands', 'command')
    DataEngine.new('test_pd')
    assert 'test_pd' in DataEngine.subclasses
    assert type(DataEngine.test_pd) == type(DataEngine)
    assert all(hasattr(DataEngine.test_pd, x) for x in ATTRS)

    DataEngine.new('data_lib')
    assert 'data_lib' in DataEngine.subclasses
    assert type(DataEngine.data_lib) == type(DataEngine)
    assert all(hasattr(DataEngine.data_lib, x) for x in ATTRS)

    for x in ATTRS:
        print(getattr(DataEngine.test_pd, x), getattr(DataEngine.data_lib, x))
        assert id(getattr(DataEngine.test_pd, x)) != id(getattr(DataEngine.data_lib, x)) != id(DataEngine.registry)

    assert id(DataEngine.test_pd.datapoints_factory) == id(DataEngine.data_lib.datapoints_factory)

    import pandas as pd
    @DataEngine.test_pd.command
    def observations(file_path):
        return pd.read_json(file_path, lines=True)

    # TODO uncomment below
    # assert len(DataEngine.test_pd.registry) == 1
    # assert len(DataEngine.data_lib.registry) == 0

    # assert all(id(getattr(DataEngine.test_pd, x) != id(getattr(DataEngine.data_lib, x)) for x in ATTRS))


def test_data_manager(test_json_data):
    import types
    from green_magic.utils.commands import Command
    from green_magic.data.backend.panda_handling.df_backend import PDTabularIterator, PDTabularRetriever, PDTabularReporter
    from green_magic.data.features.phi import PhiFunction

    DataEngine.new('test_pd')
    assert 'test_pd' in DataEngine.subclasses
    assert hasattr(DataEngine, 'state')
    assert hasattr(DataEngine, 'registry')

    data_api = DataManager(CommandsManager(), Backend(DataEngine.create('test_pd')))
    # make the datapoint_manager listen to newly created Datapoints objects events
    assert hasattr(data_api, 'commands_manager')
    assert hasattr(data_api.commands_manager, 'command')
    assert hasattr(data_api.backend, 'datapoints_manager')

    data_api.backend.engine.__class__.datapoints_factory.subject.attach(data_api.backend.datapoints_manager)
    DataEngine.test_pd.command_factory.attach(data_api.commands_manager.command.accumulator)
    PhiFunction.subject.attach(data_api.phis)

    assert data_api.backend.engine.__class__.datapoints_factory not in PhiFunction.subject._observers

    # test runtime command registration
    import pandas as pd
    @DataEngine.test_pd.dec()
    def observations(file_path):
        return pd.read_json(file_path, lines=True)

    @DataEngine.test_pd.dec()
    def add_attribute(_datapoints, values, new_attribute):
        print("CORRECT")
        _datapoints.observations[new_attribute] = values

    assert type(DataEngine.test_pd) == type(DataEngine)
    assert type(DataEngine.test_pd.registry) == dict
    assert type(DataEngine.test_pd._commands) == dict
    assert 'observations' in DataEngine.test_pd.registry
    assert 'observations' in DataEngine.test_pd._commands
    assert 'observations' not in DataEngine.registry

    assert type(DataEngine.test_pd._commands['observations']) == Command

    cmd = DataEngine.test_pd._commands['observations']
    # cmd = data_api.command.observations
    cmd.args = [test_json_data['file_path']]

    assert type(cmd._receiver) == types.FunctionType
    assert cmd._method == '__call__'
    assert cmd.args == [test_json_data['file_path']]

    from green_magic.utils.commands import Invoker, CommandHistory

    DataEngine.test_pd.retriever = PDTabularRetriever
    DataEngine.test_pd.iterator = PDTabularIterator
    DataEngine.test_pd.reporter = PDTabularReporter

    inv = Invoker(CommandHistory())
    inv.execute_command(cmd)

    datapoints = data_api.backend.datapoints_manager.datapoints
    assert len(datapoints) == test_json_data['nb_lines']
    print(datapoints.attributes)

    assert set(datapoints.attributes) == test_json_data['attributes']

    assert 'add_attribute' in DataEngine.test_pd.registry

    cmd1 = DataEngine.test_pd._commands['add_attribute']
    # cmd1 = data_api.command.observations
    cmd1.args = [datapoints, [_ for _ in range(1, len(datapoints) + 1)], 'test_attr']

    cmd1.execute()

    assert set(datapoints.attributes) == set(_ for _ in list(test_json_data['attributes']) + ['test_attr'])


1			import pytest
2			from green_magic.data.commands_manager import CommandsManager
3			from green_magic.data.backend import Backend, DataEngine
4			from green_magic.data.data_manager import DataManager
5			from green_magic.data.backend import panda_handling
6
7
8			@pytest.fixture()
9			def test_json_data(sample_json):
10			return {
11			'file_path': sample_json,
12			'nb_lines': 100,
13			'attributes': {'flavors', 'name', 'medical', 'description', 'image_urls', 'parents', 'negatives', 'grow_info', '_id', 'type', 'image_paths', 'effects'},
14			}
15
16			def test_engine_registration():
17			from green_magic.data.backend import DataEngine
18			from green_magic.data.backend.engine import EngineType
19			assert type(DataEngine) == EngineType
20			from green_magic.data.commands_manager import CommandRegistrator
21
22			# type(type(DataEngine))
23			ATTRS = ('registry', '_commands', 'command')
24			DataEngine.new('test_pd')
25			assert 'test_pd' in DataEngine.subclasses
26			assert type(DataEngine.test_pd) == type(DataEngine)
27			assert all(hasattr(DataEngine.test_pd, x) for x in ATTRS)
28
29			DataEngine.new('data_lib')
30			assert 'data_lib' in DataEngine.subclasses
31			assert type(DataEngine.data_lib) == type(DataEngine)
32			assert all(hasattr(DataEngine.data_lib, x) for x in ATTRS)
33
34			for x in ATTRS:
35			print(getattr(DataEngine.test_pd, x), getattr(DataEngine.data_lib, x))
36			assert id(getattr(DataEngine.test_pd, x)) != id(getattr(DataEngine.data_lib, x)) != id(DataEngine.registry)
37
38			assert id(DataEngine.test_pd.datapoints_factory) == id(DataEngine.data_lib.datapoints_factory)
39
40			import pandas as pd
41			@DataEngine.test_pd.command
42			def observations(file_path):
43			return pd.read_json(file_path, lines=True)
44
45			# TODO uncomment below
46			# assert len(DataEngine.test_pd.registry) == 1
47			# assert len(DataEngine.data_lib.registry) == 0
48
49			# assert all(id(getattr(DataEngine.test_pd, x) != id(getattr(DataEngine.data_lib, x)) for x in ATTRS))
50
51
52			def test_data_manager(test_json_data):
53			import types
54			from green_magic.utils.commands import Command
55			from green_magic.data.backend.panda_handling.df_backend import PDTabularIterator, PDTabularRetriever, PDTabularReporter
56			from green_magic.data.features.phi import PhiFunction
57
58			DataEngine.new('test_pd')
59			assert 'test_pd' in DataEngine.subclasses
60			assert hasattr(DataEngine, 'state')
61			assert hasattr(DataEngine, 'registry')
62
63			data_api = DataManager(CommandsManager(), Backend(DataEngine.create('test_pd')))
64			# make the datapoint_manager listen to newly created Datapoints objects events
65			assert hasattr(data_api, 'commands_manager')
66			assert hasattr(data_api.commands_manager, 'command')
67			assert hasattr(data_api.backend, 'datapoints_manager')
68
69			data_api.backend.engine.__class__.datapoints_factory.subject.attach(data_api.backend.datapoints_manager)
70			DataEngine.test_pd.command_factory.attach(data_api.commands_manager.command.accumulator)
71			PhiFunction.subject.attach(data_api.phis)
72
73			assert data_api.backend.engine.__class__.datapoints_factory not in PhiFunction.subject._observers
74
75			# test runtime command registration
76			import pandas as pd
77			@DataEngine.test_pd.dec()
78			def observations(file_path):
79			return pd.read_json(file_path, lines=True)
80
81			@DataEngine.test_pd.dec()
82			def add_attribute(_datapoints, values, new_attribute):
83			print("CORRECT")
84			_datapoints.observations[new_attribute] = values
85
86			assert type(DataEngine.test_pd) == type(DataEngine)
87			assert type(DataEngine.test_pd.registry) == dict
88			assert type(DataEngine.test_pd._commands) == dict
89			assert 'observations' in DataEngine.test_pd.registry
90			assert 'observations' in DataEngine.test_pd._commands
91			assert 'observations' not in DataEngine.registry
92
93			assert type(DataEngine.test_pd._commands['observations']) == Command
94
95			cmd = DataEngine.test_pd._commands['observations']
96			# cmd = data_api.command.observations
97			cmd.args = [test_json_data['file_path']]
98
99			assert type(cmd._receiver) == types.FunctionType
100			assert cmd._method == '__call__'
101			assert cmd.args == [test_json_data['file_path']]
102
103			from green_magic.utils.commands import Invoker, CommandHistory
104
105			DataEngine.test_pd.retriever = PDTabularRetriever
106			DataEngine.test_pd.iterator = PDTabularIterator
107			DataEngine.test_pd.reporter = PDTabularReporter
108
109			inv = Invoker(CommandHistory())
110			inv.execute_command(cmd)
111
112			datapoints = data_api.backend.datapoints_manager.datapoints
113			assert len(datapoints) == test_json_data['nb_lines']
114			print(datapoints.attributes)
115
116			assert set(datapoints.attributes) == test_json_data['attributes']
117
118			assert 'add_attribute' in DataEngine.test_pd.registry
119
120			cmd1 = DataEngine.test_pd._commands['add_attribute']
121			# cmd1 = data_api.command.observations
122			cmd1.args = [datapoints, [_ for _ in range(1, len(datapoints) + 1)], 'test_attr']
123
124			cmd1.execute()
125
126			assert set(datapoints.attributes) == set(_ for _ in list(test_json_data['attributes']) + ['test_attr'])
127

boromir674 / so-magic

Push — dev ( 5faf7b...5a9f48 )

test_data_manager.test_data_manager() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like