this is a fun webpage +
hosted by greenbean diff --git a/e2e/binary/greenbean/serve.py b/e2e/binary/greenbean/serve.py new file mode 100644 index 0000000000000000000000000000000000000000..9c5d79ba9cc877a8dafcb167c4c33d7c45c62a34 --- /dev/null +++ b/e2e/binary/greenbean/serve.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from os import read +from pathlib import Path +from select import select +from shutil import copyfileobj +from subprocess import PIPE, Popen +from urllib.request import urlopen + +from binary import Diff, Relative, Tool, sniff +from binary.port import listening +from pytest import mark + + +@mark.skipif(listening(8080), reason="port 8080 is in use") +def test_serve(tool: Tool, relative: Relative, tmp_path: Path) -> None: + binary = tool("greenbean") + expected = relative("expected.html") + output = tmp_path / "output.html" + + cmd = (binary, "1") + server = Popen(cmd, stderr=PIPE, encoding="utf8", cwd=tmp_path) + match = sniff(server.stderr.fileno(), r"listening on (http://.+)", timeout=2) + host = match.group(1) + + with urlopen(host) as src, open(output, "w+b") as dst: + assert 200 <= src.status < 300 + copyfileobj(src, dst) + + assert Diff(expected) == Diff(output) diff --git a/e2e/binary/port.py b/e2e/binary/port.py new file mode 100644 index 0000000000000000000000000000000000000000..221eb68a5be486ee2d7e56b0433b9114f4ba0d18 --- /dev/null +++ b/e2e/binary/port.py @@ -0,0 +1,8 @@ +from __future__ import annotations + +from socket import AF_INET, SOCK_STREAM, socket + + +def listening(port: int, host: str = "localhost") -> bool: + with socket(AF_INET, SOCK_STREAM) as s: + return s.connect_ex((host, port)) == 0 diff --git a/e2e/binary/sniff.py b/e2e/binary/sniff.py new file mode 100644 index 0000000000000000000000000000000000000000..cd98538ad4ca70d6b615a6446e230de35d497455 --- /dev/null +++ b/e2e/binary/sniff.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from concurrent.futures import Executor, ThreadPoolExecutor, TimeoutError +from os import read +from re import Match, Pattern, compile +from select import select + + +def consume(fd: int, /) -> bytes: + buffer = b"" + while r := read(fd, 1024): + buffer += r + if len(r) < 1024: + break + return buffer + + +def detect(fd: int, regex: Pattern[str], /, timeout: int) -> Match[str]: + buffer = b"" + while True: + ready, _, _ = select((fd,), (), (), timeout) + if fd not in ready: + raise TimeoutError(f"failed to select on {fd}") + + buffer += consume(fd) + if not buffer: + raise EOFError("no data to sniff") + line, found, buffer = buffer.partition(b"\n") + if not found: + continue + if match := regex.match(line.decode("utf8")): + return match + + +def pooled(pool: Executor, fd: int, regex: Pattern[str], /, timeout: int) -> str: + future = pool.submit(detect, fd, regex, timeout) + return future.result(timeout=timeout) + + +def threaded(fd: int, regex: Pattern[str], /, threads: int, timeout: int) -> str: + with ThreadPoolExecutor(max_workers=threads) as pool: + return pooled(pool, fd, regex, timeout=timeout) + + +def sniff(fd: int, regex: Pattern[str] | str, /, timeout: int) -> str: + if isinstance(regex, str): + regex = compile(regex) + return threaded(fd, regex, timeout=timeout, threads=1)