forked from danihodovic/celery-exporter
-
Notifications
You must be signed in to change notification settings - Fork 0
/
conftest.py
119 lines (93 loc) · 3 KB
/
conftest.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import socket
import threading
import pytest
from src.exporter import Exporter
def pytest_addoption(parser):
parser.addoption(
"--broker",
action="store",
default="redis",
help="What broker to use in tests",
choices=("redis", "rabbitmq", "memory"),
)
parser.addoption(
"--loglevel",
action="store",
default="INFO",
help="Log level of the exporter and celery worker in tests",
choices=("DEBUG", "INFO", "WARNING", "ERROR"),
)
@pytest.fixture(scope="session")
def broker(request):
return request.config.getoption("--broker")
@pytest.fixture(scope="session")
def log_level(request):
return request.config.getoption("--loglevel")
@pytest.fixture(scope="session")
def celery_config(broker):
config = dict(
task_send_sent_event=True,
worker_send_task_events=True,
)
if broker == "redis":
config["broker_url"] = "redis://localhost:6379/" # type: ignore
elif broker == "rabbitmq":
config["broker_url"] = "amqp://guest:guest@localhost:5672" # type: ignore
elif broker == "memory":
config["broker_url"] = "memory://localhost/" # type: ignore
return config
# https://github.com/celery/celery/pull/6632
@pytest.fixture(scope="session")
def celery_worker_parameters(log_level):
return dict(
loglevel=log_level,
without_heartbeat=False,
)
@pytest.fixture(scope="session")
def celery_enable_logging(log_level):
return log_level == "DEBUG"
@pytest.fixture(scope="session")
def find_free_port():
"""
https://gist.github.com/bertjwregeer/0be94ced48383a42e70c3d9fff1f4ad0
"""
def _find_free_port():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("0.0.0.0", 0))
portnum = s.getsockname()[1]
s.close()
return portnum
return _find_free_port
@pytest.fixture()
def exporter_instance(find_free_port, celery_config, log_level):
cfg = {
"host": "0.0.0.0",
"port": find_free_port(),
"broker_url": celery_config["broker_url"],
"broker_transport_option": ["visibility_timeout=7200"],
"broker_ssl_option": [],
"retry_interval": 5,
"log_level": log_level,
"accept_content": None,
"worker_timeout": 1,
"purge_offline_worker_metrics": 10,
"initial_queues": ["queue_from_command_line"],
}
exporter = Exporter(
worker_timeout_seconds=cfg["worker_timeout"],
purge_offline_worker_metrics_seconds=cfg["purge_offline_worker_metrics"],
initial_queues=cfg["initial_queues"],
)
setattr(exporter, "cfg", cfg)
yield exporter
@pytest.fixture()
def threaded_exporter(exporter_instance):
thread = threading.Thread(
target=exporter_instance.run, args=(exporter_instance.cfg,), daemon=True
)
thread.start()
yield exporter_instance
@pytest.fixture()
def hostname():
return socket.gethostname()