225 lines
8.7 KiB
Python
225 lines
8.7 KiB
Python
"""dashboard-test-runner US-001: opt-in, admin-only tracker test-runner API.
|
|
|
|
Runs against a tiny throwaway repo (tmp_path) so collection and runs stay
|
|
fast and hermetic — the manager itself discovers the real repo root when
|
|
constructed without an explicit ``repo_root``.
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from meshnet_tracker.accounts import AccountStore
|
|
from meshnet_tracker.server import TrackerServer
|
|
from meshnet_tracker.test_runner import TestRunManager as RunManager
|
|
from meshnet_tracker.test_runner import discover_repo_root
|
|
|
|
|
|
def _make_repo(tmp_path: Path) -> Path:
|
|
"""A minimal repo satisfying discover_repo_root's markers, with fast tests."""
|
|
(tmp_path / "pyproject.toml").write_text("[tool.pytest.ini_options]\n")
|
|
tests = tmp_path / "tests"
|
|
tests.mkdir()
|
|
(tests / "conftest.py").write_text("")
|
|
(tests / "test_quick.py").write_text(
|
|
"def test_passes():\n assert True\n\n"
|
|
"def test_also_passes():\n assert 1 + 1 == 2\n"
|
|
)
|
|
(tests / "test_slow.py").write_text(
|
|
"import time\n\ndef test_sleeps():\n time.sleep(30)\n"
|
|
)
|
|
(tests / "test_smoke.py").write_text("def test_smoke():\n assert True\n")
|
|
# Real-inference stand-in: must never be collected or run by default.
|
|
(tests / "test_real_stub.py").write_text(
|
|
"def test_real_inference():\n raise AssertionError('env-gated')\n"
|
|
)
|
|
return tmp_path
|
|
|
|
|
|
def _start_tracker(tmp_path: Path | None):
|
|
"""Tracker + (admin_session, user_session); test runner enabled iff tmp_path given."""
|
|
accounts = AccountStore()
|
|
admin = accounts.register(email="admin@example.com", password="admin-pass-123")
|
|
admin_session = accounts.create_session(admin["account_id"])
|
|
user = accounts.register(email="user@example.com", password="user-pass-123")
|
|
user_session = accounts.create_session(user["account_id"])
|
|
runner = RunManager(repo_root=_make_repo(tmp_path)) if tmp_path is not None else None
|
|
tracker = TrackerServer(accounts=accounts, test_runner=runner)
|
|
port = tracker.start()
|
|
return tracker, port, admin_session, user_session
|
|
|
|
|
|
def _request(port, method, path, token=None, payload=None):
|
|
body = json.dumps(payload).encode() if payload is not None else None
|
|
headers = {"Content-Type": "application/json"}
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
req = urllib.request.Request(
|
|
f"http://127.0.0.1:{port}{path}", data=body, headers=headers, method=method
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
return resp.status, json.loads(resp.read())
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.code, json.loads(exc.read())
|
|
|
|
|
|
def _wait_for_completion(port, session, timeout=90.0):
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
status, data = _request(port, "GET", "/v1/tests/status", token=session)
|
|
assert status == 200
|
|
if data["run"] is not None and data["run"]["status"] != "running":
|
|
return data
|
|
time.sleep(0.2)
|
|
raise AssertionError("test run did not finish in time")
|
|
|
|
|
|
def test_endpoints_require_admin(tmp_path):
|
|
tracker, port, _admin, user = _start_tracker(tmp_path)
|
|
try:
|
|
assert _request(port, "GET", "/v1/tests")[0] == 401
|
|
assert _request(port, "GET", "/v1/tests/status")[0] == 401
|
|
assert _request(port, "POST", "/v1/tests/run", payload={"target": "x"})[0] == 401
|
|
assert _request(port, "GET", "/v1/tests", token=user)[0] == 403
|
|
assert _request(port, "GET", "/v1/tests/status", token=user)[0] == 403
|
|
assert _request(
|
|
port, "POST", "/v1/tests/run", token=user, payload={"target": "x"}
|
|
)[0] == 403
|
|
finally:
|
|
tracker.stop()
|
|
|
|
|
|
def test_disabled_by_default_even_for_admin(monkeypatch):
|
|
monkeypatch.delenv("MESHNET_ENABLE_TEST_RUNNER", raising=False)
|
|
tracker, port, admin, _user = _start_tracker(None)
|
|
try:
|
|
assert tracker._test_runner is None # no flag, no env → fails closed
|
|
for method, path, payload in [
|
|
("GET", "/v1/tests", None),
|
|
("GET", "/v1/tests/status", None),
|
|
("POST", "/v1/tests/run", {"target": "tests/test_smoke.py::test_smoke"}),
|
|
]:
|
|
status, data = _request(port, method, path, token=admin, payload=payload)
|
|
assert status == 403
|
|
assert "disabled" in data["error"]
|
|
finally:
|
|
tracker.stop()
|
|
|
|
|
|
def test_enable_flag_constructs_runner_against_real_repo(monkeypatch):
|
|
monkeypatch.delenv("MESHNET_ENABLE_TEST_RUNNER", raising=False)
|
|
tracker = TrackerServer(enable_test_runner=True)
|
|
assert tracker._test_runner is not None
|
|
assert tracker._test_runner.repo_root == discover_repo_root()
|
|
assert (tracker._test_runner.repo_root / "tests" / "conftest.py").is_file()
|
|
|
|
|
|
def test_collection_lists_tests_and_excludes_real_inference(tmp_path, monkeypatch):
|
|
monkeypatch.delenv("MESHNET_ENABLE_REAL_INFERENCE_TESTS", raising=False)
|
|
tracker, port, admin, _user = _start_tracker(tmp_path)
|
|
try:
|
|
status, data = _request(port, "GET", "/v1/tests", token=admin)
|
|
assert status == 200
|
|
assert data["enabled"] is True
|
|
assert "tests/test_quick.py::test_passes" in data["tests"]
|
|
assert "tests/test_slow.py::test_sleeps" in data["tests"]
|
|
assert not any("test_real_" in node_id for node_id in data["tests"])
|
|
assert data["collected_at"] is not None
|
|
suite_ids = [suite["id"] for suite in data["suites"]]
|
|
assert "suite:smoke" in suite_ids
|
|
assert "suite:real-inference" not in suite_ids
|
|
finally:
|
|
tracker.stop()
|
|
|
|
|
|
def test_run_selected_test_to_completion(tmp_path):
|
|
tracker, port, admin, _user = _start_tracker(tmp_path)
|
|
try:
|
|
assert _request(port, "GET", "/v1/tests", token=admin)[0] == 200
|
|
status, started = _request(
|
|
port, "POST", "/v1/tests/run", token=admin,
|
|
payload={"target": "tests/test_quick.py::test_passes"},
|
|
)
|
|
assert status == 202
|
|
assert started["run"]["status"] == "running"
|
|
assert started["run"]["run_id"]
|
|
assert started["run"]["started_at"] is not None
|
|
|
|
finished = _wait_for_completion(port, admin)
|
|
run = finished["run"]
|
|
assert run["status"] == "passed"
|
|
assert run["exit_code"] == 0
|
|
assert run["target"] == "tests/test_quick.py::test_passes"
|
|
assert run["finished_at"] >= run["started_at"]
|
|
assert run["elapsed_seconds"] >= 0
|
|
assert "1 passed" in finished["stdout"]
|
|
assert finished["log_lines_limit"] > 0
|
|
finally:
|
|
tracker.stop()
|
|
|
|
|
|
def test_run_approved_suite_without_prior_collection(tmp_path):
|
|
tracker, port, admin, _user = _start_tracker(tmp_path)
|
|
try:
|
|
status, started = _request(
|
|
port, "POST", "/v1/tests/run", token=admin,
|
|
payload={"target": "suite:smoke"},
|
|
)
|
|
assert status == 202
|
|
assert started["run"]["args"] == ["tests/test_smoke.py"]
|
|
finished = _wait_for_completion(port, admin)
|
|
assert finished["run"]["status"] == "passed"
|
|
assert finished["run"]["exit_code"] == 0
|
|
finally:
|
|
tracker.stop()
|
|
|
|
|
|
@pytest.mark.parametrize("target", [
|
|
"",
|
|
"-x",
|
|
"--help",
|
|
"tests/test_quick.py::test_passes; rm -rf /",
|
|
"tests/never_collected.py::test_missing",
|
|
"suite:real-inference",
|
|
"suite:no-such-suite",
|
|
])
|
|
def test_rejects_arbitrary_or_uncollected_targets(tmp_path, target, monkeypatch):
|
|
monkeypatch.delenv("MESHNET_ENABLE_REAL_INFERENCE_TESTS", raising=False)
|
|
tracker, port, admin, _user = _start_tracker(tmp_path)
|
|
try:
|
|
assert _request(port, "GET", "/v1/tests", token=admin)[0] == 200
|
|
status, data = _request(
|
|
port, "POST", "/v1/tests/run", token=admin, payload={"target": target}
|
|
)
|
|
assert status == 400
|
|
assert "error" in data
|
|
# Nothing started.
|
|
assert _request(port, "GET", "/v1/tests/status", token=admin)[1]["run"] is None
|
|
finally:
|
|
tracker.stop()
|
|
|
|
|
|
def test_rejects_concurrent_runs(tmp_path):
|
|
tracker, port, admin, _user = _start_tracker(tmp_path)
|
|
try:
|
|
assert _request(port, "GET", "/v1/tests", token=admin)[0] == 200
|
|
status, _ = _request(
|
|
port, "POST", "/v1/tests/run", token=admin,
|
|
payload={"target": "tests/test_slow.py::test_sleeps"},
|
|
)
|
|
assert status == 202
|
|
status, data = _request(
|
|
port, "POST", "/v1/tests/run", token=admin,
|
|
payload={"target": "tests/test_quick.py::test_passes"},
|
|
)
|
|
assert status == 409
|
|
assert "in progress" in data["error"]
|
|
finally:
|
|
# stop() kills the sleeping pytest child via TestRunManager.shutdown().
|
|
tracker.stop()
|