Something went wrong. Try again.
Paper Mario mod hosting site - AT Protocol appview starhaven.dev
appview
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194"""Fixtures for end-to-end tests."""
import contextlibimport osimport socketimport subprocessimport tempfileimport timeimport uuidfrom collections.abc import Iteratorfrom pathlib import Pathfrom typing import NamedTuple
import pytestimport requestsfrom helpers import Actorfrom playwright.sync_api import Browser, Page
REPO_ROOT = Path(__file__).resolve().parent.parent
# dev-pds seeds this account as the last thing it does at startup, which makes# it a readiness signal for the whole PDS. Tests make their own accounts.SEEDED_HANDLE = "alice.test"
class DevPds(NamedTuple): """A dev PDS, and the PLC directory its DIDs are registered in."""
pds_url: str plc_url: str
def _free_ports(count: int) -> list[int]: """Ports nothing else is listening on, all bound at once so they differ.""" with contextlib.ExitStack() as stack: sockets = [ stack.enter_context(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) for _ in range(count) ] for s in sockets: s.bind(("127.0.0.1", 0)) return [s.getsockname()[1] for s in sockets]
def _responds_ok(url: str, params: dict[str, str] | None) -> bool: try: return requests.get(url, params=params, timeout=1).status_code == 200 except requests.RequestException: return False
@contextlib.contextmanagerdef _server( argv: list[str], url: str, params: dict[str, str] | None = None, env: dict[str, str] | None = None,) -> Iterator[Path]: """Runs argv until the block exits, entered once GET url answers 200.
Output goes to a file rather than a pipe, so a chatty server can't fill a pipe buffer and block on a write nobody is reading. The block gets the file's path, to read what the server has printed so far. It is printed on the way out, which pytest shows only for tests that failed. """ with tempfile.NamedTemporaryFile(mode="w+") as log: process = subprocess.Popen(argv, env=env, stdout=log, stderr=subprocess.STDOUT) try: deadline = time.monotonic() + 30 while not _responds_ok(url, params): if process.poll() is not None or time.monotonic() > deadline: raise RuntimeError(f"{argv[0]} is not serving {url}") time.sleep(0.1) yield Path(log.name) finally: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() log.seek(0) print(f"--- {argv[0]}\n{log.read()}")
@pytest.fixture(scope="session")def starhaven_binary() -> Path: subprocess.run(["cargo", "build", "--bin", "server"], cwd=REPO_ROOT, check=True) return REPO_ROOT / "target" / "debug" / "server"
@pytest.fixture(scope="session")def dev_pds() -> Iterator[DevPds]: """A dev PDS+PLC pair, shared by every test.""" binary = os.environ.get("STARHAVEN_DEV_PDS_BIN") assert binary, "STARHAVEN_DEV_PDS_BIN is unset: run the tests in `nix develop`"
pds_port, plc_port = _free_ports(2) pds_url = f"http://localhost:{pds_port}" env = {**os.environ, "PDS_PORT": str(pds_port), "PLC_PORT": str(plc_port)} probe = f"{pds_url}/xrpc/com.atproto.identity.resolveHandle" with _server([binary], probe, {"handle": SEEDED_HANDLE}, env): yield DevPds(pds_url, f"http://localhost:{plc_port}")
@pytest.fixturedef server_args(request: pytest.FixtureRequest) -> list[str]: """Extra flags for the appview under test.
Parametrize it indirectly to run one test against a differently configured server: `@pytest.mark.parametrize("server_args", [[...]], indirect=True)`. """ return getattr(request, "param", [])
class Appview(NamedTuple): """A freshly started server."""
url: str # Everything the server prints, such as the emails it would send. log: Path # Its SQLite database, for a test to reach past the app into. database: Path
@pytest.fixturedef appview( starhaven_binary: Path, dev_pds: DevPds, server_args: list[str]) -> Iterator[Appview]: """A freshly started server, torn down after the test.""" (port,) = _free_ports(1) base_url = f"http://127.0.0.1:{port}" with tempfile.TemporaryDirectory() as data_dir: argv = [ str(starhaven_binary), "--listen", f"127.0.0.1:{port}", "--external-base", base_url, "--data-dir", data_dir, "--dev-pds-url", dev_pds.pds_url, "--dev-plc-url", dev_pds.plc_url, # Not the devshell server's bucket, whose deliverables a test # server's sweep would see as never published. "--s3-bucket", "starhaven-test", *server_args, ] with _server(argv, base_url) as log: yield Appview(base_url, log, Path(data_dir) / "database.db")
@pytest.fixturedef appview_url(appview: Appview) -> str: """Base URL of a freshly started server, torn down after the test.""" return appview.url
def _new_actor(page: Page, appview_url: str, dev_pds: DevPds) -> Actor: """An account of its own on the dev PDS, acting through `page`.""" handle = f"test-{uuid.uuid4().hex[:8]}.test" password = "hunter2" response = requests.post( f"{dev_pds.pds_url}/xrpc/com.atproto.server.createAccount", json={"handle": handle, "email": f"{handle}@dev.test", "password": password}, timeout=10, ) response.raise_for_status() did = response.json()["did"] return Actor(page, appview_url, dev_pds.pds_url, handle, password, did)
@pytest.fixturedef actor(page: Page, appview_url: str, dev_pds: DevPds) -> Actor: """An account of its own for the test, so its repo starts out empty.
A repo belongs to one DID, so a fresh account is as clean as a fresh PDS and costs a request rather than a process. """ return _new_actor(page, appview_url, dev_pds)
@pytest.fixturedef other_actor(browser: Browser, appview_url: str, dev_pds: DevPds) -> Iterator[Actor]: """A second account, in a browser of its own, for tests between two people.""" context = browser.new_context() try: yield _new_actor(context.new_page(), appview_url, dev_pds) finally: context.close()