143 lines
4.1 KiB
Python
143 lines
4.1 KiB
Python
"""Pytest fixtures — mini-export seeded from real Levkin Slack data."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import shutil
|
|
import socket
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
# Repo root on sys.path for `slack_export` / `server` (CI and local).
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
import pytest
|
|
|
|
FIXTURE_EXPORT = REPO_ROOT / "tests" / "fixtures" / "mini-export"
|
|
TEST_MANIFEST = REPO_ROOT / "tests" / "fixtures" / "test-manifest.json"
|
|
|
|
|
|
def _load_cherry_pick_module():
|
|
path = REPO_ROOT / "scripts" / "auto-cherry-pick.py"
|
|
spec = importlib.util.spec_from_file_location("auto_cherry_pick", path)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
assert spec.loader
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def cherry_pick():
|
|
return _load_cherry_pick_module()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def fixture_export_path() -> Path:
|
|
assert FIXTURE_EXPORT.is_dir(), (
|
|
f"Run: python3 scripts/build-test-fixture.py ({FIXTURE_EXPORT})"
|
|
)
|
|
return FIXTURE_EXPORT
|
|
|
|
|
|
@pytest.fixture
|
|
def export(fixture_export_path):
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
from slack_export import SlackExport
|
|
|
|
return SlackExport(fixture_export_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def empty_manifest(fixture_export_path, tmp_path):
|
|
data = {
|
|
"version": 1,
|
|
"export_path": str(fixture_export_path),
|
|
"updated_at": None,
|
|
"global_notes": "",
|
|
"channels": {},
|
|
"users": {},
|
|
}
|
|
path = tmp_path / "manifest.json"
|
|
path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
|
|
return path, data
|
|
|
|
|
|
def _free_port() -> int:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
return s.getsockname()[1]
|
|
|
|
|
|
def _wait_health(url: str, timeout: float = 8.0) -> None:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=1) as resp:
|
|
if resp.status == 200:
|
|
return
|
|
except (urllib.error.URLError, TimeoutError):
|
|
time.sleep(0.15)
|
|
raise RuntimeError(f"Server did not start: {url}")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def review_server(tmp_path_factory):
|
|
"""Start review server in-process (for coverage) with isolated manifest copy."""
|
|
import os
|
|
import threading
|
|
from http.server import ThreadingHTTPServer
|
|
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
from server import ReviewHandler
|
|
from slack_export import SlackExport
|
|
|
|
port = _free_port()
|
|
manifest = tmp_path_factory.mktemp("manifest") / "manifest.json"
|
|
shutil.copy(TEST_MANIFEST, manifest)
|
|
export = SlackExport(FIXTURE_EXPORT)
|
|
handler_cls = type(
|
|
"BoundReviewHandler",
|
|
(ReviewHandler,),
|
|
{"export": export, "manifest_path": manifest},
|
|
)
|
|
os.environ["SLACK_IMPORT_QUIET"] = "1"
|
|
httpd = ThreadingHTTPServer(("127.0.0.1", port), handler_cls)
|
|
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
|
|
thread.start()
|
|
base = f"http://127.0.0.1:{port}"
|
|
try:
|
|
_wait_health(f"{base}/api/health")
|
|
yield {"base": base, "port": port, "httpd": httpd, "manifest_path": manifest}
|
|
finally:
|
|
httpd.shutdown()
|
|
|
|
|
|
@pytest.fixture
|
|
def api(review_server):
|
|
# Reset manifest so POST tests cannot leak state across cases.
|
|
shutil.copy(TEST_MANIFEST, review_server["manifest_path"])
|
|
base = review_server["base"]
|
|
|
|
def get(path: str) -> dict:
|
|
with urllib.request.urlopen(f"{base}{path}", timeout=5) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
def post(path: str, body: dict) -> dict:
|
|
data = json.dumps(body).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
f"{base}{path}",
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
return {"get": get, "post": post, "base": base}
|