6510444197

Committed 13 Oct 2023 04:01PM UTC coverage: 89.043% (+1.0%) from 88.051%

Build # 6510444197

Build Type

Pull #120

github

Committed by

web-flow

Commit Message

Merge 09a3c124a into c744dd551

Pull Request Pull Request #120: feat: add prometheus instrumentation

Run Details

279 of 348 branches covered (0.0%)

Branch coverage included in aggregate %.

84 of 84 new or added lines in 8 files covered. (100.0%)

28 existing lines in 2 files now uncovered.

1265 of 1386 relevant lines covered (91.27%)

2.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.08

/hurricane/testing/testcases.py

from typing import Any

from django.test import SimpleTestCase

from hurricane.testing.actors import HTTPClient, WebhookTestHandler
from hurricane.testing.drivers import (
    HurricaneAMQPDriver,
    HurricaneServerDriver,
    HurricaneWebhookServerDriver,
)


class HurricanBaseTest(SimpleTestCase):
    driver: Any = None

    @property
    def probe_client(self):
        host, port = self.driver.get_server_host_port(probe_port=True)
        return HTTPClient(host, port)

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.driver = cls.driver()


class HurricanServerTest(HurricanBaseTest):
    driver = HurricaneServerDriver

    @property
    def app_client(self):
        host, port = self.driver.get_server_host_port()
        return HTTPClient(host, port)

    @classmethod
    def tearDownClass(cls):
        cls.driver.stop_server()
        super().tearDownClass()

    def _retrieve_env(self, kwargs):
        return kwargs["env"] if "env" in kwargs else {}

    @staticmethod
    def cycle_server(*args, **kwargs):
        def _cycle_server(function):
            def wrapper(self):
                if "args" in kwargs:
                    _args = kwargs["args"]
                else:
                    _args = None
                # run this hurricane server with coverage
                # default is True
                if "coverage" in kwargs:
                    coverage = kwargs["coverage"]
                else:
                    coverage = True
                env = self._retrieve_env(kwargs)
                self.driver.start_server(_args, coverage, env)
                try:
                    function(self)
                except Exception as e:
                    self.driver.stop_server()
                    raise e
                else:
                    self.driver.stop_server()

            return wrapper

        if len(args) == 1 and callable(args[0]):
            return _cycle_server(args[0])
        else:
            return _cycle_server


class HurricaneWebhookServerTest(HurricanServerTest):
    driver = HurricaneWebhookServerDriver  # type: ignore

    @property
    def probe(self):
        return WebhookTestHandler()


class HurricaneAMQPTest(HurricanBaseTest):
    driver = HurricaneAMQPDriver

    @classmethod
    def tearDownClass(cls):
        cls.driver.stop_consumer()
        cls.driver.stop_amqp()
        super().tearDownClass()

    def _retrieve_env(self, kwargs):
        return kwargs["env"] if "env" in kwargs else {}

    @staticmethod
    def cycle_consumer(*args, **kwargs):
        def _cycle_consumer(function):
            def wrapper(self):
                if "args" in kwargs:
                    _args = kwargs["args"]
                else:
                    _args = None
                # run this hurricane consumer with coverage
                # default is True
                if "coverage" in kwargs:
                    coverage = kwargs["coverage"]
                else:
                    coverage = True
                env = self._retrieve_env(kwargs)
                self.driver.start_amqp()
                if "--no_host_port" not in _args:
                    host, port = self.driver.get_amqp_host_port()
                    _args += ["--amqp-port", str(port), "--amqp-host", host]
                else:
                    _args = _args[:-1]
                self.driver.start_consumer(_args, coverage, env)
                try:
                    function(self)
                except Exception as e:
                    self.driver.stop_consumer()
                    self.driver.stop_amqp()
                    raise e
                else:
                    self.driver.stop_consumer()
                    self.driver.stop_amqp()

            return wrapper

        if len(args) == 1 and callable(args[0]):
            return _cycle_consumer(args[0])
        else:
            return _cycle_consumer

1	from typing import Any	3✔
2
3	from django.test import SimpleTestCase	3✔
4
5	from hurricane.testing.actors import HTTPClient, WebhookTestHandler	3✔
6	from hurricane.testing.drivers import (	3✔
7	HurricaneAMQPDriver,
8	HurricaneServerDriver,
9	HurricaneWebhookServerDriver,
10	)
11
12
13	class HurricanBaseTest(SimpleTestCase):	3✔
14	driver: Any = None	3✔
15
16	@property	3✔
17	def probe_client(self):	3✔
18	host, port = self.driver.get_server_host_port(probe_port=True)	3✔
19	return HTTPClient(host, port)	3✔
20
21	@classmethod	3✔
22	def setUpClass(cls):	3✔
23	super().setUpClass()	3✔
24	cls.driver = cls.driver()	3✔
25
26
27	class HurricanServerTest(HurricanBaseTest):	3✔
28	driver = HurricaneServerDriver	3✔
29
30	@property	3✔
31	def app_client(self):	3✔
32	host, port = self.driver.get_server_host_port()	3✔
33	return HTTPClient(host, port)	3✔
34
35	@classmethod	3✔
36	def tearDownClass(cls):	3✔
37	cls.driver.stop_server()	3✔
38	super().tearDownClass()	3✔
39
40	def _retrieve_env(self, kwargs):	3✔
41	return kwargs["env"] if "env" in kwargs else {}	3✔
42
43	@staticmethod	3✔
44	def cycle_server(args, *kwargs):	3✔
45	def _cycle_server(function):	3✔
46	def wrapper(self):	3✔
47	if "args" in kwargs:	3✔
48	_args = kwargs["args"]	3✔
49	else:
50	_args = None	3✔
51	# run this hurricane server with coverage
52	# default is True
53	if "coverage" in kwargs:	3✔
54	coverage = kwargs["coverage"]	3✔
55	else:
56	coverage = True	3✔
57	env = self._retrieve_env(kwargs)	3✔
58	self.driver.start_server(_args, coverage, env)	3✔
59	try:	3✔
60	function(self)	3✔
UNCOV 61	except Exception as e:	×
UNCOV 62	self.driver.stop_server()	×
UNCOV 63	raise e	×
64	else:
65	self.driver.stop_server()	3✔
66
67	return wrapper	3✔
68
69	if len(args) == 1 and callable(args[0]):	3✔
70	return _cycle_server(args[0])	3✔
71	else:
72	return _cycle_server	3✔
73
74
75	class HurricaneWebhookServerTest(HurricanServerTest):	3✔
76	driver = HurricaneWebhookServerDriver # type: ignore	3✔
77
78	@property	3✔
79	def probe(self):	3✔
80	return WebhookTestHandler()	×
81
82
83	class HurricaneAMQPTest(HurricanBaseTest):	3✔
84	driver = HurricaneAMQPDriver	3✔
85
86	@classmethod	3✔
87	def tearDownClass(cls):	3✔
88	cls.driver.stop_consumer()	3✔
89	cls.driver.stop_amqp()	3✔
90	super().tearDownClass()	3✔
91
92	def _retrieve_env(self, kwargs):	3✔
93	return kwargs["env"] if "env" in kwargs else {}	3✔
94
95	@staticmethod	3✔
96	def cycle_consumer(args, *kwargs):	3✔
97	def _cycle_consumer(function):	3✔
98	def wrapper(self):	3✔
99	if "args" in kwargs:	3!
100	_args = kwargs["args"]	3✔
101	else:
102	_args = None	×
103	# run this hurricane consumer with coverage
104	# default is True
105	if "coverage" in kwargs:	3✔
106	coverage = kwargs["coverage"]	3✔
107	else:
108	coverage = True	3✔
109	env = self._retrieve_env(kwargs)	3✔
110	self.driver.start_amqp()	3✔
111	if "--no_host_port" not in _args:	3✔
112	host, port = self.driver.get_amqp_host_port()	3✔
113	_args += ["--amqp-port", str(port), "--amqp-host", host]	3✔
114	else:
115	_args = _args[:-1]	3✔
116	self.driver.start_consumer(_args, coverage, env)	3✔
117	try:	3✔
118	function(self)	3✔
119	except Exception as e:	×
120	self.driver.stop_consumer()	×
121	self.driver.stop_amqp()	×
122	raise e	×
123	else:
124	self.driver.stop_consumer()	3✔
125	self.driver.stop_amqp()	3✔
126
127	return wrapper	3✔
128
129	if len(args) == 1 and callable(args[0]):	3!
130	return _cycle_consumer(args[0])	×
131	else:
132	return _cycle_consumer	3✔

django-hurricane / django-hurricane / 6510444197

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous