__init__.py
python
sha256:c07f2f34a0db9f43fe866f157d1935322008545ff8940c06c7c921eb219c55ab
NXP-b DONE: independent BV-r2 pass + ISR (SD-17)
Human
minor
⚠ breaking
18 hours ago
| 1 | """Fixture helpers for Track Q / Q1 app tests.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import shutil |
| 7 | import socket |
| 8 | import urllib.error |
| 9 | import urllib.request |
| 10 | from pathlib import Path |
| 11 | from typing import Any |
| 12 | |
| 13 | import yaml |
| 14 | |
| 15 | from cli.context import CliContext |
| 16 | from cli.kit_root import kit_root |
| 17 | from cli.output import OutputContext |
| 18 | from tests.support import FIXTURES, HONESTY, git_status_runner, run_cli, seed_freeze_repo |
| 19 | from tools.app.server import AppServerHandle, build_server_config, start_app_server |
| 20 | |
| 21 | |
| 22 | def free_port() -> int: |
| 23 | """Return an ephemeral loopback port for tests.""" |
| 24 | with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: |
| 25 | sock.bind(("127.0.0.1", 0)) |
| 26 | return int(sock.getsockname()[1]) |
| 27 | |
| 28 | |
| 29 | def seed_app_repo(tmp_path: Path, *, config_name: str = "config-git-only.yaml") -> None: |
| 30 | """Initialize a minimal repo suitable for app handler tests.""" |
| 31 | code = run_cli( |
| 32 | ["init", "--from-config", str(FIXTURES / config_name), "--non-interactive"], |
| 33 | cwd=tmp_path, |
| 34 | kit=kit_root(), |
| 35 | runner=git_status_runner(), |
| 36 | ) |
| 37 | assert code == 0 |
| 38 | docs = tmp_path / "docs" |
| 39 | docs.mkdir(parents=True, exist_ok=True) |
| 40 | (docs / "ROADMAP.md").write_text("# Roadmap fixture\n", encoding="utf-8") |
| 41 | (docs / "OVERSEER-HANDOVER.md").write_text("# Handover fixture\n", encoding="utf-8") |
| 42 | |
| 43 | |
| 44 | def seed_app_e2e_repo(tmp_path: Path) -> Path: |
| 45 | """Seed a repo with freeze review + honesty modules enabled for e2e flows.""" |
| 46 | artifact = seed_freeze_repo(tmp_path) |
| 47 | for rel in ("artifacts", "entries"): |
| 48 | src = HONESTY / rel |
| 49 | dest = tmp_path / rel |
| 50 | if dest.exists(): |
| 51 | shutil.rmtree(dest) |
| 52 | shutil.copytree(src, dest) |
| 53 | config_path = tmp_path / ".overseer" / "config.yaml" |
| 54 | cfg = yaml.safe_load(config_path.read_text(encoding="utf-8")) |
| 55 | honesty_cfg = yaml.safe_load((HONESTY / "config-honesty-enabled.yaml").read_text(encoding="utf-8")) |
| 56 | cfg["honesty"] = honesty_cfg["honesty"] |
| 57 | cfg["modules"] = honesty_cfg["modules"] |
| 58 | config_path.write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8") |
| 59 | return artifact |
| 60 | |
| 61 | |
| 62 | class AppHttpClient: |
| 63 | """Minimal HTTP client for the local app server.""" |
| 64 | |
| 65 | def __init__(self, base_url: str, session: str, csrf: str) -> None: |
| 66 | self.base_url = base_url.rstrip("/") |
| 67 | self.session = session |
| 68 | self.csrf = csrf |
| 69 | |
| 70 | def request( |
| 71 | self, |
| 72 | method: str, |
| 73 | path: str, |
| 74 | *, |
| 75 | body: dict[str, Any] | None = None, |
| 76 | headers: dict[str, str] | None = None, |
| 77 | origin: str | None = None, |
| 78 | ) -> tuple[int, dict[str, Any]]: |
| 79 | url = f"{self.base_url}{path}" |
| 80 | req_headers = dict(headers or {}) |
| 81 | if origin is not None: |
| 82 | req_headers["Origin"] = origin |
| 83 | data = None |
| 84 | if body is not None: |
| 85 | req_headers["Content-Type"] = "application/json" |
| 86 | data = json.dumps(body).encode("utf-8") |
| 87 | request = urllib.request.Request(url, data=data, method=method, headers=req_headers) |
| 88 | try: |
| 89 | with urllib.request.urlopen(request, timeout=10) as response: |
| 90 | payload = json.loads(response.read().decode("utf-8")) |
| 91 | return response.status, payload |
| 92 | except urllib.error.HTTPError as exc: |
| 93 | payload = json.loads(exc.read().decode("utf-8")) |
| 94 | return exc.code, payload |
| 95 | |
| 96 | def get(self, path: str, **kwargs: Any) -> tuple[int, dict[str, Any]]: |
| 97 | headers = kwargs.pop("headers", {}) |
| 98 | headers.setdefault("Authorization", f"Bearer {self.session}") |
| 99 | return self.request("GET", path, headers=headers, **kwargs) |
| 100 | |
| 101 | def post(self, path: str, body: dict[str, Any] | None = None, **kwargs: Any) -> tuple[int, dict[str, Any]]: |
| 102 | headers = kwargs.pop("headers", {}) |
| 103 | headers.setdefault("Authorization", f"Bearer {self.session}") |
| 104 | headers.setdefault("X-Overseer-CSRF", self.csrf) |
| 105 | return self.request("POST", path, body=body or {}, headers=headers, **kwargs) |
| 106 | |
| 107 | |
| 108 | def start_test_app( |
| 109 | repo_root: Path, |
| 110 | *, |
| 111 | port: int | None = None, |
| 112 | runner=None, |
| 113 | review_provider_factory=None, |
| 114 | ) -> tuple[AppServerHandle, AppHttpClient]: |
| 115 | """Start an app server against ``repo_root`` and return handle + client.""" |
| 116 | chosen = port or free_port() |
| 117 | ctx = CliContext.create( |
| 118 | cwd=repo_root, |
| 119 | kit=kit_root(), |
| 120 | runner=runner or git_status_runner(), |
| 121 | output=OutputContext(), |
| 122 | review_provider_factory=review_provider_factory, |
| 123 | ) |
| 124 | config = build_server_config( |
| 125 | repo_root=repo_root, |
| 126 | bind="127.0.0.1", |
| 127 | port=chosen, |
| 128 | ctx=ctx, |
| 129 | session_credential="test-session-credential", |
| 130 | csrf_token="test-csrf-token", |
| 131 | ) |
| 132 | handle = start_app_server(config) |
| 133 | client = AppHttpClient(handle.base_url, config.session_credential, config.csrf_token) |
| 134 | return handle, client |
File History
1 commit
sha256:c07f2f34a0db9f43fe866f157d1935322008545ff8940c06c7c921eb219c55ab
NXP-b DONE: independent BV-r2 pass + ISR (SD-17)
Human
minor
⚠
18 hours ago