-
Notifications
You must be signed in to change notification settings - Fork 1
/
conftest.py
82 lines (55 loc) · 1.49 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
"""
Testing utility fixtures
"""
import uuid
import pytest
from kafka import KafkaConsumer as Consumer, KafkaAdminClient
from kafka import KafkaProducer as Producer
from nameko.containers import ServiceContainer
@pytest.fixture
def topic():
return "test"
@pytest.fixture
def random_topic():
return str(uuid.uuid4())
@pytest.fixture
def partition():
return 0
@pytest.fixture
def kafka_admin():
_client = KafkaAdminClient()
yield _client
_client.close()
@pytest.fixture
def producer():
producer = Producer()
yield producer
producer.close()
@pytest.fixture
def consumer(topic, kafka_admin):
consumer = Consumer(topic, group_id=topic)
yield consumer
consumer.close()
@pytest.fixture
def entrypoint_tracker():
class CallTracker(object):
def __init__(self):
self.calls = []
def __len__(self):
return len(self.calls)
def track(self, **call):
self.calls.append(call)
def get_results(self):
return [call['result'] for call in self.calls]
def get_exceptions(self):
return [call['exc_info'] for call in self.calls]
return CallTracker()
@pytest.fixture
def wait_for_result(entrypoint_tracker):
def cb(worker_ctx, res, exc_info):
entrypoint_tracker.track(result=res, exc_info=exc_info)
return True
return cb
@pytest.fixture
def mock_container(mocker):
return mocker.Mock(spec=ServiceContainer, config={}, service_name="test")