userver: samples/testsuite-support/tests/test_logcapture.py
Loading...
Searching...
No Matches
samples/testsuite-support/tests/test_logcapture.py
1import pytest
2from pytest_userver.plugins import log_capture
3
4
5# /// [select]
6async def test_select(service_client):
7 async with service_client.capture_logs(log_level='INFO') as capture:
8 response = await service_client.get('/logcapture')
9 assert response.status == 200
10
11 records = capture.select(
12 text='Message to capture',
13 link=response.headers['x-yarequestid'],
14 )
15 assert len(records) == 1, capture.select()
16 # /// [select]
17
18
19async def test_subscribe(service_client, mockserver):
20 async with service_client.capture_logs() as capture:
21
22 @capture.subscribe(
23 text='Message to capture',
24 trace_id=mockserver.trace_id,
25 )
26 def log_event(link, **other):
27 pass
28
29 response = await service_client.get(
30 '/logcapture',
31 headers={'x-yatraceid': mockserver.trace_id},
32 )
33 assert response.status == 200
34
35 call = await log_event.wait_call()
36 assert call['link'] == response.headers['x-yarequestid']
37
38
39async def test_select_error(service_client):
40 async with service_client.capture_logs(log_level='INFO') as capture:
41 response = await service_client.get('/logcapture')
42 assert response.status == 200
43
44 with pytest.raises(log_capture.IncorrectUsageError):
45 capture.select(
46 text='Message to capture',
47 link=response.headers['x-yarequestid'],
48 )
49
50
51async def test_subscribe_error(service_client, mockserver):
52 async with service_client.capture_logs() as capture:
53 response = await service_client.get(
54 '/logcapture',
55 headers={'x-yatraceid': mockserver.trace_id},
56 )
57 assert response.status == 200
58
59 with pytest.raises(log_capture.IncorrectUsageError):
60
61 @capture.subscribe(
62 text='Message to capture',
63 trace_id=mockserver.trace_id,
64 )
65 def log_event(link, **other):
66 pass