nh2.mock

[nmlorg]

Before building anything else, I want to formalize the mocking process.

I want to be able to reuse this in consumers (like ntelebot), which won't necessarily expose direct access to a Connection, etc.

I'm thinking of pulling:

nh2/nh2/connection.py

Lines 21 to 24 in 350d703

	def __init__(self, host, port):
	self.host = host
	sock = socket.create_connection((host, port))
	self.s = ctx.wrap_socket(sock, server_hostname=host)

into a separate Connection.connect, then having the mocking system monkeypatch Connection.connect so it instantiates a mock server that's actually listening on a random port, then actually connecting the Connection to it (through a real socket), and having the server pull expected (and to-be-replied) events from a global transcript.

The general pattern, using pytest.raises as an example, might look something like:

def test_xxx():
    with nh2.mock.expect(
        ⋮  # Describe the RemoteSettingsChanged for consumption, and whatever its response is supposed to be for emission.
    ):
        conn = nh2.connection.Connection('example.com', 443)

    with nh2.mock.expect(
        ⋮  # Consume the RequestReceived, emit whatever send_headers emits.
    ):
        conn.request('GET', '/test')

 
Tests would feed events into the mocking system by targeting the individual Connection instance:  host, port, and Connection instance number (to be able to test reconnection/request migration in an upcoming ConnectionManager, etc.). This could almost look like:

with nh2.mock.expect(
    'example.com', 443, 1, h2.events.RemoteSettingsChanged.from_settings(…),
):
    conn = nh2.connection.Connection('example.com', 443)

with nh2.mock.expect(
    'example.com', 443, 1, h2.events.RequestReceived(stream_id=1, headers=[(':method', 'GET'), (':path', ('/test'), (':authority', 'example.com'), (':scheme', 'https')]),
).respond(
    'example.com', 443, 1, ???,
):
    conn.request('GET', '/test')

but h2.events.RequestReceived's __init__ doesn't accept arguments like that (and there's no equivalent to h2.events.RemoteSettingsChanged.from_settings). (And frankly that syntax is really clunky.)

Adapting the transcript format from foyerbot/ntelebot/metabot might look something like:

with nh2.mock.expect("""
example.com 443 1 <<< <RemoteSettingsChanged changed_settings:{ChangedSetting(setting=SettingCodes.HEADER_TABLE_SIZE, original_value=4096, new_value=4096), ChangedSetting(setting=SettingCodes.ENABLE_PUSH, original_value=1, new_value=1), ChangedSetting(setting=SettingCodes.INITIAL_WINDOW_SIZE, original_value=65535, new_value=65535), ChangedSetting(setting=SettingCodes.MAX_FRAME_SIZE, original_value=16384, new_value=16384), ChangedSetting(setting=SettingCodes.ENABLE_CONNECT_PROTOCOL, original_value=0, new_value=0), ChangedSetting(setting=SettingCodes.MAX_CONCURRENT_STREAMS, original_value=None, new_value=100), ChangedSetting(setting=SettingCodes.MAX_HEADER_LIST_SIZE, original_value=None, new_value=65536)}>
"""):
    conn = nh2.connection.Connection('example.com', 443)

with nh2.mock.expect("""
example.com 443 1 <<< <RequestReceived stream_id:1, headers:[(b':method', b'GET'), (b':path', b'/test'), (b':authority', b'example.com'), (b':scheme', b'https')]>
example.com 443 1 <<< <StreamEnded stream_id:1>
example.com 443 1 >>> ??? send_headers(???)
"""):
    conn.request('GET', '/test')

but that RemoteSettingsChanged line is abominable, and I'm not sure how to represent the response. (It doesn't look like responses are ever actually instantiated as events in h2, methods like H2Connection.send_headers instantiate, manipulate, then serialize hyperframe.frame.HeadersFrame, etc., so the only readily available syntax describing a response is the literal send_headers method call.)

Eventually I'll want to expose just simple fully-formed requests and fully-formed responses, but testing things like:

nh2/nh2/connection.py

Lines 115 to 124 in 350d703

	def send(self):
	"""Send as much of the request's body as the stream's window allows."""
	while self.tosend and (window := self.connection.c.local_flow_control_window(
	self.stream_id)):
	limit = min(window, self.connection.c.max_outbound_frame_size)
	data = self.tosend[:limit]
	self.tosend = self.tosend[limit:]
	self.connection.c.send_data(self.stream_id, data, end_stream=not self.tosend)
	self.connection.flush()

requires much lower-level control:

nh2/nh2/test_connection.py

Lines 37 to 87 in 350d703

	def test_LiveRequest_send(): # pylint: disable=invalid-name
	"""Verify the body-chunking logic."""
	class MockH2Connection: # pylint: disable=missing-class-docstring,missing-function-docstring
	max_outbound_frame_size = 7
	window = 5
	def __init__(self):
	self.sent = []
	def local_flow_control_window(self, unused_stream_id):
	return self.window
	@staticmethod
	def send_headers(stream_id, unused_headers, **unused_kwargs):
	pass
	def send_data(self, unused_stream_id, data, **unused_kwargs):
	self.sent.append(data)
	self.window -= len(data)
	class MockConnection: # pylint: disable=missing-class-docstring,missing-function-docstring
	c = MockH2Connection()
	@staticmethod
	def new_stream(unused_live_request):
	return 101
	@staticmethod
	def flush():
	pass
	conn = MockConnection()
	request = nh2.rex.Request('POST', 'example.com', '/data', body='555557777777333')
	live_request = nh2.connection.LiveRequest(conn, request)
	assert conn.c.window == 0
	assert conn.c.sent == [b'55555']
	assert live_request.tosend == b'7777777333'
	live_request.send()
	assert conn.c.window == 0
	assert conn.c.sent == [b'55555']
	assert live_request.tosend == b'7777777333'
	conn.c.window = 100
	live_request.send()
	assert conn.c.window == 90
	assert conn.c.sent == [b'55555', b'7777777', b'333']
	assert live_request.tosend == b''

[nmlorg]

For ntelebot, I created an autouse fixture that both enabled requests-mock and made it straightforward to set mock responses right on ntelebot.bot.Bot instances. To use this in consumers (like metabot), I explicitly imported that fixture from ntelebot.conftest into metabot.conftest to make it take effect.

I remember the concept of "entry points" (and the string "pytest11" is also very familiar), and I'm not sure why I didn't use that. (I don't appear to have documented anything 🙁.)

Assuming there was no good reason (that I'm just forgetting), I'm currently thinking I should create another autouse fixture, but this time register it as an entry point in pyproject.toml:

[project.entry-points.pytest11]
nh2_mock = 'nh2._pytest_plugin'

Then any project that installs nh2 would get the plugin installed into itself as well (no need to import one conftest into another), and should therefore get the fixture, which should be able to universally disable all network I/O for nh2 (unless explicitly allowed by interacting with the nh2_mock fixture).

So the general pattern might look like:

def test_basic(nh2_mock):
    with nh2_mock("""
GET https://example.com/test -> 200 {"a": "b"}
"""):
        conn = nh2.connection.Connection('example.com', 443)
        assert conn.request('GET', '/test').wait().json() == {'a': 'b'}


def test_live_request(nh2_mock):
    with nh2_mock.live:
        conn = nh2.connection.Connection('httpbin.org', 443)
        assert conn.request('GET', …  # Make an actual request to httpbin.org.


def test_low_level(nh2_mock):
    with nh2_mock.lowlevel("""
example.com 443 1 <<< <RequestReceived stream_id:1, headers:[(b':method', b'GET'), (b':path', b'/test'), (b':authority', b'example.com'), (b':scheme', b'https')]>
example.com 443 1 <<< <StreamEnded stream_id:1>
example.com 443 1 >>> ??? send_headers(???)
"""):
        conn = nh2.connection.Connection('example.com', 443)
        assert conn.request('GET', '/test').wait().json() == {'a': 'b'}

---

To do:

1. Come up with an idea for how to express responses in the lowlevel transcript format.
2. Come up with some way to express stuff like:

       def _handler(request, unused_context):
           responses.append(json.loads(request.body.decode('ascii')))
           return {'ok': True, 'result': {}}
   
       self.bot.answer_inline_query.respond(json=_handler)

3. Should I also support ntelebot-like nh2.connection.Connection('example.com', 443).respond('GET', '/test').with(json={'a': 'b'})? That would make 2 a lot simpler.

[nmlorg]

Stage 1

Monkeypatched Connection.connect() sets Connection.mock_server = MockServer() and connects to it.

MockServer has a fully functional h2 connection instance and simply stores all events it receives.

async def test_blah():
    conn = await nh2.connection.Connection('example.com', 443)
    assert conn.mock_server.get_events() == [/"""
 ⋮
]/"""

Maybe even:

    assert conn.get_events() = …

?

Then just:

    conn.mock_server.send_headers(…)

duh! (Right?)

 

Stage 2

Make MockServer() [more] explicit:

async def test_blah():
    with ng2.mock.expect('example.com', 443) as mock_server:
        conn = await nh2.connection.Connection('example.com', 443)
    assert conn.mock_server is mock_server
    assert mock_server.get_events() == [/"""
 ⋮
]/"""
    mock_server.send_headers(…)
    assert conn.get_events() = …

 

Stage 3

For things like ntelebot.bot.Bot, the Connection instance will never be directly exposed, so anything like conn.get_events() will have to be called on the MockServer:

async def test_bot():
    bot = ntelebot.bot.Bot(token)
    with ng2.mock.expect('api.telegram.org', 443) as tba:
        assert await bot.get_me() == …
    assert tba.client.get_events() == …

 
However, await bot.get_me() exposes the separation between the two internal awaits:  to Connection.s.send() (via Connection.flush()) and then to Connection.s.receive() (via LiveRequest.wait()). I can't assert what the server has received until after the first await, but I can't then assert the response of the call until I have provided the response (as the server).

This suggests I won't even be able to define bot.get_me() (etc.) as:

def get_me(self):
    live_request = await self.conn.request(…)
    response = await live_request.wait()
    return response.json()

because — for testing — I need to take control (to assert the request and provide the response) between the two awaits.

Off the top of my head, I'm considering something like run_until_send(coro), to do something like monkeypatch LiveRequest.wait to have it maybe raise an exception… or something…? The test would then be:

async def test_bot():
    bot = ntelebot.bot.Bot(token)
    with ng2.mock.expect('api.telegram.org', 443) as tba:
        paused_coro = nh2.mock.run_until_send(bot.get_me())
    assert tba.get_events() == …
    tba.send_headers(…)
    response = nh2.mock.run_until_send(paused_coro)
    assert response == …

I think this is the only way to handle testing "workflow" functions (which send multiple HTTP requests — and await their responses — before returning).

[nmlorg]

Given a function like:

async def blah(sock):
    await sock.send('cat')
    data = await sock.receive()
    assert data == 'mew'
    await sock.send('dog')
    data = await sock.receive()
    assert data == 'arf'
    return 'zebra'

I want to be able to write a test like roughly:

async def test_blah():
    left, right = create_pipe()
    coro = blah(left)
    run_until_next_send(coro)
    data = await right.receive()
    assert data == 'cat'
    await right.send('mew')
    run_until_next_send(coro)
    data = await right.receive()
    assert == 'dog'
    await right.send('arf')
    assert run_until_next_send(coro) == 'zebra'

With sock.send and run_until_next_send working together in some way.

---

I had been thinking sock.send would just throw an exception, which run_until_next_send would catch, run the server's side of the pipe, then wind the coroutine back and continue from the exception, but it doesn't look like anything like that is possible:

class MyError(Exception):
    pass

async def blah():
    print('running')
    raise MyError('hi')
    print('finishing')

coro = blah()
try:
    coro.send(None)
except MyError as e:
    print('got', e)
coro.send(None)

running
got hi
Traceback (most recent call last):
  File "test.py", line 14, in <module>
    coro.send(None)
RuntimeError: cannot reuse already awaited coroutine

---

This is basically setting a breakpoint, so I started looking into things like bdb and breakpoint(), and it seems like it literally just inserts a call to sys.breakpointhook() (which defaults to pdb.set_trace) at the call point — so the function being debugged isn't "paused" and then "resumed" so much as Python just acts as if there's a call to a function that isn't actually in its code.

---

I could go back to what I was doing with nsync and finish writing my own coroutine runner… but the code would still be doing heavy anyio socket I/O, etc., so I think I'd have to end up actually subclassing or monkeypatching anyio's runner to actually implement this (rather than calling coro.send(value), etc., myself).

---

For the simple case of directly interacting with nh2.connection.Connection instances, I can get away with:

async def test_blah():
    mock_server = await nh2.mock.expect_connect(…)
    conn = await nh2.connection.Connection(…)

    live_request = await conn.request(…)
    assert await mock_server.read() == …
    await mock_server.[respond]

    response = await live_request.wait()
    assert response.data == …

because nothing causes Connection to block waiting for data from the server implicitly.

However, for complicated workflows (like blah), I think I'll need to resort to just running the function as a background task.

Honestly, the main stumbling block is just how awkward working with return values from coroutines is (at least with anyio). This seems to have been intentional, and I still don't understand why — something to do with what happens when you cancel a task, I guess.

Anyway, if I recreate asyncio.create_task's behavior of returning an awaitable that returns the function's return value (basically reimplementing asyncio.Future out of an anyio.Event, which in turn is built on top of asyncio.Event — which is just an asyncio.Future that uses True as the finalized value 🙄), I should be able to do something like:

async def blah(sock):
    await sock.send('cat')
    data = await sock.receive()
    assert data == 'mew'
    await sock.send('dog')
    data = await sock.receive()
    assert data == 'arf'
    return 'zebra'

async def test_blah():
    with TaskGroupWrapper() as tg:
        left, right = create_pipe()
        future = tg.start_soon(blah, left)  #coro = blah(left)
        #run_until_next_send(coro) — blah automatically starts.
        data = await right.receive()
        assert data == 'cat'
        await right.send('mew')
        #run_until_next_send(coro) — blah automatically resumes.
        data = await right.receive()
        assert == 'dog'
        await right.send('arf')
        #run_until_next_send(coro) — blah automatically resumes.
        assert await future == 'zebra'