| 1 | from __future__ import annotations
|
| 2 |
|
| 3 | import importlib.util
|
| 4 | import json
|
| 5 | import shutil
|
| 6 | import sys
|
| 7 | import tempfile
|
| 8 | import unittest
|
| 9 | from pathlib import Path
|
| 10 |
|
| 11 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 12 |
|
| 13 | import agent_metrics_schema # noqa: E402
|
| 14 |
|
| 15 |
|
| 16 | def load_agent_workflow():
|
| 17 | path = Path(__file__).resolve().parents[1] / "scripts" / "agent_workflow.py"
|
| 18 | spec = importlib.util.spec_from_file_location("agent_workflow", path)
|
| 19 | module = importlib.util.module_from_spec(spec)
|
| 20 | sys.modules["agent_workflow"] = module
|
| 21 | spec.loader.exec_module(module)
|
| 22 | return module
|
| 23 |
|
| 24 |
|
| 25 | class AgentWorkflowShimMetricsTest(unittest.TestCase):
|
| 26 | def setUp(self) -> None:
|
| 27 | self.tmp = Path(tempfile.mkdtemp(prefix="agent-workflow-shim-test-"))
|
| 28 | self.addCleanup(shutil.rmtree, self.tmp, True)
|
| 29 | self.aw = load_agent_workflow()
|
| 30 |
|
| 31 | def write_sidecar(self, *rows: dict) -> Path:
|
| 32 | sidecar = self.tmp / "vcs_shim_sidecar.jsonl"
|
| 33 | sidecar.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows))
|
| 34 | return sidecar
|
| 35 |
|
| 36 | def assert_schema_accepts_vcs_metrics(self, shim_fields: dict) -> None:
|
| 37 | base = {
|
| 38 | "commands_total": None,
|
| 39 | "status_commands": None,
|
| 40 | "diff_commands": None,
|
| 41 | "snapshot_commands": None,
|
| 42 | "branch_commands": None,
|
| 43 | "merge_commands": None,
|
| 44 | "cleanup_commands": None,
|
| 45 | "commits_created": 0,
|
| 46 | "branches_created": 0,
|
| 47 | "conflicts_total": None,
|
| 48 | "conflicts_resolved": None,
|
| 49 | "dirty_files_at_end": 0,
|
| 50 | "cleanup_success": True,
|
| 51 | "measurement_source": "test",
|
| 52 | }
|
| 53 | errors = agent_metrics_schema.validate_value(
|
| 54 | {**base, **shim_fields}, agent_metrics_schema.VCS_METRICS_SCHEMA, "$.vcs_metrics"
|
| 55 | )
|
| 56 | self.assertEqual(errors, [])
|
| 57 |
|
| 58 | def test_synthetic_sidecar_populates_blocked_share_and_polling_loop_counts(self) -> None:
|
| 59 | sidecar = self.write_sidecar(
|
| 60 | {
|
| 61 | "command": "git status --short",
|
| 62 | "start_ms": 100.0,
|
| 63 | "end_ms": 110.0,
|
| 64 | "elapsed_ms": 10.0,
|
| 65 | "returncode": 0,
|
| 66 | },
|
| 67 | {
|
| 68 | "command": "git status",
|
| 69 | "start_ms": 200.0,
|
| 70 | "end_ms": 220.0,
|
| 71 | "elapsed_ms": 20.0,
|
| 72 | "returncode": 0,
|
| 73 | },
|
| 74 | {
|
| 75 | "command": "git status --porcelain",
|
| 76 | "start_ms": 300.0,
|
| 77 | "end_ms": 330.0,
|
| 78 | "elapsed_ms": 30.0,
|
| 79 | "returncode": 0,
|
| 80 | },
|
| 81 | )
|
| 82 |
|
| 83 | fields = self.aw.shim_metrics_from_sidecar(sidecar, shim_overhead_ms=1.25, task_wall_ms=1000.0)
|
| 84 |
|
| 85 | self.assertEqual(fields["shim_sidecar_path"], str(sidecar))
|
| 86 | self.assertEqual(fields["shim_overhead"], 1.25)
|
| 87 | self.assertEqual(fields["shim_overhead_ms"], 1.25)
|
| 88 | self.assertEqual(fields["vcs_ms"], 60.0)
|
| 89 | self.assertEqual(fields["vcs_call_count_shim"], 3)
|
| 90 | self.assertEqual(fields["agent_blocked_on_vcs_ms"], 60.0)
|
| 91 | self.assertEqual(fields["vcs_share_of_task_wall"], 0.06)
|
| 92 | self.assertEqual(fields["thrash_events_count"], 1)
|
| 93 | self.assertEqual(fields["polling_loop_count"], 1)
|
| 94 | for key, value in fields.items():
|
| 95 | if key.endswith("_unavailable_reason"):
|
| 96 | self.assertIsNone(value, key)
|
| 97 | self.assert_schema_accepts_vcs_metrics(fields)
|
| 98 |
|
| 99 | def test_empty_sidecar_keeps_metrics_null_with_reasons(self) -> None:
|
| 100 | sidecar = self.write_sidecar()
|
| 101 |
|
| 102 | fields = self.aw.shim_metrics_from_sidecar(sidecar, shim_overhead_ms=None, task_wall_ms=1000.0)
|
| 103 |
|
| 104 | self.assertIsNone(fields["agent_blocked_on_vcs_ms"])
|
| 105 | self.assertEqual(fields["agent_blocked_on_vcs_ms_unavailable_reason"], "sidecar_empty")
|
| 106 | self.assertIsNone(fields["vcs_share_of_task_wall"])
|
| 107 | self.assertEqual(fields["vcs_share_of_task_wall_unavailable_reason"], "sidecar_empty")
|
| 108 | self.assertIsNone(fields["thrash_events_count"])
|
| 109 | self.assertEqual(fields["thrash_events_count_unavailable_reason"], "sidecar_empty")
|
| 110 | self.assertIsNone(fields["polling_loop_count"])
|
| 111 | self.assertEqual(fields["polling_loop_count_unavailable_reason"], "sidecar_empty")
|
| 112 | self.assertIsNone(fields["shim_overhead"])
|
| 113 | self.assertEqual(fields["shim_overhead_unavailable_reason"], "shim_overhead_unmeasured")
|
| 114 | self.assert_schema_accepts_vcs_metrics(fields)
|
| 115 |
|
| 116 | def test_malformed_sidecar_rows_keep_partial_metrics_with_specific_reasons(self) -> None:
|
| 117 | sidecar = self.write_sidecar({"command": "git status", "returncode": 0})
|
| 118 |
|
| 119 | fields = self.aw.shim_metrics_from_sidecar(sidecar, shim_overhead_ms=0.5, task_wall_ms=1000.0)
|
| 120 |
|
| 121 | self.assertEqual(fields["vcs_call_count_shim"], 1)
|
| 122 | self.assertIsNone(fields["vcs_call_count_shim_unavailable_reason"])
|
| 123 | self.assertIsNone(fields["vcs_ms"])
|
| 124 | self.assertEqual(fields["vcs_ms_unavailable_reason"], "sidecar_rows_missing_elapsed_ms")
|
| 125 | self.assertIsNone(fields["agent_blocked_on_vcs_ms"])
|
| 126 | self.assertEqual(
|
| 127 | fields["agent_blocked_on_vcs_ms_unavailable_reason"],
|
| 128 | "sidecar_rows_missing_vcs_timing_or_command",
|
| 129 | )
|
| 130 | self.assertIsNone(fields["vcs_share_of_task_wall"])
|
| 131 | self.assertEqual(
|
| 132 | fields["vcs_share_of_task_wall_unavailable_reason"],
|
| 133 | "agent_blocked_on_vcs_ms_unavailable",
|
| 134 | )
|
| 135 | self.assertIsNone(fields["thrash_events_count"])
|
| 136 | self.assertEqual(
|
| 137 | fields["thrash_events_count_unavailable_reason"],
|
| 138 | "sidecar_rows_missing_vcs_timestamp_or_command",
|
| 139 | )
|
| 140 | self.assertIsNone(fields["polling_loop_count"])
|
| 141 | self.assertEqual(
|
| 142 | fields["polling_loop_count_unavailable_reason"],
|
| 143 | "sidecar_rows_missing_vcs_timestamp_or_command",
|
| 144 | )
|
| 145 | self.assert_schema_accepts_vcs_metrics(fields)
|
| 146 |
|
| 147 | def test_disabled_shim_null_metrics_include_disabled_reason(self) -> None:
|
| 148 | fields = self.aw.shim_metrics_unavailable("vcs_shim_disabled")
|
| 149 |
|
| 150 | self.assertIsNone(fields["agent_blocked_on_vcs_ms"])
|
| 151 | self.assertIsNone(fields["polling_loop_count"])
|
| 152 | for key, value in fields.items():
|
| 153 | if key.endswith("_unavailable_reason"):
|
| 154 | self.assertEqual(value, "vcs_shim_disabled", key)
|
| 155 | self.assert_schema_accepts_vcs_metrics(fields)
|
| 156 |
|
| 157 |
|
| 158 | if __name__ == "__main__":
|
| 159 | unittest.main()
|