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
|
import itertools
import pytest
from hat import aio
from hat import util
import restlog.server
pytestmark = pytest.mark.asyncio
@pytest.fixture
def port():
return util.get_unused_tcp_port()
class Backend(aio.Resource):
def __init__(self):
self._async_group = aio.Group()
self._next_entry_ids = itertools.count(1)
@property
def async_group(self):
return self._async_group
async def register(self, timestamp, address, source, type, data):
return {'entry_id': next(self._next_entry_ids),
'timestamp': timestamp,
'address': address,
'source': source,
'type': type,
'data': data}
async def get_entries(self, source=None, type=None, last_entry_id=None,
max_results=None):
return {'entries': [],
'more': False}
async def get_entry(self, entry_id):
return
async def test_create(port):
backend = Backend()
server = await restlog.server.create('127.0.0.1', port, backend)
assert server.is_open
await server.async_close()
|
