40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
|
"""Fixtures for crystallized regression tests.
|
||
|
|
|
||
|
|
The only framework code a crystallized test touches is the observation channel —
|
||
|
|
because judging still requires observing the system independently, whether or not
|
||
|
|
a model was involved in getting there (D-07). Nothing here discovers anything.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import threading
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from lab.http_api import serve
|
||
|
|
from lab.mutations import ObservationChannel, build_lab
|
||
|
|
from testdriver.observers import StateObserver, Watch
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def crystallized_world(request):
|
||
|
|
"""A live lab, the owner's token, and an independent observation function."""
|
||
|
|
mutations = getattr(request, "param", ())
|
||
|
|
app, tokens = build_lab(*mutations)
|
||
|
|
app.request(tokens["alice"], "create_resource",
|
||
|
|
resource_id="R", content="the secret")
|
||
|
|
server = serve(app)
|
||
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||
|
|
observer = StateObserver(
|
||
|
|
ObservationChannel(app), (Watch("bob", "R"), Watch("carol", "R"))
|
||
|
|
)
|
||
|
|
try:
|
||
|
|
yield (
|
||
|
|
f"http://127.0.0.1:{server.server_address[1]}",
|
||
|
|
tokens["alice"],
|
||
|
|
observer.snapshot,
|
||
|
|
)
|
||
|
|
finally:
|
||
|
|
server.shutdown()
|
||
|
|
server.server_close()
|