import json import os import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app import benchmarks from app.benchmarks import ( SCHEMA_VERSION, BenchmarkProfile, peer_agent, peer_essentials, ) from app.models import CheckResult from app.rubric import ESSENTIALS_CHECK_GROUPS, ESSENTIALS_CHECKED_MAX # --- Helpers -------------------------------------------------------------- def _all_check_names() -> set[str]: return {g.check_name for g in ESSENTIALS_CHECK_GROUPS} def _v2_profile_payload(*, peer_name: str = "Fixture Peer", score: int = 1) -> dict: """Build a minimal-but-valid v2 profile dict for tests.""" checks: dict[str, dict] = {} for g in ESSENTIALS_CHECK_GROUPS: checks[g.check_name] = { "score": score, "max": g.max_score, "state": "partial" if score > 0 else "fail", "evidence_level": "verified", } return { "schema_version": SCHEMA_VERSION, "snapshot_date": "2026-07-20", "profiles": [ { "name": peer_name, "category": "Fixture category", "group": "fixture", "size": "test", "url": "https://fixture.example", "essentials_checks": checks, "agent_passed": [], } ], } def _checks_fixture() -> list[CheckResult]: return [ CheckResult( pillar="scrapability", check_name="html_structure", label="Semantic HTML", state="pass", score=28, max_score=40, finding="Readable HTML.", fix="No action needed.", effort="low", ) ] # --- Existing fixture-based tests (rewritten for v2) --------------------- class BenchmarkFixtureTests(unittest.TestCase): def test_sample_profiles_load_under_v2(self) -> None: # Sample seeds must parse, validate, and produce a non-empty profile set. self.assertGreaterEqual(len(benchmarks._BENCHMARK_PROFILES), 12) for profile in benchmarks._BENCHMARK_PROFILES: self.assertIn("essentials_checks", profile) self.assertIn("agent_passed", profile) self.assertEqual( set(profile["essentials_checks"].keys()), _all_check_names(), ) def test_configured_profile_path_loads_fixture_without_private_data(self) -> None: payload = _v2_profile_payload() with tempfile.TemporaryDirectory() as temp_dir: profile_path = Path(temp_dir) / "benchmark_profiles.json" profile_path.write_text(json.dumps(payload), encoding="utf-8") with patch.dict(os.environ, {"MACHINEREAD_BENCHMARK_PROFILE_PATH": str(profile_path)}): with patch.object(benchmarks, "_DEFAULT_BENCHMARK_PROFILE_PATH", Path(temp_dir) / "unused.json"): loaded_profiles = benchmarks._load_benchmark_profiles() self.assertEqual(len(loaded_profiles), 1) self.assertEqual(loaded_profiles[0]["name"], "Fixture Peer") # Snapshot is immutable after load (no list / dict values to mutate) self.assertIsInstance(loaded_profiles[0]["essentials_checks"], dict) self.assertIsInstance(loaded_profiles[0]["agent_passed"], tuple) with patch.object(benchmarks, "_BENCHMARK_PROFILES", loaded_profiles): comparison = benchmarks.build_benchmark_comparison(_checks_fixture()) self.assertEqual(comparison.benchmark_count, 1) self.assertEqual(comparison.entries[0].name, "Fixture Peer") self.assertEqual(comparison.entries[0].checked_max, ESSENTIALS_CHECKED_MAX) def test_configured_missing_profile_path_fails_clearly(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: missing_profile_path = Path(temp_dir) / "missing.json" with patch.dict(os.environ, {"MACHINEREAD_BENCHMARK_PROFILE_PATH": str(missing_profile_path)}): with self.assertRaises(FileNotFoundError) as error_context: benchmarks._load_benchmark_profiles() self.assertIn("Benchmark profile file not found", str(error_context.exception)) def test_essentials_benchmark_returns_defaults_when_scope_has_no_peers(self) -> None: with ( patch.object(benchmarks, "_entries", return_value=[]), patch.object(benchmarks, "_checked_points") as checked_points, patch.object(benchmarks, "median") as benchmark_median, patch.object(benchmarks, "_percentile") as percentile, patch.object(benchmarks, "_position_label") as position_label, ): comparison = benchmarks.build_benchmark_comparison(_checks_fixture()) checked_points.assert_not_called() benchmark_median.assert_not_called() percentile.assert_not_called() position_label.assert_not_called() self.assertEqual(comparison.score, 0) self.assertEqual(comparison.checked_score, 0) self.assertEqual(comparison.checked_max, 0) self.assertEqual(comparison.benchmark_count, 0) self.assertEqual(comparison.median_score, 0) self.assertEqual(comparison.percentile, 0) self.assertEqual(comparison.position_label, "No peers available") self.assertEqual(comparison.nearest, []) self.assertEqual(comparison.entries, []) def test_agent_benchmark_returns_defaults_when_scope_has_no_peers(self) -> None: with ( patch.object(benchmarks, "_entries", return_value=[]), patch.object(benchmarks, "median") as benchmark_median, patch.object(benchmarks, "_agent_percentile") as percentile, patch.object(benchmarks, "_agent_position_label") as position_label, ): comparison = benchmarks.build_agent_benchmark_comparison( score=25, earned=2, maximum=8, ) benchmark_median.assert_not_called() percentile.assert_not_called() position_label.assert_not_called() self.assertEqual(comparison.score, 0) self.assertEqual(comparison.earned, 0) self.assertEqual(comparison.max, 0) self.assertEqual(comparison.benchmark_count, 0) self.assertEqual(comparison.median_score, 0) self.assertEqual(comparison.percentile, 0) self.assertEqual(comparison.position_label, "No peers available") self.assertEqual(comparison.nearest, []) self.assertEqual(comparison.entries, []) # --- Per-check behaviour tests ------------------------------------------- class PeerProfileBehaviorTests(unittest.TestCase): def test_peer_agent_differs_across_scopes(self) -> None: profile = { "name": "Synthetic Scope Peer", "category": "test", "group": "test", "size": "test", "url": "https://scope.example", "essentials_checks": {}, "agent_passed": [ "robots.txt published", "valid sitemap discovery", "API Catalog", "MCP Server Card", ], } base_earned, base_max = peer_agent(profile, False, False, False) protocols_earned, protocols_max = peer_agent(profile, True, False, False) full_earned, full_max = peer_agent(profile, True, True, True) self.assertEqual(base_max, 8) self.assertEqual(base_earned, 2) self.assertEqual(protocols_max, 14) self.assertEqual(protocols_earned, base_earned + 2) self.assertEqual(full_max, 21) self.assertEqual(full_earned, protocols_earned) def test_peer_essentials_excludes_unknown_evidence(self) -> None: # Build a profile where one check has evidence_level=unknown. # The same peer with all rows verified should have a strictly # larger checked_max. checks_verified: dict[str, dict] = {} checks_with_unknown: dict[str, dict] = {} for g in ESSENTIALS_CHECK_GROUPS: checks_verified[g.check_name] = { "score": g.max_score, "max": g.max_score, "state": "pass", "evidence_level": "verified", } level = "verified" if g.check_name != "wikipedia" else "unknown" checks_with_unknown[g.check_name] = { "score": 0, "max": g.max_score, "state": "warn", "evidence_level": level, } profile_verified = { "name": "Verified Peer", "category": "test", "group": "test", "size": "test", "url": "https://verified.example", "essentials_checks": checks_verified, "agent_passed": [], } profile_unknown = { "name": "Unknown Peer", "category": "test", "group": "test", "size": "test", "url": "https://unknown.example", "essentials_checks": checks_with_unknown, "agent_passed": [], } _, max_verified, _ = peer_essentials(profile_verified) _, max_unknown, _ = peer_essentials(profile_unknown) # The "wikipedia" row (4 points) is excluded from the unknown # peer's denominator; verified peer keeps the full ESSENTIALS_CHECKED_MAX. self.assertEqual(max_verified, ESSENTIALS_CHECKED_MAX) self.assertEqual(max_unknown, ESSENTIALS_CHECKED_MAX - 4) def test_peer_agent_ignores_unknown_probe_labels(self) -> None: # Profile's agent_passed includes a label that no longer exists in # the current scope probe list. The peer_agent helper must drop it # silently with no error and no contribution. profile = { "name": "Stale Peer", "category": "test", "group": "test", "size": "test", "url": "https://stale.example", "essentials_checks": {}, "agent_passed": [ "this probe no longer exists", "robots.txt published", # one real label ], } # Even though agent_passed has 2 entries, only 1 is a real probe. earned, maximum = peer_agent(profile, False, False, False) self.assertEqual(earned, 1) self.assertEqual(maximum, 8) # base scope def test_peer_agent_probe_outside_scope_contributes_zero(self) -> None: # Peer passes "API Catalog" (a protocol-scope probe) but the user # is at base scope → that probe is not in the scope's probe list # and contributes 0 to earned. profile = { "name": "Protocol Peer", "category": "test", "group": "test", "size": "test", "url": "https://protocol.example", "essentials_checks": {}, "agent_passed": ["API Catalog", "MCP Server Card", "A2A Agent Card"], } # At base scope, no protocol probes are included earned, maximum = peer_agent(profile, False, False, False) self.assertEqual(earned, 0) # At full scope, those protocol passes count earned_full, maximum_full = peer_agent(profile, True, False, False) self.assertGreater(earned_full, 0) self.assertGreater(maximum_full, maximum) # --- Validation tests (eager, at load) ----------------------------------- class BenchmarkValidationTests(unittest.TestCase): def _write_profile(self, profiles: list, snapshot_date: str = "2026-07-20") -> Path: payload = { "schema_version": SCHEMA_VERSION, "snapshot_date": snapshot_date, "profiles": profiles, } tmp = Path(tempfile.mkstemp(suffix=".json")[1]) tmp.write_text(json.dumps(payload), encoding="utf-8") return tmp def test_v1_bare_list_raises_migration_error(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: v1_path = Path(temp_dir) / "v1.json" v1_path.write_text(json.dumps([{"name": "X"}]), encoding="utf-8") with patch.dict(os.environ, {"MACHINEREAD_BENCHMARK_PROFILE_PATH": str(v1_path)}): with self.assertRaises(ValueError) as ctx: benchmarks._load_benchmark_profiles() self.assertIn("refresh_benchmarks.py", str(ctx.exception)) def test_unknown_check_key_raises_at_load(self) -> None: base = _v2_profile_payload()["profiles"][0] base["essentials_checks"]["made_up_check"] = { "score": 0, "max": 1, "state": "fail", "evidence_level": "verified", } path = self._write_profile([base]) with patch.dict(os.environ, {"MACHINEREAD_BENCHMARK_PROFILE_PATH": str(path)}): with self.assertRaises(ValueError) as ctx: benchmarks._load_benchmark_profiles() self.assertIn("made_up_check", str(ctx.exception)) def test_score_exceeds_max_raises_at_load(self) -> None: base = _v2_profile_payload()["profiles"][0] first_check = next(iter(base["essentials_checks"])) base["essentials_checks"][first_check]["score"] = ( base["essentials_checks"][first_check]["max"] + 1 ) path = self._write_profile([base]) with patch.dict(os.environ, {"MACHINEREAD_BENCHMARK_PROFILE_PATH": str(path)}): with self.assertRaises(ValueError): benchmarks._load_benchmark_profiles() def test_bad_state_value_raises_at_load(self) -> None: base = _v2_profile_payload()["profiles"][0] first_check = next(iter(base["essentials_checks"])) base["essentials_checks"][first_check]["state"] = "nope" path = self._write_profile([base]) with patch.dict(os.environ, {"MACHINEREAD_BENCHMARK_PROFILE_PATH": str(path)}): with self.assertRaises(ValueError) as ctx: benchmarks._load_benchmark_profiles() self.assertIn("nope", str(ctx.exception)) def test_bad_evidence_level_raises_at_load(self) -> None: base = _v2_profile_payload()["profiles"][0] first_check = next(iter(base["essentials_checks"])) base["essentials_checks"][first_check]["evidence_level"] = "maybe" path = self._write_profile([base]) with patch.dict(os.environ, {"MACHINEREAD_BENCHMARK_PROFILE_PATH": str(path)}): with self.assertRaises(ValueError) as ctx: benchmarks._load_benchmark_profiles() self.assertIn("maybe", str(ctx.exception)) # --- Refresh-script round-trip ------------------------------------------- class RefreshScriptRoundTripTests(unittest.TestCase): def test_refresh_script_round_trips_through_loader(self) -> None: # Build a fake profile dict the refresh script would emit. payload = _v2_profile_payload(peer_name="Round Trip Peer", score=2) with tempfile.TemporaryDirectory() as temp_dir: peers_path = Path(temp_dir) / "peers.json" peers_path.write_text( json.dumps([{"name": "Round Trip Peer", "category": "x", "group": "y", "size": "z", "url": "https://rt.example"}]), encoding="utf-8", ) out_path = Path(temp_dir) / "out.json" import importlib import scripts.refresh_benchmarks as refresh # type: ignore # Patch _audit_one to return a v2 profile without HTTP. async def fake_audit_one(peer): return payload["profiles"][0] with patch.object(refresh, "_audit_one", side_effect=fake_audit_one): rc = refresh.main() if False else None # main() reads argv; instead call _run with explicit args. import argparse args = argparse.Namespace( peers=peers_path, out=out_path, concurrency=1 ) rc = asyncio_run(refresh._run(args)) self.assertEqual(rc, 0) self.assertTrue(out_path.exists()) on_disk = json.loads(out_path.read_text(encoding="utf-8")) self.assertEqual(on_disk["schema_version"], SCHEMA_VERSION) self.assertEqual(len(on_disk["profiles"]), 1) self.assertEqual(on_disk["profiles"][0]["name"], "Round Trip Peer") # Now round-trip through the loader with patch.dict(os.environ, {"MACHINEREAD_BENCHMARK_PROFILE_PATH": str(out_path)}): with patch.object(benchmarks, "_DEFAULT_BENCHMARK_PROFILE_PATH", Path(temp_dir) / "unused.json"): loaded = benchmarks._load_benchmark_profiles() self.assertEqual(len(loaded), 1) self.assertEqual(loaded[0]["name"], "Round Trip Peer") def asyncio_run(coro): import asyncio return asyncio.run(coro) # --- Exclusion rule tests (kept from v1) --------------------------------- class CheckedPointsExclusionTests(unittest.TestCase): def test_warn_unknown_evidence_rows_excluded_from_denominator(self) -> None: # Two real scored rows + one warn-state fallback. The fallback carries # evidence_level='unknown' (per _fallback_check_result) and contributes # 0/group_max to the denominator; transient fetch failures must not # silently drag the Evidence score down. The denominator should reflect # only the rows the audit could actually evaluate. checks = [ CheckResult( pillar="scrapability", check_name="html_structure", label="Semantic HTML", state="pass", score=8, max_score=10, finding="Readable HTML.", fix="No action needed.", effort="low", ), CheckResult( pillar="seo", check_name="sitemap", label="XML sitemap", state="partial", score=4, max_score=8, finding="Sitemap present.", fix="No action needed.", effort="low", ), CheckResult( pillar="off_site", check_name="wikipedia", label="Wikipedia entity", state="warn", evidence_level="unknown", score=0, max_score=6, finding=( "MachineRead could not complete this check during the audit, " "so this row is inconclusive." ), fix="Retry the audit. If this warning repeats, review server logs.", effort="medium", ), ] score, maximum = benchmarks._checked_points(checks) self.assertEqual(score, 12) # 10 + 8 = 18. The warn row's 6 points must NOT be in the denominator. self.assertEqual(maximum, 18) def test_warn_unknown_rows_excluded_from_free_evidence_score(self) -> None: # The end-to-end consequence: free_evidence_score is calculated purely # over the rows the audit actually evaluated. Same fixture as above. checks = [ CheckResult( pillar="scrapability", check_name="html_structure", label="Semantic HTML", state="pass", score=8, max_score=10, finding="Readable HTML.", fix="No action needed.", effort="low", ), CheckResult( pillar="seo", check_name="sitemap", label="XML sitemap", state="partial", score=4, max_score=8, finding="Sitemap present.", fix="No action needed.", effort="low", ), CheckResult( pillar="off_site", check_name="wikipedia", label="Wikipedia entity", state="warn", evidence_level="unknown", score=0, max_score=6, finding=( "MachineRead could not complete this check during the audit, " "so this row is inconclusive." ), fix="Retry the audit. If this warning repeats, review server logs.", effort="medium", ), ] score = benchmarks.free_evidence_score(checks) # 12/18 = 66.67 -> 67. With the warn row kept (12/24 = 50), the score # would silently misrepresent the audit. After the fix, denominator # excludes warn rows so the score reflects what was actually evaluated. self.assertEqual(score, round((12 / 18) * 100)) def test_locked_rows_still_excluded_from_denominator(self) -> None: # Regression guard: the existing locked-row exclusion must still hold # alongside the new warn-row exclusion. checks = [ CheckResult( pillar="scrapability", check_name="html_structure", label="Semantic HTML", state="pass", score=5, max_score=10, finding="Readable HTML.", fix="No action needed.", effort="low", ), CheckResult( pillar="off_site", check_name="paid_reserved", label="Paid-only signal", state="locked", evidence_level="not_applicable", score=0, max_score=12, finding="Reserved for paid tier.", fix="Upgrade to evaluate.", effort="low", ), ] score, maximum = benchmarks._checked_points(checks) self.assertEqual(score, 5) self.assertEqual(maximum, 10) if __name__ == "__main__": unittest.main()