Spaces:
Running
Running
File size: 10,763 Bytes
62e0b07 f0d5cea 62e0b07 f0d5cea 8d92c0c f0d5cea 8d92c0c f0d5cea ad552a7 8d92c0c ad552a7 125a807 62e0b07 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 | """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()
|