Add triage tests, prepare-export, and UI fixes for Levkin migration.
Cherry-pick heuristics, manifest merge, global file search, scrollbar styling, and pytest/Playwright coverage unblock review and filtered export for mmetl.
This commit is contained in:
@@ -0,0 +1,138 @@
|
||||
"""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
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
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}
|
||||
Reference in New Issue
Block a user