okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffaufrefactor(test): share inspection and stream supportcba5ff8

main
branchesmainall refs →
1.2 KiB38 linesraw
1"""Unread stream fixtures and bounded subscriber inspection for E2E tests."""
2
3import socket
4
5from harness import ROOT
6from inspect_support import Inspection
7
8
9class StreamProbe:
10    def __init__(self, app):
11        self.inspection = Inspection(
12            app, ROOT / "test/e2e/albedo_stream_pressure_probe.erl"
13        )
14
15    def call(self, function, arguments="[]"):
16        return self.inspection.call_json(function, arguments)
17
18
19def unread_stream(app, path):
20    connection = socket.socket()
21    connection.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
22    connection.settimeout(15)
23    connection.connect(("127.0.0.1", app.connection["port"]))
24    connection.sendall(
25        (
26            f"GET {path} HTTP/1.1\r\nHost: localhost:{app.connection['port']}\r\n"
27            f"Authorization: Bearer {app.connection['token']}\r\n"
28            "Accept: text/event-stream\r\n\r\n"
29        ).encode()
30    )
31    # Read only the readiness batch, then leave all later socket data unread.
32    initial = b""
33    while b"data: " not in initial or b"\n\n" not in initial.split(b"data: ", 1)[1]:
34        part = connection.recv(1)
35        if not part:
36            raise AssertionError("stream closed before readiness")
37        initial += part
38    return connection