import base64 import importlib.util import json import sys import tempfile import unittest import urllib.error from argparse import Namespace from unittest.mock import patch from pathlib import Path SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "manus_op_grafana.py" SPEC = importlib.util.spec_from_file_location("manus_op_grafana", SCRIPT) MODULE = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = MODULE SPEC.loader.exec_module(MODULE) def der(tag, value): if len(value) < 128: length = bytes([len(value)]) else: raw = len(value).to_bytes((len(value).bit_length() + 7) // 8, "big") length = bytes([0x80 | len(raw)]) + raw return bytes([tag]) + length + value def integer(value): raw = value.to_bytes((value.bit_length() + 7) // 8, "big") if raw[0] & 0x80: raw = b"\x00" + raw return der(0x02, raw) class ContractTests(unittest.TestCase): def test_batch_executes_twelve_semantic_queries_strictly_serially(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) context = root / "context.json" plan = root / "plan.json" result = root / "result.json" window = "historical:absolute;start=100;end=300;timezone=UTC;source=test" context.write_text(json.dumps({ "schemaVersion": "auto-oncall-op-query-context-v1", "provider": "manus", "scenarioId": "query-execution", "timeWindow": window, "target": "vida", "auth": {"status": "not_required"}, }), encoding="utf-8") plan.write_text(json.dumps({ "schemaVersion": "auto-oncall-op-query-batch-plan-v1", "provider": "manus", "scenarioId": "query-execution", "timeWindow": window, "target": "vida", "queries": [{ "queryId": f"metric-check-{index}", "operation": "metrics", "params": { "name": "qps", "query": "sum(rate(tidb_server_query_total[2m]))", }, } for index in range(1, 13)], }), encoding="utf-8") calls = [] def execute(argv): calls.append(argv) return 0, {"summary": {"point_count": 0 if len(calls) == 12 else 2}} args = Namespace(context=context, plan=plan, result=result) with patch.object(MODULE, "_run_manus_subcommand", side_effect=execute): output = MODULE.cmd_batch(args) self.assertEqual([call[0] for call in calls], ["metrics"] * 12) self.assertEqual(output["completion"], "complete") self.assertEqual(len(output["queries"]), 12) self.assertEqual(output["queries"][-1]["queryId"], "metric-check-12") self.assertEqual(output["queries"][-1]["status"], "empty") self.assertEqual(json.loads(result.read_text(encoding="utf-8"))["completion"], "complete") def test_batch_rejects_thirteen_queries(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) context = root / "context.json" plan = root / "plan.json" result = root / "result.json" window = "historical:absolute;start=100;end=300;timezone=UTC;source=test" context.write_text(json.dumps({ "schemaVersion": "auto-oncall-op-query-context-v1", "provider": "manus", "scenarioId": "query-execution", "timeWindow": window, "target": "vida", "auth": {"status": "not_required"}, }), encoding="utf-8") plan.write_text(json.dumps({ "schemaVersion": "auto-oncall-op-query-batch-plan-v1", "provider": "manus", "scenarioId": "query-execution", "timeWindow": window, "target": "vida", "queries": [{ "queryId": f"metric-check-{index}", "operation": "metrics", "params": {"name": "qps", "query": "up"}, } for index in range(1, 14)], }), encoding="utf-8") with self.assertRaisesRegex(MODULE.QueryError, "one to 12 queries"): MODULE.cmd_batch(Namespace(context=context, plan=plan, result=result)) def test_batch_rejects_a_query_outside_the_runner_window(self): with self.assertRaisesRegex(MODULE.QueryError, "runner-owned historical window"): MODULE._bounded_query_range({"start": 99, "end": 200}, (100, 300)) def test_batch_requires_seven_separate_daily_comparison_ranges(self): incident = (2_000_000, 2_000_600) context = { "scenarioId": "workload-hotspot", "timeWindow": ( f"live:absolute;start={incident[0]};end={incident[1]};" "timezone=UTC;source=test" ), "comparisonWindows": [{ "windowId": f"day-{index}", "start": incident[0] - index * 86400, "end": incident[1] - index * 86400, } for index in range(1, 8)], } windows = MODULE._comparison_bounds(context) queries = [{ "queryId": f"day-{index}-cpu", "operation": "metrics", "params": { "name": "cpu", "query": "sum by (instance) (rate(process_cpu_seconds_total[1m]))", "start": start, "end": end, }, } for index, (start, end) in enumerate(windows, start=1)] MODULE._assert_comparison_plan(context, queries, incident, windows) with self.assertRaisesRegex(MODULE.QueryError, "seven workload-hotspot metric queries"): MODULE._assert_comparison_plan(context, [{ "queryId": "combined-history", "operation": "metrics", "params": { "name": "cpu", "query": "up", "start": windows[-1][0], "end": windows[0][1], }, }], incident, windows) def test_batch_normalizes_flat_agent_query_fields_and_metadata(self): query = MODULE._normalize_manus_batch_query({ "queryId": "alignment", "description": "aligned metric", "operation": "metrics", "name": "qps", "query": "sum(rate(tidb_server_query_total[2m]))", "step": 30, "start": 100, "end": 300, "subDeadlineMs": 120000, "futureMetadata": "ignored", }) argv = MODULE._manus_batch_argv( "vida", query, (100, 300), Path("/tmp/evidence") ) self.assertIn("30", argv) self.assertNotIn("futureMetadata", argv) self.assertEqual(MODULE._normalize_manus_batch_target({"target": "monica"}), "monica") self.assertEqual(query["emptyResultExpectation"], "unknown") expected = MODULE._normalize_manus_batch_query({ "queryId": "baseline-cpu", "operation": "metrics", "emptyResultExpectation": "expected_non_empty", "params": {"name": "cpu", "query": "up"}, }) self.assertEqual(expected["emptyResultExpectation"], "expected_non_empty") self.assertNotIn("emptyResultExpectation", expected["params"]) with self.assertRaises(MODULE.QueryError): MODULE._normalize_manus_batch_query({ "queryId": "invalid-empty-policy", "operation": "metrics", "emptyResultExpectation": "always-normal", "params": {"name": "cpu", "query": "up"}, }) slow_query = MODULE._normalize_manus_batch_query({ "queryId": "slow-query-details", "operation": "dashboard", "handoffFields": [ "stats", "cop_wait_max", "cop_wait_addr", "memory_max", ], "params": {"kind": "slow-query", "start": 100, "end": 300}, }) self.assertEqual(slow_query["handoffFields"], [ "stats", "cop_wait_max", "cop_wait_addr", "memory_max", ]) selected_fields = slow_query["params"]["fields"].split(",") self.assertIn("stats", selected_fields) self.assertIn("cop_wait_max", selected_fields) self.assertIn("cop_wait_addr", selected_fields) with self.assertRaises(MODULE.QueryError): MODULE._normalize_manus_batch_query({ "queryId": "slow-query-invalid", "operation": "dashboard", "handoffFields": ["raw_query_text"], "params": {"kind": "slow-query", "start": 100, "end": 300}, }) def test_duration_and_range_contract(self): self.assertEqual(MODULE.parse_duration("15m"), 900) self.assertEqual(MODULE.parse_duration("1h"), 3600) with self.assertRaises(MODULE.QueryError): MODULE.parse_duration("0m") def test_exact_target_has_no_credentials_or_fallback(self): vida = MODULE.load_target("vida") monica = MODULE.load_target("monica") self.assertEqual(vida["grafana_datasource_name"], "vida") self.assertEqual(monica["prometheus_url"], "https://tidb-prometheus-monica.prod.ops.butterfly-effect.dev") self.assertEqual(MODULE.dashboard_credential_names(vida), ( "MANUS_DASHBOARD_USERNAME", "MANUS_DASHBOARD_PASSWORD" )) self.assertEqual(MODULE.dashboard_credential_names(monica), ( "MANUS_MONICA_DASHBOARD_USERNAME", "MANUS_MONICA_DASHBOARD_PASSWORD" )) text = json.dumps({"vida": vida, "monica": monica}).lower() self.assertNotIn("password", text) self.assertNotIn("clinic", text) with self.assertRaises(MODULE.QueryError): MODULE.load_target("unknown") def test_every_command_requires_an_explicit_supported_target(self): parser = MODULE.build_parser() with patch("sys.stderr"): with self.assertRaises(SystemExit): parser.parse_args(["health"]) self.assertEqual(parser.parse_args(["health", "--target", "monica"]).target, "monica") def test_credentials_are_read_only_from_environment(self): with patch.dict(MODULE.os.environ, {}, clear=True): with self.assertRaisesRegex(MODULE.QueryError, "MANUS_GRAFANA_USERNAME"): MODULE.require_env("MANUS_GRAFANA_USERNAME", "MANUS_GRAFANA_PASSWORD") with patch.dict(MODULE.os.environ, { "MANUS_GRAFANA_USERNAME": "user", "MANUS_GRAFANA_PASSWORD": "secret" }, clear=True): self.assertEqual( MODULE.require_env("MANUS_GRAFANA_USERNAME", "MANUS_GRAFANA_PASSWORD"), ["user", "secret"] ) def test_metric_summary_skips_non_numeric_points(self): payload = {"data": {"result": [{"values": [[1, "2"], [2, "NaN"], [3, "4"]]}]}} self.assertEqual(MODULE.metric_summary(payload), { "series_count": 1, "point_count": 2, "min": 2.0, "max": 4.0, "avg": 3.0, "current": 4.0, }) def test_empty_504_is_classified_as_timeout(self): with self.assertRaisesRegex(MODULE.QueryError, r"timed out \(HTTP 504\)"): MODULE.expect_ok(MODULE.Response(504, b""), "Grafana query_range") def test_direct_transport_normalizes_socket_timeout(self): transport = MODULE.DirectTransport() with patch.object(MODULE.urllib.request, "urlopen", side_effect=TimeoutError("timed out")): with self.assertRaisesRegex(MODULE.QueryError, "network request timed out"): transport.request("GET", "https://example.test") def test_direct_transport_normalizes_url_timeout(self): transport = MODULE.DirectTransport() error = urllib.error.URLError(TimeoutError("timed out")) with patch.object(MODULE.urllib.request, "urlopen", side_effect=error): with self.assertRaisesRegex(MODULE.QueryError, "network request timed out"): transport.request("GET", "https://example.test") def test_metrics_timeout_retries_once_with_centered_half_window(self): class Client: calls = [] def query_range(self, query, start, end, step): self.calls.append((query, start, end, step)) if len(self.calls) == 1: raise MODULE.QueryError("Grafana query_range timed out (HTTP 504)") return {"id": 1, "name": "vida", "type": "prometheus"}, { "status": "success", "data": {"result": [{"values": [[1, "7"]]}]} } args = Namespace( target="vida", start=100, end=300, last="20m", query="up{instance=\"tikv-1\"}", step="30", name="up", unit="", output_dir=None, ) client = Client() with patch.object(MODULE, "load_target", return_value={"_name": "vida", "grafana_url": "https://example.test"}), \ patch.object(MODULE, "make_transport"), \ patch.object(MODULE, "GrafanaClient", return_value=client), \ patch("sys.stderr"): report = MODULE.cmd_metrics(args) self.assertEqual(client.calls, [ (args.query, 100, 300, "30"), (args.query, 150, 250, "30"), ]) self.assertEqual(report["time_window"], {"start": 150, "end": 250}) self.assertEqual(report["recovery"], { "attempts": 2, "window_narrowed": True, "initial_time_window": {"start": 100, "end": 300}, }) def test_metrics_non_timeout_does_not_retry(self): class Client: calls = 0 def query_range(self, query, start, end, step): self.calls += 1 raise MODULE.QueryError("Grafana query_range failed with HTTP 403") args = Namespace( target="vida", start=100, end=300, last="20m", query="up", step="30", name="up", unit="", output_dir=None, ) client = Client() with patch.object(MODULE, "load_target", return_value={"_name": "vida", "grafana_url": "https://example.test"}), \ patch.object(MODULE, "make_transport"), \ patch.object(MODULE, "GrafanaClient", return_value=client): with self.assertRaisesRegex(MODULE.QueryError, "HTTP 403"): MODULE.cmd_metrics(args) self.assertEqual(client.calls, 1) def test_metrics_parser_defaults_to_bounded_window(self): args = MODULE.build_parser().parse_args([ "metrics", "--target", "vida", "--name", "up", "--query", "up" ]) self.assertEqual(args.last, "20m") def test_direct_prometheus_query_is_bound_to_monica(self): class Transport: requests = [] def request(self, method, url, *, params=None, headers=None, body=None, timeout=60): self.requests.append({"method": method, "url": url, "params": params, "headers": headers}) return MODULE.Response(200, json.dumps({ "status": "success", "data": {"resultType": "matrix", "result": []} }).encode()) transport = Transport() client = MODULE.PrometheusClient(transport, MODULE.load_target("monica")) source, payload = client.query_range("up", 1, 2, "15") self.assertEqual(source, {"name": "monica", "type": "prometheus", "access": "direct"}) self.assertEqual(payload["status"], "success") self.assertEqual( transport.requests[0]["url"], "https://tidb-prometheus-monica.prod.ops.butterfly-effect.dev/api/v1/query_range", ) self.assertNotIn("Authorization", transport.requests[0]["headers"]) def test_topsql_nested_summary_and_chart_data_compaction(self): payload = {"summary": {"data": [{"sql_digest": "x", "is_other": False, "chart_data": [1, 2]}]}} rows = MODULE.dashboard_rows("topsql", payload) self.assertEqual(len(rows), 1) self.assertEqual(MODULE.compact_rows(rows, 5), [{"sql_digest": "x", "is_other": False}]) def test_topsql_timeout_retries_once(self): class Client: calls = 0 def get(self, path, params=None, timeout=60): self.calls += 1 if self.calls == 1: raise MODULE.QueryError("TopSQL summary timed out (HTTP 504)") return {"data": []} args = Namespace( kind="topsql", instance="127.0.0.1:10080", instance_type="tidb", group_by="query", limit=10, window="10s" ) original_sleep = MODULE.time.sleep MODULE.time.sleep = lambda _seconds: None try: payload, _request = MODULE.query_dashboard(Client(), args, 1, 2) finally: MODULE.time.sleep = original_sleep self.assertEqual(payload["summary"], {"data": []}) def test_topsql_requires_metric_selected_instance(self): class Client: calls = 0 def get(self, path, params=None, timeout=60): self.calls += 1 return {"data": []} client = Client() args = Namespace( kind="topsql", instance=None, instance_type="tidb", group_by="query", limit=10, window="10s" ) with self.assertRaisesRegex(MODULE.QueryError, "requires --instance"): MODULE.query_dashboard(client, args, 1, 2) self.assertEqual(client.calls, 0) def test_dashboard_parser_defaults_to_top_ten(self): args = MODULE.build_parser().parse_args([ "dashboard", "--target", "vida", "--kind", "topsql", "--instance", "127.0.0.1:10080" ]) self.assertEqual(args.limit, 10) self.assertEqual(args.preview_limit, 10) def test_slow_query_selects_supported_diagnostic_fields(self): class Client: request = None def get_optional(self, path, params=None, timeout=60): self.assert_path = path return ["query", "cop_time", "process_keys", "total_keys"] def get(self, path, params=None, timeout=60): self.request = {"path": path, "params": params, "timeout": timeout} return [{"digest": "d1", "process_keys": 24, "total_keys": 156}] client = Client() args = Namespace(kind="slow-query", fields=None, limit=5, text="") payload, request = MODULE.query_dashboard(client, args, 1, 2) self.assertEqual(client.assert_path, "/api/slow_query/available_fields") self.assertEqual(client.request["path"], "/api/slow_query/list") self.assertEqual(client.request["params"]["fields"], "query,cop_time,process_keys,total_keys") self.assertEqual(client.request["params"]["end_time"], 902) self.assertEqual(client.request["params"]["limit"], 100) self.assertEqual(request["incident_time_window"], {"start": 1, "end": 2}) self.assertEqual(request["result_limit"], 5) self.assertEqual(request["field_selection"]["availability_status"], "available") self.assertIn("wait_time", request["field_selection"]["unsupported_fields"]) self.assertEqual(payload[0]["total_keys"], 156) def test_slow_query_derives_finish_interval_and_excludes_non_overlap(self): rows, counts = MODULE.slow_query_execution_rows([ {"timestamp": 1030, "query_time": 40, "memory_max": 4096}, {"timestamp": 1100, "query_time": 10, "memory_max": 8192}, ], 1000, 1020, 5) self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["start_time"], 990) self.assertEqual(rows[0]["finish_time"], 1030) self.assertEqual(rows[0]["incident_overlap_seconds"], 20) self.assertTrue(rows[0]["overlaps_incident_window"]) self.assertEqual(counts["raw_record_count"], 2) self.assertEqual(counts["overlapping_record_count"], 1) self.assertEqual(counts["excluded_non_overlapping_record_count"], 1) def test_slow_query_explicit_fields_and_compact_preview(self): selected, state = MODULE.slow_query_field_selection( MODULE.field_list("process_keys,total_keys", MODULE.SLOW_QUERY_DIAGNOSTIC_FIELDS), ["process_keys", "total_keys"], ) self.assertEqual(selected, ["process_keys", "total_keys"]) self.assertEqual(state["profile"], "explicit") rows = [{"digest": "d1", "process_keys": 24, "total_keys": 156, "binary_plan": "large"}] self.assertEqual( MODULE.compact_rows(rows, 1, ["digest", *selected]), [{"digest": "d1", "process_keys": 24, "total_keys": 156}], ) def test_slow_query_falls_back_to_requested_projection_when_schema_endpoint_is_absent(self): selected, state = MODULE.slow_query_field_selection( list(MODULE.SLOW_QUERY_DIAGNOSTIC_FIELDS), None ) self.assertEqual(selected, list(MODULE.SLOW_QUERY_DIAGNOSTIC_FIELDS)) self.assertEqual(state["availability_status"], "unavailable") self.assertEqual(state["unsupported_fields"], []) def test_rsa_public_key_parse_and_pkcs1_encryption(self): # Deterministic structural test with a synthetic 256-bit modulus. modulus = (1 << 255) + 76543 exponent = 65537 rsa = der(0x30, integer(modulus) + integer(exponent)) algorithm = der(0x30, der(0x06, b"\x2a\x86\x48\x86\xf7\x0d\x01\x01\x01") + der(0x05, b"")) spki = der(0x30, algorithm + der(0x03, b"\x00" + rsa)) public_key = base64.b64encode(spki).decode() self.assertEqual(MODULE.parse_rsa_public_key(public_key), (modulus, exponent)) encrypted = base64.b64decode(MODULE.rsa_encrypt_password("secret", public_key)) self.assertEqual(len(encrypted), 32) def test_persist_is_explicit_and_report_excludes_raw(self): self.assertEqual(MODULE.persist(None, "metrics", {"raw": True}, {"status": "ok"}), []) with tempfile.TemporaryDirectory() as directory: paths = MODULE.persist(directory, "metrics", {"raw": True}, {"status": "ok"}) self.assertEqual(len(paths), 2) report = json.loads((Path(directory) / "workspace_evidence_report.json").read_text()) self.assertEqual(report, {"status": "ok"}) if __name__ == "__main__": unittest.main()