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 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)


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