tests.conftest.Watcher.__init__() - Code Metrics - isomeric/isomer - Measure and Improve Code Quality continuously with Scrutinizer

tests.conftest.Watcher.init() A
last analyzed 2021-02-24 12:38 UTC

↳ Parent: tests.conftest

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	3
dl	0
loc	4
rs	10
c	0
b	0
f	0

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

# Isomer - The distributed application framework
# ==============================================
# Copyright (C) 2011-2020 Heiko 'riot' Weinen <[email protected]> and others.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os
import sys
import shutil
import threading
import collections
from time import sleep, strftime
from collections import deque

import pytest
import pymongo
from click.testing import CliRunner
from circuits.core.manager import TIMEOUT
from circuits import handler, BaseComponent, Debugger, Manager
from formal import model_factory
from isomer.database import initialize
from isomer.component import ConfigurableComponent
from isomer.misc.path import set_etc_path, set_instance
from isomer.schemata.component import ComponentConfigSchemaTemplate
from isomer.tool.etc import create_configuration

"""Basic Test suite bits and pieces"""

DEFAULT_DATABASE_NAME = "isomer-test-internal"
DEFAULT_DATABASE_HOST = "localhost"
DEFAULT_DATABASE_PORT = 27017
DEFAULT_INSTANCE_NAME = "isomer-test-internal"
COLORS = False


class TestComponent(ConfigurableComponent):
    """Very basic testing component"""

    configprops = {
        'test': {'type': 'string'}
    }


class Watcher(BaseComponent):
    """Watches for incoming events"""

    def __init__(self, *args, **kwargs):
        super(Watcher).__init__(*args, **kwargs)
        self.events = deque()
        self._lock = threading.Lock()

    @handler(channel="*", priority=999.9)
    def _on_event(self, event, *args, **kwargs):
        with self._lock:
            self.events.append(event)

    def clear(self):
        """Reset caught events"""

        self.events.clear()

    def wait(self, name, channel=None, timeout=6.0):
        """Linger and wait for specified incoming events"""

        for i in range(int(timeout / TIMEOUT)):
            with self._lock:
                for event in self.events:
                    if event.name == name and event.waitingHandlers == 0:
                        if (channel is None) or (channel in event.channels):
                            return True
            sleep(TIMEOUT)
        else:
            return False


class Flag(object):
    """Flag object for Watcher component"""
    status = False
    event = None


def call_event_from_name(manager, event, event_name, *channels):
    """Fire a named event and wait for a specified response"""

    fired = False
    value = None
    for r in manager.waitEvent(event_name):
        if not fired:
            fired = True
            value = manager.fire(event, *channels)
        sleep(0.1)
    return value


def call_event(manager, event, *channels):
    """Simply fire and forget a specified event"""

    return call_event_from_name(manager, event, event.name, *channels)


class WaitEvent(object):
    """Simple component substitute that waits for a specified Event"""

    def __init__(self, manager, name, channel=None, timeout=2):
        if channel is None:
            channel = getattr(manager, "channel", None)

        self.timeout = timeout
        self.manager = manager

        flag = Flag()

        @handler(name, channel=channel)
        def on_event(self, event):
            """An event has been received"""

            flag.status = True
            flag.event = event

        self.handler = self.manager.addHandler(on_event)
        self.flag = flag

    def wait(self):
        """Wait for the (upon instantiation) specified timeout for an event"""
        try:
            for i in range(int(self.timeout / TIMEOUT)):
                if self.flag.status:
                    return self.flag.event
                sleep(TIMEOUT)
        finally:
            self.manager.removeHandler(self.handler)


def wait_for(obj, attr, value=True, timeout=3.0):
    """Wait until timeout or an object acquires a specified attribute"""

    from circuits.core.manager import TIMEOUT
    for i in range(int(timeout / TIMEOUT)):
        if isinstance(value, collections.Callable):
            if value(obj, attr):
                return True
        elif getattr(obj, attr) == value:
            return True
        sleep(TIMEOUT)


@pytest.fixture
def manager(request):
    """Component testing manager/fixture"""

    manager = Manager()

    def finalizer():
        """Stop the testing"""

        manager.stop()

    request.addfinalizer(finalizer)

    waiter = WaitEvent(manager, "started")
    manager.start()
    assert waiter.wait()

    if request.config.option.verbose:
        verbose = True
    else:
        verbose = False

    Debugger(events=verbose).register(manager)

    return manager


@pytest.fixture
def watcher(request, manager):
    """Fixture that cleans up after unregistering"""

    watcher = Watcher().register(manager)

    def finalizer():
        """Setup the manager and wait for completion, then unregister"""

        waiter = WaitEvent(manager, "unregistered")
        watcher.unregister()
        waiter.wait()

    request.addfinalizer(finalizer)

    return watcher


def run_cli(cmd, args, full_log=False):
    """Runs a command"""

    if COLORS is False:
        args.insert(0, '-nc')

    if full_log:
        timestamp = strftime("%Y%m%d-%H%M%S")

        log_args = [
            '--clog', '5', '--flog', '5',
            '--log-file', '/tmp/isomer-test_%s' % timestamp
        ]
        args = log_args + args

    args = ['--config-path', '/tmp/isomer-test/etc/isomer'] + args
    args = ['--prefix-path', '/tmp/isomer-test'] + args
    args = ['--instance', pytest.INSTANCENAME] + args

    # pprint(args)

    runner = CliRunner()
    result = runner.invoke(cmd, args, catch_exceptions=True, obj={})
    with open('/tmp/isomer_test_run_cli_logfile', 'a') as f:
        f.write(result.output)
    return result


def reset_base(unset_instance=False):
    """Prepares a testing folder and sets Isomer's base to that"""

    class ctx_mock():
        obj = {'config': None}

    if os.path.exists('/tmp/isomer-test'):
        shutil.rmtree('/tmp/isomer-test')

    os.makedirs('/tmp/isomer-test/etc/isomer/instances')
    os.makedirs('/tmp/isomer-test/var/log/isomer')

    set_etc_path('/tmp/isomer-test/etc/isomer')
    create_configuration(ctx_mock())
    if unset_instance is False:
        set_instance('foobar', 'green', '/tmp/isomer-test/')


def clean_test_components():
    """Removes test-generated component data"""

    print("Removing test components...")
    for item in model_factory(ComponentConfigSchemaTemplate).find({
        'componentclass': 'TestComponent'
    }):
        item.delete()


def clean_test_database(config):
    """Removes all of the test-generated database content"""

    db_name = config.getoption("--dbname", default=DEFAULT_DATABASE_NAME)
    host = config.getoption("--dbhost", default=DEFAULT_DATABASE_HOST)
    port = config.getoption("--dbport", default=DEFAULT_DATABASE_PORT)

    client = pymongo.MongoClient(host=host, port=int(port))
    if db_name in client.list_database_names():
        print("Dropping test database", db_name)
        client.drop_database(db_name)
    else:
        print("Test database does not exist")


@pytest.hookimpl()
def pytest_unconfigure(config):
    """Clear test generated data after test completion"""

    clean_test_database(config)


def pytest_addoption(parser):
    parser.addoption(
        "--dbname", action="store", default=DEFAULT_DATABASE_NAME,
        help="test db name"
    )
    parser.addoption(
        "--dbhost", action="store", default=DEFAULT_DATABASE_HOST,
        help="test db hostname"
    )
    parser.addoption(
        "--dbport", action="store", default=DEFAULT_DATABASE_PORT,
        help="test db port", type="int"
    )


def pytest_configure(config):
    """Setup the testing namespace"""

    dbname = config.getoption("--dbname", default=DEFAULT_DATABASE_NAME)
    dbhost = config.getoption("--dbhost", default=DEFAULT_DATABASE_HOST)
    dbport = config.getoption("--dbport", default=DEFAULT_DATABASE_PORT)
    instance_name = config.getoption("--instance", default=DEFAULT_INSTANCE_NAME)

    pytest.TestComponent = TestComponent
    pytest.clean_test_components = clean_test_components
    pytest.WaitEvent = WaitEvent
    pytest.wait_for = wait_for
    pytest.call_event = call_event
    pytest.PLATFORM = sys.platform
    pytest.PYVER = sys.version_info[:3]
    pytest.DBNAME = dbname
    pytest.DBHOST = dbhost
    pytest.DBPORT = dbport
    pytest.INSTANCENAME = instance_name
    pytest.call_event_from_name = call_event_from_name
    pytest.run_cli = run_cli
    pytest.reset_base = reset_base

    clean_test_database(config)

    initialize(
        address="%s:%i" % (dbhost, dbport),
        database_name=dbname,
        instance_name=instance_name
    )


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			# Isomer - The distributed application framework
5			# ==============================================
6			# Copyright (C) 2011-2020 Heiko 'riot' Weinen <[email protected]> and others.
7			#
8			# This program is free software: you can redistribute it and/or modify
9			# it under the terms of the GNU Affero General Public License as published by
10			# the Free Software Foundation, either version 3 of the License, or
11			# (at your option) any later version.
12			#
13			# This program is distributed in the hope that it will be useful,
14			# but WITHOUT ANY WARRANTY; without even the implied warranty of
15			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16			# GNU Affero General Public License for more details.
17			#
18			# You should have received a copy of the GNU Affero General Public License
19			# along with this program. If not, see <http://www.gnu.org/licenses/>.
20
21			import os
22			import sys
23			import shutil
24			import threading
25			import collections
26			from time import sleep, strftime
27			from collections import deque
28
29			import pytest
30			import pymongo
31			from click.testing import CliRunner
32			from circuits.core.manager import TIMEOUT
33			from circuits import handler, BaseComponent, Debugger, Manager
34			from formal import model_factory
35			from isomer.database import initialize
36			from isomer.component import ConfigurableComponent
37			from isomer.misc.path import set_etc_path, set_instance
38			from isomer.schemata.component import ComponentConfigSchemaTemplate
39			from isomer.tool.etc import create_configuration
40
41			"""Basic Test suite bits and pieces"""
42
43			DEFAULT_DATABASE_NAME = "isomer-test-internal"
44			DEFAULT_DATABASE_HOST = "localhost"
45			DEFAULT_DATABASE_PORT = 27017
46			DEFAULT_INSTANCE_NAME = "isomer-test-internal"
47			COLORS = False
48
49
50			class TestComponent(ConfigurableComponent):
51			"""Very basic testing component"""
52
53			configprops = {
54			'test': {'type': 'string'}
55			}
56
57
58			class Watcher(BaseComponent):
59			"""Watches for incoming events"""
60
61			def __init__(self, args, *kwargs):
62			super(Watcher).__init__(args, *kwargs)
63			self.events = deque()
64			self._lock = threading.Lock()
65
66			@handler(channel="*", priority=999.9)
67			def _on_event(self, event, args, *kwargs):
68			with self._lock:
69			self.events.append(event)
70
71			def clear(self):
72			"""Reset caught events"""
73
74			self.events.clear()
75
76			def wait(self, name, channel=None, timeout=6.0):
77			"""Linger and wait for specified incoming events"""
78
79			for i in range(int(timeout / TIMEOUT)):
80			with self._lock:
81			for event in self.events:
82			if event.name == name and event.waitingHandlers == 0:
83			if (channel is None) or (channel in event.channels):
84			return True
85			sleep(TIMEOUT)
86			else:
87			return False
88
89
90			class Flag(object):
91			"""Flag object for Watcher component"""
92			status = False
93			event = None
94
95
96			def call_event_from_name(manager, event, event_name, *channels):
97			"""Fire a named event and wait for a specified response"""
98
99			fired = False
100			value = None
101			for r in manager.waitEvent(event_name):
102			if not fired:
103			fired = True
104			value = manager.fire(event, *channels)
105			sleep(0.1)
106			return value
107
108
109			def call_event(manager, event, *channels):
110			"""Simply fire and forget a specified event"""
111
112			return call_event_from_name(manager, event, event.name, *channels)
113
114
115			class WaitEvent(object):
116			"""Simple component substitute that waits for a specified Event"""
117
118			def __init__(self, manager, name, channel=None, timeout=2):
119			if channel is None:
120			channel = getattr(manager, "channel", None)
121
122			self.timeout = timeout
123			self.manager = manager
124
125			flag = Flag()
126
127			@handler(name, channel=channel)
128			def on_event(self, event):
129			"""An event has been received"""
130
131			flag.status = True
132			flag.event = event
133
134			self.handler = self.manager.addHandler(on_event)
135			self.flag = flag
136
137			def wait(self):
138			"""Wait for the (upon instantiation) specified timeout for an event"""
139			try:
140			for i in range(int(self.timeout / TIMEOUT)):
141			if self.flag.status:
142			return self.flag.event
143			sleep(TIMEOUT)
144			finally:
145			self.manager.removeHandler(self.handler)
146
147
148			def wait_for(obj, attr, value=True, timeout=3.0):
149			"""Wait until timeout or an object acquires a specified attribute"""
150
151			from circuits.core.manager import TIMEOUT
152			for i in range(int(timeout / TIMEOUT)):
153			if isinstance(value, collections.Callable):
154			if value(obj, attr):
155			return True
156			elif getattr(obj, attr) == value:
157			return True
158			sleep(TIMEOUT)
159
160
161			@pytest.fixture
162			def manager(request):
163			"""Component testing manager/fixture"""
164
165			manager = Manager()
166
167			def finalizer():
168			"""Stop the testing"""
169
170			manager.stop()
171
172			request.addfinalizer(finalizer)
173
174			waiter = WaitEvent(manager, "started")
175			manager.start()
176			assert waiter.wait()
177
178			if request.config.option.verbose:
179			verbose = True
180			else:
181			verbose = False
182
183			Debugger(events=verbose).register(manager)
184
185			return manager
186
187
188			@pytest.fixture
189			def watcher(request, manager):
190			"""Fixture that cleans up after unregistering"""
191
192			watcher = Watcher().register(manager)
193
194			def finalizer():
195			"""Setup the manager and wait for completion, then unregister"""
196
197			waiter = WaitEvent(manager, "unregistered")
198			watcher.unregister()
199			waiter.wait()
200
201			request.addfinalizer(finalizer)
202
203			return watcher
204
205
206			def run_cli(cmd, args, full_log=False):
207			"""Runs a command"""
208
209			if COLORS is False:
210			args.insert(0, '-nc')
211
212			if full_log:
213			timestamp = strftime("%Y%m%d-%H%M%S")
214
215			log_args = [
216			'--clog', '5', '--flog', '5',
217			'--log-file', '/tmp/isomer-test_%s' % timestamp
218			]
219			args = log_args + args
220
221			args = ['--config-path', '/tmp/isomer-test/etc/isomer'] + args
222			args = ['--prefix-path', '/tmp/isomer-test'] + args
223			args = ['--instance', pytest.INSTANCENAME] + args
224
225			# pprint(args)
226
227			runner = CliRunner()
228			result = runner.invoke(cmd, args, catch_exceptions=True, obj={})
229			with open('/tmp/isomer_test_run_cli_logfile', 'a') as f:
230			f.write(result.output)
231			return result
232
233
234			def reset_base(unset_instance=False):
235			"""Prepares a testing folder and sets Isomer's base to that"""
236
237			class ctx_mock():
238			obj = {'config': None}
239
240			if os.path.exists('/tmp/isomer-test'):
241			shutil.rmtree('/tmp/isomer-test')
242
243			os.makedirs('/tmp/isomer-test/etc/isomer/instances')
244			os.makedirs('/tmp/isomer-test/var/log/isomer')
245
246			set_etc_path('/tmp/isomer-test/etc/isomer')
247			create_configuration(ctx_mock())
248			if unset_instance is False:
249			set_instance('foobar', 'green', '/tmp/isomer-test/')
250
251
252			def clean_test_components():
253			"""Removes test-generated component data"""
254
255			print("Removing test components...")
256			for item in model_factory(ComponentConfigSchemaTemplate).find({
257			'componentclass': 'TestComponent'
258			}):
259			item.delete()
260
261
262			def clean_test_database(config):
263			"""Removes all of the test-generated database content"""
264
265			db_name = config.getoption("--dbname", default=DEFAULT_DATABASE_NAME)
266			host = config.getoption("--dbhost", default=DEFAULT_DATABASE_HOST)
267			port = config.getoption("--dbport", default=DEFAULT_DATABASE_PORT)
268
269			client = pymongo.MongoClient(host=host, port=int(port))
270			if db_name in client.list_database_names():
271			print("Dropping test database", db_name)
272			client.drop_database(db_name)
273			else:
274			print("Test database does not exist")
275
276
277			@pytest.hookimpl()
278			def pytest_unconfigure(config):
279			"""Clear test generated data after test completion"""
280
281			clean_test_database(config)
282
283
284			def pytest_addoption(parser):
285			parser.addoption(
286			"--dbname", action="store", default=DEFAULT_DATABASE_NAME,
287			help="test db name"
288			)
289			parser.addoption(
290			"--dbhost", action="store", default=DEFAULT_DATABASE_HOST,
291			help="test db hostname"
292			)
293			parser.addoption(
294			"--dbport", action="store", default=DEFAULT_DATABASE_PORT,
295			help="test db port", type="int"
296			)
297
298
299			def pytest_configure(config):
300			"""Setup the testing namespace"""
301
302			dbname = config.getoption("--dbname", default=DEFAULT_DATABASE_NAME)
303			dbhost = config.getoption("--dbhost", default=DEFAULT_DATABASE_HOST)
304			dbport = config.getoption("--dbport", default=DEFAULT_DATABASE_PORT)
305			instance_name = config.getoption("--instance", default=DEFAULT_INSTANCE_NAME)
306
307			pytest.TestComponent = TestComponent
308			pytest.clean_test_components = clean_test_components
309			pytest.WaitEvent = WaitEvent
310			pytest.wait_for = wait_for
311			pytest.call_event = call_event
312			pytest.PLATFORM = sys.platform
313			pytest.PYVER = sys.version_info[:3]
314			pytest.DBNAME = dbname
315			pytest.DBHOST = dbhost
316			pytest.DBPORT = dbport
317			pytest.INSTANCENAME = instance_name
318			pytest.call_event_from_name = call_event_from_name
319			pytest.run_cli = run_cli
320			pytest.reset_base = reset_base
321
322			clean_test_database(config)
323
324			initialize(
325			address="%s:%i" % (dbhost, dbport),
326			database_name=dbname,
327			instance_name=instance_name
328			)
329

isomeric / isomer

tests.conftest.Watcher.__init__() A last analyzed 2021-02-24 12:38 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

tests.conftest.Watcher.init() A
last analyzed 2021-02-24 12:38 UTC