"""DEV / TEST ONLY — the catalogue's composition served on this machine, so the LIVE runner can be pointed at it.
with local.serve(CAT, engine_for, issuer) as (base_url, public_base_url):
LiveRunner(CAT, base_url=base_url, public_base_url=public_base_url, allow=True, secrets=d).run_all()
The staff app runs behind scenarios.gateway.LocalGateway (bearer tokens from a LocalIssuer, verified as the
Ignite gateway would), the public app beside it, each on its own uvicorn server on a free local port — real
sockets and real HTTP, so the live runner's whole path is exercised. `engine_for` decides where the rows go:
SQLite files (tests/test_scenarios.py), or one real DataK3 engine (scripts/live_scenario_check.py).
Needs uvicorn (the `test` extra).
"""
import contextlib
import os
import socket
import threading
import time
from contextlib import ExitStack
from typing import Any, Callable
from erp_platform.scenarios.gateway import LocalGateway, LocalIssuer
from erp_platform.scenarios.inprocess import stage
def free_port() -> int:
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@contextlib.contextmanager
def _uvicorn(app: Any, port: int):
import uvicorn
server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning",
access_log=False, lifespan="on"))
thread = threading.Thread(target=server.run, name=f"uvicorn-{port}", daemon=True)
thread.start()
deadline = time.monotonic() + 20
while not server.started:
if time.monotonic() > deadline or not thread.is_alive():
raise RuntimeError(f"the local server on port {port} did not start")
time.sleep(0.02)
try:
yield f"http://127.0.0.1:{port}"
finally:
server.should_exit = True
thread.join(10)
@contextlib.contextmanager
def serve(catalogue, engine_for: Callable[[str], Any], issuer: LocalIssuer, *, create: bool = True):
with ExitStack() as st:
staff_port, public_port = free_port(), free_port()
public_base = f"http://127.0.0.1:{public_port}"
before = os.environ.get("PUBLIC_LINK_BASE_URL")
os.environ["PUBLIC_LINK_BASE_URL"] = public_base
st.callback(lambda: os.environ.__setitem__("PUBLIC_LINK_BASE_URL", before) if before is not None
else os.environ.pop("PUBLIC_LINK_BASE_URL", None))
staged = stage(catalogue, st, engine_for, create=create)
public_url = st.enter_context(_uvicorn(staged.public_app, public_port))
staff_url = st.enter_context(_uvicorn(LocalGateway(staged.app, issuer), staff_port))
yield staff_url, public_url
def write_secrets(directory, issuer_key: bytes, *, issuer: str, audience: str) -> None:
"""A secrets directory for a local issuer: personas.json + the key, both mode 600."""
import json
import pathlib
d = pathlib.Path(directory)
d.mkdir(parents=True, exist_ok=True)
os.chmod(d, 0o700)
for name, content in (("signing-key.txt", issuer_key.decode()),
("personas.json", json.dumps({"local_signing_key": "signing-key.txt", "issuer": issuer,
"audience": audience}))):
path = d / name
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as f:
f.write(content)
os.chmod(path, 0o600)