source/study/tests/test_space.py

"""Offline routing and input-boundary checks for the small study view."""
from contextlib import asynccontextmanager
import importlib.util
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest import mock

try:
    from starlette.applications import Starlette
    from starlette.responses import JSONResponse
    from starlette.routing import Route
    from starlette.testclient import TestClient
except ImportError:
    Starlette = None


@unittest.skipIf(Starlette is None, "Space tests require the pinned tracking environment")
class SpaceViewTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        path = Path(__file__).parents[1] / "space" / "app.py"
        spec = importlib.util.spec_from_file_location("study_space_app", path)
        cls.app_module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(cls.app_module)

    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name)
        self.transitions = []
        @asynccontextmanager
        async def lifespan(app):
            self.transitions.append("started")
            yield
            self.transitions.append("stopped")
        async def preserved(request):
            return JSONResponse({"trackio": True, "project": request.query_params.get("project")})
        backend = Starlette(routes=[Route("/", preserved), Route("/api/test", preserved),
                                    Route("/oauth/callback", preserved)], lifespan=lifespan)
        self.app = self.app_module.create_app(backend, results_dir=self.root)

    def tearDown(self):
        self.temp.cleanup()

    def test_existing_dashboard_api_oauth_and_lifespan_survive_wrapper(self):
        with TestClient(self.app) as client:
            for route in ["/?project=beam-pi-programbench-20261009", "/api/test", "/oauth/callback"]:
                response = client.get(route)
                self.assertEqual(response.status_code, 200)
                self.assertTrue(response.json()["trackio"])
            self.assertEqual(self.transitions, ["started"])
            self.assertIn("Beam + Pi · ProgramBench", client.get("/study").text)
        self.assertEqual(self.transitions, ["started", "stopped"])

    def test_missing_aggregates_have_empty_chart_and_never_fabricated_scores(self):
        plotter = mock.Mock(return_value={"data": [], "layout": {}})
        with mock.patch.dict(sys.modules, {"plot_results": SimpleNamespace(build_plotly=plotter)}):
            with TestClient(self.app) as client:
                payload = client.get("/study/data").json()
                self.assertFalse(payload["available"])
                self.assertEqual(payload["figure"]["data"], [])
                self.assertEqual(payload["downloads"], [])
                plotter.assert_called_once_with(None, [], axis="reference_seconds")

    def test_invalid_axis_unknown_download_and_raw_exception_are_not_exposed(self):
        with TestClient(self.app) as client:
            self.assertEqual(client.get("/study/data?axis=untrusted").status_code, 400)
            self.assertEqual(client.get("/study/download/budget.sqlite").status_code, 404)
            with mock.patch.object(self.app_module, "study_snapshot", side_effect=ValueError("private path and credential")):
                response = client.get("/study/data")
                self.assertEqual(response.status_code, 503)
                self.assertNotIn("private", response.text)

    def test_symlink_download_is_rejected(self):
        target = self.root / "private"
        target.write_text("credential")
        (self.root / "summary.json").symlink_to(target)
        with TestClient(self.app) as client:
            self.assertEqual(client.get("/study/download/summary.json").status_code, 404)

    @unittest.skipUnless(shutil.which("node"), "Responsive layout check requires Node.js")
    def test_minimal_chart_preserves_score_curves_at_each_width(self):
        from plot_results import build_plotly
        from study.tests.test_plot_results import fixture
        summary, rows = fixture()
        figures = [build_plotly(None, [])]
        figures.extend(build_plotly(summary, rows, axis=axis) for axis in self.app_module.AXES)
        # Run the same pure layout helper delivered to browsers, without a browser dependency.
        script = self.app_module.PAGE.split("// Responsive layout start\n", 1)[1].split(
            "// Responsive layout end", 1)[0]
        script += "\nfor(const original of " + json.dumps(figures) + "){\n"
        script += """
const before=JSON.stringify(original);
for(const width of [341,1000]){
 const view=fitFigure(original,width), result=view.figure;
 if(JSON.stringify(original)!==before) throw new Error('Published figure mutated');
 if(view.config.displayModeBar!==false) throw new Error('Download toolbar remains');
 if(result.layout.title.text || result.layout.annotations.length) throw new Error('Extra title or notes remain');
 if(result.layout.yaxis2 || result.layout.xaxis.anchor!=='y' || JSON.stringify(result.layout.yaxis.domain)!=='[0,1]') throw new Error('Coverage axis remains');
 const source=original.data.filter(trace=>trace.yaxis!=='y2');
 if(result.data.length!==source.length) throw new Error('Score curves lost');
 for(let i=0;i<source.length;i++){
  const {name:a,hovertemplate:b,...expected}=source[i];
  const {name:c,hovertemplate:d,...actual}=result.data[i];
  if(JSON.stringify(expected)!==JSON.stringify(actual)) throw new Error('Curve data changed');
  if(d.includes('tasks graded')) throw new Error('Coverage hover remains');
 }
 if(JSON.stringify(result.layout.meta)!==JSON.stringify(original.layout.meta)) throw new Error('Provenance changed');
}
}
console.log('ok');
"""
        completed = subprocess.run([shutil.which("node"), "-e", script], capture_output=True, text=True, check=True)
        self.assertEqual(completed.stdout.strip(), "ok")

    def test_recovery_view_distinguishes_final_coverage_and_exposes_audit_counts(self):
        from study.tests.test_publish_results import analysis, recovery_fixture
        config = json.loads((Path(__file__).parents[1] / "config.json").read_text())
        summary, _ = analysis.analyze(self.root, config)
        episode = summary["episodes"][0]
        first = {**episode["attempts"][0], "invalid_grade_count": 2}
        second = {**episode["attempts"][0], "attempt": 2, "invalid_grade_count": 1,
                  "has_valid_grade": True}
        episode.update(attempt_count=2, attempts=[first, second], has_valid_grade=True,
                       has_valid_final_grade=False, invalid_grade_count=3, current_attempt_invalid_grade_count=1)
        (self.root / "summary.json").write_text(json.dumps(summary))
        usage = {"api_tokens_charged": 0}
        (self.root / "usage-summary.json").write_text(json.dumps(usage))
        (self.root / "study-status.json").write_text(json.dumps({"status": "failed", "planned_episodes": 15,
             "recorded_episodes": 0, "failed_episodes": 0, "usage": usage, "runs": {}}))
        (self.root / "recovery-report.json").write_text(json.dumps(recovery_fixture()))
        with TestClient(self.app) as client:
            payload = client.get("/study/data").json()
            self.assertIn("Recovered pilot", payload["message"])
            self.assertIn("Final artifacts graded: 0/15; any valid checkpoint: 1/15", payload["message"])
            self.assertIn("2 historical + 1 current", payload["message"])
            self.assertIn("fixed official mask", payload["message"])
            self.assertEqual(payload["episodes"][0]["historical_invalid_grade_count"], 2)
            self.assertFalse(payload["episodes"][0]["has_valid_final_grade"])
            self.assertNotIn("RECOVERY.md", client.get("/study").text)
            report = client.get("/study/download/recovery-report.json")
            self.assertEqual(report.status_code, 200)
            self.assertFalse(report.json()["entries"][0]["complete"])

    def test_completed_subset_preserves_excluded_failures_and_original_accounting(self):
        from study.tests.test_publish_results import analysis
        config = json.loads((Path(__file__).parents[1] / "config.json").read_text())
        summary, _ = analysis.analyze(self.root, config)
        summary["analysis_scope"] = analysis.POST_HOC_SCOPE
        for row in summary["episodes"]:
            if row["task"] in analysis.POST_HOC_SCOPE["included_tasks"]:
                row.update(status="completed", has_valid_grade=True, has_valid_final_grade=True,
                           final_score=.5, clock_valid=True,
                           points=[dict(order=0, wall_seconds=30, reference_seconds=30,
                                        api_prefill_reference_seconds=30, score=.5)])
            else:
                row.update(status="failed", invalid_grade_count=1, current_attempt_invalid_grade_count=1,
                           clock_valid=False, api_tokens_charged=100)
        (self.root / "summary.json").write_text(json.dumps(summary))
        usage = {"api_tokens_charged": 300}
        (self.root / "usage-summary.json").write_text(json.dumps(usage))
        (self.root / "study-status.json").write_text(json.dumps({"status": "failed", "planned_episodes": 15,
             "recorded_episodes": 15, "failed_episodes": 3, "usage": usage,
             "runs": {row["run_id"]: {"status": row["status"]} for row in summary["episodes"]}}))
        with TestClient(self.app) as client:
            response = client.get("/study/data")
            self.assertEqual(response.status_code, 200)
            payload = response.json()
            self.assertEqual(payload["analysis_status"], "complete")
            self.assertIn("Included final artifacts graded: 12/12", payload["message"])
            self.assertIn("Study status: failed", payload["message"])
            self.assertIn("0 historical + 3 current", payload["message"])
            self.assertEqual(len(payload["episodes"]), 15)
            self.assertEqual(sum(row["analysis_included"] for row in payload["episodes"]), 12)
            self.assertEqual({row["status"] for row in payload["episodes"] if not row["analysis_included"]}, {"failed"})
            self.assertEqual(sum(row["api_tokens_charged"] for row in payload["episodes"]), 300)
            self.assertEqual(payload["figure"]["layout"]["meta"]["tasks_total"], 4)
            self.assertTrue(any("original 5-task evidence" in row["label"] for row in payload["downloads"]))
            self.assertEqual(client.get("/study/download/usage-summary.json").json()["api_tokens_charged"], 300)


if __name__ == "__main__":
    unittest.main()