Spaces:
Running
Running
| import copy | |
| import importlib.util | |
| import math | |
| from pathlib import Path | |
| import unittest | |
| spec = importlib.util.spec_from_file_location("plot_results", Path(__file__).parents[1] / "plot_results.py") | |
| plot = importlib.util.module_from_spec(spec) | |
| spec.loader.exec_module(plot) | |
| def fixture(): | |
| summary = {"episodes": [{"condition": c, "task": f"task-{i}", "has_valid_grade": True, "has_valid_final_grade": True, | |
| "clock_valid": True, "invalid_grade_count": 0, "status": "completed"} | |
| for c in plot.CONDITIONS for i in range(5)]} | |
| rows = [] | |
| for axis in plot.AXES: | |
| for condition in plot.CONDITIONS: | |
| scores = [0, .2, .7, .5] if condition == "single" else [0, .7, .8, .6] | |
| for time, score in zip([0, 30, 60, 120], scores): | |
| rows.append(dict(axis=axis, time_seconds=time, condition=condition, | |
| mean_hidden_test_fraction=score, tasks_total=5, | |
| tasks_with_valid_grade=5, episodes_configured=5, clock_valid=True)) | |
| return summary, rows | |
| class PlotTests(unittest.TestCase): | |
| def test_post_hoc_scope_recomputes_balanced_four_task_curves_without_dropping_evidence(self): | |
| from study.analyze import POST_HOC_SCOPE | |
| summary, original_rows = fixture() | |
| summary["analysis_scope"] = copy.deepcopy(POST_HOC_SCOPE) | |
| tasks = POST_HOC_SCOPE["included_tasks"] + POST_HOC_SCOPE["excluded_tasks"] | |
| summary["tasks"] = tasks | |
| for index, episode in enumerate(summary["episodes"]): | |
| episode["task"] = tasks[index % 5] | |
| excluded = index % 5 == 4 | |
| episode.update(clock_valid=not excluded, status="failed" if excluded else "completed", | |
| has_valid_final_grade=not excluded, invalid_grade_count=int(excluded), | |
| points=[dict(order=i, wall_seconds=t + (999 if excluded else 0), | |
| reference_seconds=t + (999 if excluded else 0), | |
| api_prefill_reference_seconds=t + (999 if excluded else 0), score=score) | |
| for i, (t, score) in enumerate([(30, 1 if excluded else .75), | |
| (60, 1 if excluded else .25)])]) | |
| original = copy.deepcopy((summary, original_rows)) | |
| for axis in plot.AXES: | |
| figure = plot.build_plotly(summary, original_rows, axis) | |
| meta = figure["layout"]["meta"] | |
| self.assertEqual((meta["tasks_total"], meta["included_episodes"], meta["excluded_episodes"]), (4, 12, 3)) | |
| self.assertEqual(meta["analysis_scope"], POST_HOC_SCOPE) | |
| self.assertIn("Post-hoc four-task subset", meta["status"]) | |
| self.assertIn("All 12 included final artifacts graded", meta["status"]) | |
| self.assertEqual(meta["coverage"]["single"]["observed"], 4) | |
| self.assertEqual(figure["data"][0]["y"][:2], [.75, .25]) | |
| self.assertNotIn(1029, figure["data"][0]["x"]) | |
| self.assertNotIn(1059, figure["data"][0]["x"]) | |
| self.assertEqual(figure["layout"]["yaxis2"]["tickvals"], [0, 4]) | |
| self.assertIn("all 15", " ".join(meta["notes"])) | |
| self.assertEqual((summary, original_rows), original) | |
| def test_scope_rejects_unbalanced_or_unrecognized_task_selection(self): | |
| from study.analyze import POST_HOC_SCOPE | |
| summary, rows = fixture() | |
| summary["analysis_scope"] = copy.deepcopy(POST_HOC_SCOPE) | |
| with self.assertRaises(ValueError): | |
| plot.figure_data(summary, rows) | |
| summary["analysis_scope"]["included_tasks"].pop() | |
| with self.assertRaises(ValueError): | |
| plot.figure_data(summary, rows) | |
| def test_pending_does_not_fabricate_zero_curves(self): | |
| figure = plot.build_plotly(None, []) | |
| self.assertFalse(figure["layout"]["meta"]["has_results"]) | |
| self.assertEqual(len(figure["data"]), 3) | |
| self.assertTrue(all(trace["y"] == [None] for trace in figure["data"])) | |
| self.assertIsNone(figure["layout"]["meta"]["threshold_comparison"]) | |
| def test_regression_and_zero_time_are_not_smoothed_or_shifted(self): | |
| summary, rows = fixture() | |
| data = plot.figure_data(summary, rows) | |
| self.assertEqual([r["score"] for r in data["groups"]["single"]["rows"]], [.2, .7, .5]) | |
| self.assertEqual([r["seconds"] for r in data["groups"]["single"]["rows"]], [30, 60, 120]) | |
| figure = plot.build_plotly(summary, rows) | |
| self.assertEqual(figure["data"][0]["line"]["shape"], "hv") | |
| def test_speedup_requires_complete_coverage_and_observed_crossing(self): | |
| summary, rows = fixture() | |
| comparison = plot.threshold_comparison(plot.figure_data(summary, rows)) | |
| self.assertEqual(comparison["ratio"], 2) | |
| self.assertEqual(comparison["single"], 60) | |
| annotation = next(a for a in plot.build_plotly(summary, rows)["layout"]["annotations"] if a.get("axref") == "x") | |
| self.assertEqual(annotation["x"], math.log10(60)) | |
| self.assertEqual(annotation["ax"], math.log10(30)) | |
| summary["episodes"][0]["has_valid_grade"] = False | |
| self.assertIsNone(plot.threshold_comparison(plot.figure_data(summary, rows))) | |
| def test_unreached_threshold_is_not_extrapolated(self): | |
| summary, rows = fixture() | |
| for row in rows: | |
| row["mean_hidden_test_fraction"] *= .5 | |
| self.assertIsNone(plot.threshold_comparison(plot.figure_data(summary, rows))) | |
| def test_bad_clock_suppresses_only_normalized_condition(self): | |
| summary, rows = fixture() | |
| summary["episodes"][0]["clock_valid"] = False | |
| normalized = plot.figure_data(summary, rows) | |
| wall = plot.figure_data(summary, rows, "wall_seconds") | |
| self.assertFalse(normalized["groups"]["single"]["available"]) | |
| self.assertIn("unavailable", normalized["status"]) | |
| self.assertIn("clock unavailable", plot.legend_suffix(normalized["groups"]["single"])) | |
| self.assertTrue(wall["groups"]["single"]["available"]) | |
| for row in rows: | |
| row["clock_valid"] = "False" | |
| self.assertFalse(plot.figure_data(summary, rows)["available"]) | |
| def test_missing_results_are_provisional_and_cannot_claim_speedup(self): | |
| summary, rows = fixture() | |
| summary["episodes"][0].update(status="failed", invalid_grade_count=1) | |
| data = plot.figure_data(summary, rows) | |
| self.assertIn("Provisional", data["status"]) | |
| self.assertIsNone(plot.threshold_comparison(data)) | |
| self.assertIn("zero before any valid grade", data["detail"]) | |
| def test_wrong_denominator_or_nonfinite_data_is_rejected(self): | |
| summary, rows = fixture() | |
| for key, value in [("tasks_total", 4), ("mean_hidden_test_fraction", float("nan")), | |
| ("time_seconds", -1), ("tasks_with_valid_grade", 6)]: | |
| invalid = copy.deepcopy(rows) | |
| invalid[0][key] = value | |
| with self.subTest(key=key), self.assertRaises(ValueError): | |
| plot.figure_data(summary, invalid) | |
| def test_plotly_schema_accepts_complete_and_pending_figures(self): | |
| import plotly.graph_objects as go | |
| summary, rows = fixture() | |
| for axis in plot.AXES: | |
| go.Figure(plot.build_plotly(summary, rows, axis)) | |
| go.Figure(plot.build_plotly(None, [], axis)) | |
| def test_single_positive_checkpoint_is_visible(self): | |
| summary, rows = fixture() | |
| rows = [r for r in rows if r["time_seconds"] == 30] | |
| figure = plot.build_plotly(summary, rows) | |
| self.assertEqual(figure["data"][0]["marker"]["size"], [5]) | |
| self.assertEqual(figure["data"][0]["mode"], "lines+markers") | |
| def test_earlier_valid_grade_does_not_claim_terminal_coverage(self): | |
| summary, rows = fixture() | |
| summary["episodes"][0]["has_valid_final_grade"] = False | |
| data = plot.figure_data(summary, rows) | |
| self.assertEqual(data["groups"]["single"]["observed"], 5) | |
| self.assertEqual(data["groups"]["single"]["final_observed"], 4) | |
| self.assertIn("Provisional", data["status"]) | |
| self.assertIsNone(plot.threshold_comparison(data)) | |
| def test_recovered_pilot_discloses_history_and_preserves_cumulative_clocks(self): | |
| summary, rows = fixture() | |
| summary["episodes"][0].update(attempt_count=2, invalid_grade_count=3, current_attempt_invalid_grade_count=1) | |
| summary["recovery_report"] = {"entries": [{"kind": "candidate_collection_failed"}]} | |
| data = plot.figure_data(summary, rows) | |
| self.assertTrue(data["status"].startswith("Recovered pilot")) | |
| self.assertEqual(data["historical_invalid_grades"], 2) | |
| self.assertEqual(data["current_invalid_grades"], 1) | |
| self.assertEqual([r["seconds"] for r in data["groups"]["single"]["rows"]], [30, 60, 120]) | |
| self.assertIn("original 130M", " ".join(plot.notes(data))) | |
| self.assertIn("fixed official test mask", " ".join(plot.notes(data))) | |
| figure = plot.build_plotly(summary, rows) | |
| self.assertEqual(figure["layout"]["meta"]["recovery_source"], plot.RECOVERY_SOURCE) | |
| self.assertIsNone(figure["layout"]["meta"]["threshold_comparison"]) | |
| def test_audited_artifact_correction_alone_labels_recovered_pilot(self): | |
| summary, rows = fixture() | |
| summary["episodes"][0].update(has_valid_final_grade=False, status="failed") | |
| summary["recovery_report"] = {"entries": [{"kind": "candidate_collection_failed"}]} | |
| data = plot.figure_data(summary, rows) | |
| self.assertTrue(data["status"].startswith("Recovered pilot · provisional")) | |
| self.assertFalse(data["groups"]["single"]["final_complete"]) | |
| if __name__ == "__main__": | |
| unittest.main() | |