from __future__ import annotations import importlib.util import json import os import subprocess import sys import urllib.error import urllib.request import zipfile from pathlib import Path SCRIPT_PATH = ( Path(__file__).resolve().parents[1] / "scripts" / "clinic_cli.py" ) SPEC = importlib.util.spec_from_file_location("clinic_cli", SCRIPT_PATH) MODULE = importlib.util.module_from_spec(SPEC) assert SPEC and SPEC.loader SPEC.loader.exec_module(MODULE) API_SCRIPT_PATH = SCRIPT_PATH.parent / "clinic_api.py" API_SPEC = importlib.util.spec_from_file_location("clinic_api", API_SCRIPT_PATH) API_MODULE = importlib.util.module_from_spec(API_SPEC) assert API_SPEC and API_SPEC.loader API_SPEC.loader.exec_module(API_MODULE) def _set_workspace_dir(tmp_path, monkeypatch) -> Path: workspace_dir = tmp_path / "ws-test" monkeypatch.setenv("INVESTIBOOK_WORKSPACE_DIR", str(workspace_dir)) monkeypatch.delenv("WORKSPACE_EVIDENCE_DIR", raising=False) return workspace_dir def _write_json(path: Path, payload: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def test_query_evidence_merge_preserves_statement_summary_and_unknown_fields(): merged = MODULE._merge_evidence_payload( "query", {"statement_summary": [{"digest": "old"}], "future": {"keep": True}}, {"statement_summary": [{"digest": "new"}], "decoded_plan": {"plan": "p"}}, ) assert [row["digest"] for row in merged["statement_summary"]] == ["old", "new"] assert merged["future"] == {"keep": True} assert merged["decoded_plan"] == {"plan": "p"} def test_query_evidence_compact_includes_statement_summary(): compact = MODULE._compact_report_evidence( "query", {"statement_summary": [{"digest": "d1"}]} ) assert compact["queries"] == [{"digest": "d1", "source": "statement_summary"}] def _run_concurrent_persist_processes( tmp_path: Path, workspace_dir: Path, calls: list[tuple[str, str, str, dict]], ) -> None: start_file = tmp_path / "start" child_code = r""" import importlib.util import json import os import sys import time from pathlib import Path script_path = Path(sys.argv[1]) command = sys.argv[2] kind = sys.argv[3] cluster_id = sys.argv[4] payload = json.loads(sys.argv[5]) start_file = os.environ["CLINIC_CLI_TEST_START_FILE"] while not os.path.exists(start_file): time.sleep(0.001) spec = importlib.util.spec_from_file_location("clinic_cli_child", script_path) module = importlib.util.module_from_spec(spec) assert spec and spec.loader spec.loader.exec_module(module) module._persist_raw_evidence(command, kind, cluster_id, payload) """ env = os.environ.copy() env["INVESTIBOOK_WORKSPACE_DIR"] = str(workspace_dir) env["CLINIC_CLI_TEST_START_FILE"] = str(start_file) processes = [ subprocess.Popen( [ sys.executable, "-c", child_code, str(SCRIPT_PATH), command, kind, cluster_id, json.dumps(payload), ], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) for command, kind, cluster_id, payload in calls ] start_file.write_text("go", encoding="utf-8") failures = [] for process in processes: stdout, stderr = process.communicate(timeout=20) if process.returncode != 0: failures.append((process.returncode, stdout, stderr)) assert not failures class FakeAPI: def __init__(self, *, deploy_type: str = "dedicated", parent_id: str = ""): self.sql_timeouts = [] self.sql_queries = [] self.instant_queries = [] self.range_join_queries = [] self.deploy_type = deploy_type self.parent_id = parent_id self.resource_pool_component_calls = [] self.search_cluster_calls = [] self.ticdc_calls = [] def search_clusters( self, *, query: str = "", tenant_id: str = "", cluster_id: str = "", show_deleted: bool = False, cluster_status: str = "", limit: int = 20, page: int = 1, ): self.search_cluster_calls.append( { "query": query, "tenant_id": tenant_id, "cluster_id": cluster_id, "show_deleted": show_deleted, "cluster_status": cluster_status, "limit": limit, "page": page, } ) return { "status": "success", "raw_count": 3, "total": 3, "items": [ { "clusterID": "c-1", "orgID": "42", "tenantID": "99", "clusterDeployTypeV2": "dedicated", "name": "demo-a", "status": "active", }, { "clusterID": "c-2", "orgID": "43", "tenantID": "98", "clusterDeployTypeV2": "dedicated", "name": "demo-b", "status": "active", }, { "clusterID": "c-3", "orgID": "44", "tenantID": "99", "clusterDeployTypeV2": "starter", "name": "demo-c", "status": "active", }, ], } def search_cluster_exact(self, cluster_id: str, show_deleted: bool = True): return { "status": "success", "item": { "clusterID": cluster_id, "orgID": "42", "tenantID": "99", "clusterDeployTypeV2": self.deploy_type, "parentID": self.parent_id, "name": "demo", }, } def get_cluster_by_id(self, cluster_id: str): metadata = { "clusterId": cluster_id, "orgId": "99", "deployType": self.deploy_type, } if self.parent_id: metadata["diagnosticContext"] = {"parentId": self.parent_id} return metadata @staticmethod def is_byoc_metadata(metadata: dict): return str(metadata.get("deployType") or "").lower() in {"byoc", "byoc-host"} def get_cluster_detail(self, org_id: str, cluster_id: str): return { "clusterID": cluster_id, "orgID": org_id, "clusterDeployTypeV2": self.deploy_type, "parentID": self.parent_id, "region": "aws-us-west-2", "components": [{"component": "stale-detail-component", "count": 99}], } def get_resource_pool_components(self, org_id: str, cluster_id: str): self.resource_pool_component_calls.append((org_id, cluster_id)) return { "items": [ { "component_type": "tidb", "node_count": 2, "cpu": "4", "memory_gi": "16", "image": {"tag": "v8.5.4-runtime"}, "storages": [{"type": "data", "size_gi": 100, "iops": 3000, "throughput_mbs": 125}], } ] } def query_range(self, cluster_id: str, query: str, start: int, end: int, step: str = "60"): return { "status": "success", "data": { "resultType": "matrix", "result": [ {"metric": {"instance": "tidb-0"}, "values": [[start, "1.0"], [end, "3.0"]]}, {"metric": {"instance": "tidb-1"}, "values": [[start, "2.0"], [end, "4.0"]]}, ], }, } def query_instant(self, cluster_id: str, query: str, time: int | None = None): self.instant_queries.append({"cluster_id": cluster_id, "query": query, "time": time}) return { "status": "success", "data": { "resultType": "vector", "result": [{"metric": {}, "value": [time or 0, "3"]}], }, } def query_range_join_instance_to_pod( self, cluster_id: str, query: str, namespace: str, pod_pattern: str, start: int, end: int, step: str = "60", instance_label: str = "instance", pod_label: str = "pod", source_pod_label: str = "source_pod", ): self.range_join_queries.append( { "cluster_id": cluster_id, "query": query, "namespace": namespace, "pod_pattern": pod_pattern, "start": start, "end": end, "step": step, } ) return { "status": "success", "data": { "resultType": "matrix", "result": [ { "metric": {"pod": "db-tikv-2", "instance": "node-a"}, "values": [[start, "10"], [end, "20"]], }, { "metric": {"pod": "db-tikv-1", "instance": "node-b"}, "values": [[start, "30"], [end, "40"]], }, ], }, } def query_sql(self, cluster_id: str, sql: str, timeout: int = 60): self.sql_timeouts.append(timeout) self.sql_queries.append(sql) return { "columns": ["sql_digest", "cpu_ms"], "rows": [["abc", "100"], ["def", "50"]], } def ticdc_changefeeds_list(self, cluster_id: str, *, org_id: str = ""): self.ticdc_calls.append(("list", cluster_id, org_id)) return { "changefeeds": [ { "id": "1234", "display_id": "9007199254740993", "display_name": "orders-to-kafka", "status": "warning", "checkpoint_tso": "4611686018427387904", "checkpoint_ts": 1787900000, "downstream_type": "kafka", "network_info": {"endpoint": "private.customer.example"}, "downstream_config": {"password": "secret"}, }, { "id": "", "display_id": "9007199254740994", "display_name": "creating-feed", "status": "creating", }, ] } def ticdc_changefeed_get( self, cluster_id: str, changefeed_id: str, *, org_id: str = "", ): self.ticdc_calls.append(("detail", cluster_id, changefeed_id, org_id)) return { "changefeed": { "id": changefeed_id, "display_id": "9007199254740993", "display_name": "orders-to-kafka", "status": "warning", "checkpoint_tso": "4611686018427387904", "downstream_type": "kafka", "rcu": "4", "table_config": {"filter_rules": ["secret.*"]}, "downstream_config": {"password": "secret"}, } } def loki_query_range( self, cluster_id: str, query: str, start_ns: int, end_ns: int, *, limit: int = 200, direction: str = "backward", ): return { "status": "success", "data": { "resultType": "streams", "result": [ { "stream": {"container": "tidb", "instance": "tidb-0"}, "values": [ [str(start_ns), "2026-04-27T12:00:00Z [ERROR] rpc timeout"], [str(end_ns), "2026-04-27T12:00:30Z [ERROR] rpc timeout"], ], } ], }, } def activity_events( self, org_id: str, cluster_id: str, start: int, end: int, *, name: str = "", severity: int = 0, limit: int = 100, cursor: str = "", ): return { "activities": [ {"displayName": "ScaleCluster", "severity": severity, "targetID": cluster_id} ], "partial": False, "cursor": "", } def continuous_profiling_group_profiles(self, cluster_id: str, start: int, end: int): return { "group_profiles": [ {"ts": start, "state": "finished", "component_num": {"tidb": 1}} ] } def continuous_profiling_detail(self, cluster_id: str, ts: int): return { "target_profiles": [ {"profile_type": "cpu", "state": "finished", "target": {"component": "tidb"}} ] } def continuous_profiling_action_token( self, cluster_id: str, *, ts: int, profile_type: str, component: str, address: str, ): return {"status": "success", "text": "download-token"} def continuous_profiling_download(self, cluster_id: str, *, token: str, output_path: Path): output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_bytes(b"profile-bytes") return {"status": "success", "path": str(output_path), "size_bytes": output_path.stat().st_size} def diagnostic_files(self, cluster_id: str, file_type: str, start: int, end: int): return { "records": [ {"instance": "tidb-0", "files": [{"filename": "replayer.zip", "key": "a/b/replayer.zip"}]} ] } def diagnostic_file_download( self, cluster_id: str, *, key: str, output_path: Path, provider: str = "", region: str = "", ): output_path.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(output_path, "w") as zf: zf.writestr("plan.json", '{"ok": true}') return {"status": "success", "path": str(output_path), "size_bytes": output_path.stat().st_size} class FakeBYOCAPI(FakeAPI): def __init__(self): super().__init__(deploy_type="byoc", parent_id="byoc-host-1") self.byoc_calls = [] def byoc_physical_cluster_id(self, cluster_id: str, metadata: dict | None = None): return "byoc-host-1" def byoc_slow_query_available_fields(self, cluster_id: str): self.byoc_calls.append(("slow-fields", cluster_id)) return ["query", "digest"] def byoc_slow_query_list(self, cluster_id: str, start: int, end: int, **kwargs): self.byoc_calls.append(("slow-list", cluster_id, start, end, kwargs)) return [{"digest": "slow-digest", "query_time": 1.25}] def byoc_slow_query_detail(self, cluster_id: str, digest: str, timestamp: str, connect_id: str, **kwargs): self.byoc_calls.append(("slow-detail", cluster_id, digest, timestamp, connect_id, kwargs)) return {"digest": digest, "connection_id": connect_id, "decoded_plan": "plan"} def byoc_top_sql_instances(self, cluster_id: str, start: int, end: int): self.byoc_calls.append(("top-instances", cluster_id, start, end)) return {"status": "success", "data": [{"instance": "tidb-0:4000"}]} def byoc_top_sql_summary(self, cluster_id: str, start: int, end: int, window: str, top: int, instance: str, instance_type: str): self.byoc_calls.append(("top-summary", cluster_id, start, end, window, top, instance, instance_type)) return {"status": "success", "data": [{"sql_digest": "top-digest", "cpu_ms": 10}]} def byoc_statements_available_fields(self, cluster_id: str): self.byoc_calls.append(("statement-fields", cluster_id)) return ["digest", "schema_name"] def byoc_statements_statement_types(self, cluster_id: str): self.byoc_calls.append(("statement-types", cluster_id)) return ["Select", "Insert"] def byoc_statements_list(self, cluster_id: str, start: int, end: int, **kwargs): self.byoc_calls.append(("statement-list", cluster_id, start, end, kwargs)) return [{"digest": "statement-digest", "schema_name": "test"}] def byoc_statements_plans(self, cluster_id: str, start: int, end: int, digest: str, **kwargs): self.byoc_calls.append(("statement-plans", cluster_id, start, end, digest, kwargs)) return [{"plan_digest": "plan-digest"}] def byoc_statements_plan_detail(self, cluster_id: str, start: int, end: int, digest: str, **kwargs): self.byoc_calls.append(("statement-plan-detail", cluster_id, start, end, digest, kwargs)) return {"digest": digest, "binary_plan_text": "TableReader"} def byoc_continuous_profiling_group_profiles(self, cluster_id: str, start: int, end: int): self.byoc_calls.append(("profile-list", cluster_id, start, end)) return [{"ts": start, "state": "finished", "component_num": {"tidb": 1}}] def byoc_continuous_profiling_group_detail(self, cluster_id: str, ts: int): self.byoc_calls.append(("profile-detail", cluster_id, ts)) return {"target_profiles": [{"profile_type": "cpu", "state": "finished"}]} def test_run_search_clusters_filters_by_tenant_id_without_persisting(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args(["search", "clusters", "--tenant-id", "99", "--limit", "50"]) api = FakeAPI() result = MODULE.run_search_clusters(args, api=api) assert result["status"] == "ok" assert result["command"] == "search clusters" assert result["kind"] == "cluster_search" assert result["summary"]["count"] == 2 assert [cluster["cluster_id"] for cluster in result["clusters"]] == ["c-1", "c-3"] assert {cluster["tenant_id"] for cluster in result["clusters"]} == {"99"} assert api.search_cluster_calls == [ { "query": "99", "tenant_id": "99", "cluster_id": "", "show_deleted": False, "cluster_status": "", "limit": 50, "page": 1, } ] assert "persisted_evidence" not in result assert not (tmp_path / "ws-test" / "evidences" / "cluster_meta.json").exists() def test_run_search_clusters_filters_by_cluster_id_and_tenant_id(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( ["search", "clusters", "--query", "demo", "--tenant-id", "99", "--cluster-id", "c-3"] ) result = MODULE.run_search_clusters(args, api=FakeAPI()) assert result["status"] == "ok" assert [cluster["cluster_id"] for cluster in result["clusters"]] == ["c-3"] assert result["clusters"][0]["tenant_id"] == "99" def test_run_cluster_metadata_returns_compact_payload(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args(["cluster-metadata", "--cluster-id", "123", "--include-components"]) api = FakeAPI() result = MODULE.run_cluster_metadata(args, api=api) assert result["status"] == "ok" assert result["cluster"]["cluster_id"] == "123" assert result["cluster"]["components"][0]["component_type"] == "tidb" assert result["cluster"]["components"][0]["image"]["tag"] == "v8.5.4-runtime" assert result["cluster"]["components"][0]["storages"][0]["iops"] == 3000 assert result["cluster"]["topology_source"] == "resource_pool_components" assert result["cluster"]["topology_cluster_id"] == "123" assert api.resource_pool_component_calls == [("42", "123")] assert "workspace_evidence" not in result assert "payload" not in result def test_cluster_metadata_prefers_deploy_type_over_cluster_type_cloud(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) class CloudDetailAPI(FakeAPI): def get_cluster_detail(self, org_id: str, cluster_id: str): detail = super().get_cluster_detail(org_id, cluster_id) detail["clusterType"] = "cloud" detail.pop("clusterDeployTypeV2", None) return detail parser = MODULE.build_parser() args = parser.parse_args(["cluster-metadata", "--cluster-id", "123", "--include-components"]) api = CloudDetailAPI(deploy_type="dedicated") result = MODULE.run_cluster_metadata(args, api=api) assert result["status"] == "ok" assert result["cluster"]["topology_source"] == "resource_pool_components" assert api.resource_pool_component_calls == [("42", "123")] def test_cluster_metadata_parent_scope_requires_explicit_parent(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "cluster-metadata", "--cluster-id", "123", "--include-components", "--resource-pool-scope", "parent", ] ) api = FakeAPI(deploy_type="premium", parent_id="pool-1") result = MODULE.run_cluster_metadata(args, api=api) assert result["status"] == "ok" assert result["cluster"]["topology_cluster_id"] == "pool-1" assert api.resource_pool_component_calls == [("42", "pool-1")] def test_cluster_metadata_rejects_shared_topology(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args(["cluster-metadata", "--cluster-id", "123", "--include-components"]) api = FakeAPI(deploy_type="starter") result = MODULE.run_cluster_metadata(args, api=api) assert result["status"] == "error" assert "not supported" in result["error"] assert api.resource_pool_component_calls == [] def test_helper_root_prefers_builtin_skill_copy(): assert MODULE._clinic_helper_root() == SCRIPT_PATH.parents[1] def test_run_metrics_returns_chart_ref_payload(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "metrics", "--cluster-id", "123", "--name", "TiDB Total QPS", "--query", "sum(rate(tidb_server_query_total[1m]))", "--start", "100", "--end", "200", "--metric-group-id", "tidb_qps_incident_window", ] ) result = MODULE.run_metrics(args, api=FakeAPI()) evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) metric = evidence["metrics"][0] assert result["status"] == "ok" assert "workspace_evidence" not in result assert "payload" not in result assert metric["id"] == "tidb_qps_incident_window" assert metric["group"] == "tidb_qps_incident_window" assert metric["ref"] == { "type": "metric", "metric_group_id": "tidb_qps_incident_window", } assert metric["metric_group_id"] == "tidb_qps_incident_window" assert metric["chart_data_ref"]["kind"] == "promql_range" assert metric["summary"]["avg"] == 2.5 assert metric["series_summaries"] == [ { "labels": {"instance": "tidb-0"}, "point_count": 2, "min": 1.0, "max": 3.0, "avg": 2.0, }, { "labels": {"instance": "tidb-1"}, "point_count": 2, "min": 2.0, "max": 4.0, "avg": 3.0, }, ] def test_time_range_args_accept_iso8601_and_epoch_ms(): parser = MODULE.build_parser() args = parser.parse_args( [ "metrics", "--cluster-id", "123", "--name", "TiDB Total QPS", "--query", "sum(rate(tidb_server_query_total[1m]))", "--start", "2026-04-22T06:24:00Z", "--end", "1776842640000", ] ) assert args.start == 1776839040 assert args.end == 1776842640 def test_logs_time_range_preserves_epoch_ns(): parser = MODULE.build_parser() args = parser.parse_args( [ "logs", "--cluster-id", "123", "--query", '{component="tidb"}', "--start", "1776839040000000000", "--end", "2026-04-22 07:24:00+00:00", ] ) assert args.start == 1776839040000000000 assert args.end == 1776842640 def test_run_sql_shapes_topsql_payload(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "sql", "--cluster-id", "123", "--query-type", "topsql", "--name", "TopSQL by CPU", "--sql", "SELECT * FROM clinic_data_proxy.topsql", ] ) result = MODULE.run_sql(args, api=FakeAPI()) assert result["status"] == "ok" assert "workspace_evidence" not in result evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["topsql"][0]["sql_digest"] == "abc" assert evidence["slow_queries"] == [] def test_sql_default_timeout_is_60s(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "sql", "--cluster-id", "123", "--query-type", "topsql", "--name", "TopSQL by CPU", "--sql", "SELECT * FROM clinic_data_proxy.topsql", ] ) api = FakeAPI() MODULE.run_sql(args, api=api) assert api.sql_timeouts == [60] def test_byoc_sql_fails_closed_without_warehouse_fallback(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "sql", "--cluster-id", "byoc-logical", "--query-type", "slow_queries", "--name", "wrong path", "--sql", "SELECT * FROM clinic_data_proxy.slow_query_logs", ] ) api = FakeBYOCAPI() result = MODULE.run_sql(args, api=api) assert result["status"] == "error" assert "warehouse SQL is not supported for BYOC" in result["error"] assert api.sql_queries == [] def test_byoc_slow_query_list_persists_query_evidence(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "slow-query", "list", "--cluster-id", "byoc-logical", "--start", "100", "--end", "200", "--db", "test", "--plan", "plan-1", ] ) result = MODULE.run_slow_query(args, api=FakeBYOCAPI()) assert result["status"] == "ok" evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["slow_queries"][0]["digest"] == "slow-digest" assert evidence["slow_queries"][0]["operation"] == "list" assert evidence["statements"] == [] def test_byoc_top_sql_summary_persists_query_evidence(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "top-sql", "summary", "--cluster-id", "byoc-logical", "--start", "100", "--end", "200", "--instance", "tidb-0:4000", "--instance-type", "tidb", ] ) result = MODULE.run_top_sql(args, api=FakeBYOCAPI()) assert result["status"] == "ok" evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["topsql"][0]["sql_digest"] == "top-digest" assert evidence["topsql"][0]["operation"] == "summary" def test_byoc_top_sql_instance_envelope_flattens_component_groups(): rows = MODULE._top_sql_instance_rows({ "status": "success", "data": {"tidb": ["tidb-0:4000"], "tikv": [{"instance": "tikv-0:20160"}]}, }) assert rows == [ { "instance": "tidb-0:4000", "instance_type": "tidb", "source": "topsql", "operation": "instances", }, { "instance": "tikv-0:20160", "instance_type": "tikv", "source": "topsql", "operation": "instances", }, ] def test_byoc_statements_operations_merge_into_queries_and_report(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() api = FakeBYOCAPI() commands = [ [ "statements", "list", "--cluster-id", "byoc-logical", "--start", "100", "--end", "200", "--schema", "test", ], [ "statements", "plans", "--cluster-id", "byoc-logical", "--start", "100", "--end", "200", "--digest", "statement-digest", ], [ "statements", "plan-detail", "--cluster-id", "byoc-logical", "--start", "100", "--end", "200", "--digest", "statement-digest", "--plan", "plan-digest", ], ] result = None for argv in commands: result = MODULE.run_statements(parser.parse_args(argv), api=api) assert result["status"] == "ok" assert result is not None evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert {row["operation"] for row in evidence["statements"]} == {"list", "plans", "plan-detail"} report = MODULE._json_load(Path(result["persisted_evidence"]["report_path"])) statement_rows = [row for row in report["evidence"]["queries"] if row.get("source") == "statement"] assert len(statement_rows) == 3 def test_byoc_statements_discovery_does_not_persist_evidence(tmp_path, monkeypatch): workspace_dir = _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( ["statements", "available-fields", "--cluster-id", "byoc-logical"] ) result = MODULE.run_statements(args, api=FakeBYOCAPI()) assert result["status"] == "ok" assert result["kind"] == "query_metadata" assert "persisted_evidence" not in result assert not (workspace_dir / "evidences" / "queries.json").exists() def test_ticdc_list_persists_only_diagnostic_fields(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( ["ticdc", "list", "--cluster-id", "123", "--org-id", "99"] ) api = FakeAPI() result = MODULE.run_ticdc(args, api=api) assert result["status"] == "ok" assert result["summary"] == {"changefeed_count": 2, "operation": "list"} assert api.ticdc_calls == [("list", "123", "99")] assert result["changefeeds"][0]["checkpoint_tso"] == "4611686018427387904" assert "network_info" not in result["changefeeds"][0] assert "downstream_config" not in result["changefeeds"][0] evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["changefeeds"] == result["changefeeds"] assert "secret" not in json.dumps(evidence) report = MODULE._json_load(Path(result["persisted_evidence"]["report_path"])) rows = [ row for row in report["evidence"]["queries"] if row.get("source") == "ticdc_changefeed" ] assert len(rows) == 2 assert rows[0]["operation"] == "list" def test_ticdc_detail_uses_kernel_id_and_redacts_configuration(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "ticdc", "detail", "--cluster-id", "123", "--changefeed-id", "1234", ] ) api = FakeAPI() result = MODULE.run_ticdc(args, api=api) assert result["status"] == "ok" assert api.ticdc_calls == [("detail", "123", "1234", "")] assert result["changefeeds"] == [ { "id": "1234", "display_id": "9007199254740993", "display_name": "orders-to-kafka", "status": "warning", "downstream_type": "kafka", "checkpoint_tso": "4611686018427387904", "rcu": "4", "source": "ticdc_changefeed", "operation": "detail", } ] assert "secret" not in json.dumps(result) def test_clinic_api_ticdc_routes_validate_ids_and_use_canonical_org(monkeypatch): monkeypatch.setenv("CLINIC_API_KEY", "test-key") api = API_MODULE.ClinicAPI() calls = [] def fake_get(path, params=None, extra_headers=None): calls.append((path, params, extra_headers)) return {"changefeeds": []} monkeypatch.setattr(api, "_get", fake_get) monkeypatch.setattr(api, "_canonical_org_id", lambda cluster_id: "99") api.ticdc_changefeeds_list("123") api.ticdc_changefeed_get("123", "456", org_id="88") assert calls == [ ( "/data-proxy/api/v1/orgs/99/clusters/123/ticdc/changefeeds", None, None, ), ( "/data-proxy/api/v1/orgs/88/clusters/123/ticdc/changefeeds/456", None, None, ), ] try: api.ticdc_changefeed_get("123", "display-name") except ValueError as exc: assert "kernel_changefeed_id" in str(exc) else: raise AssertionError("non-numeric changefeed IDs must be rejected") def test_clinic_api_retries_transient_url_error(monkeypatch): calls = [] class FakeResponse: def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def read(self): return json.dumps({"status": "success"}).encode() def fake_urlopen(req, timeout): calls.append(timeout) if len(calls) == 1: raise urllib.error.URLError("temporary timeout") return FakeResponse() monkeypatch.setattr(API_MODULE.urllib.request, "urlopen", fake_urlopen) monkeypatch.setattr(API_MODULE.time, "sleep", lambda _: None) req = urllib.request.Request("https://clinic.example.test/data-proxy/query") assert API_MODULE.ClinicAPI._do(req) == {"status": "success"} assert calls == [60, 60] def test_run_logs_groups_log_patterns(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "logs", "--cluster-id", "123", "--query", '{container="tidb"} |~ `(?i)\\[error\\]`', "--start", "100", "--end", "200", ] ) result = MODULE.run_logs(args, api=FakeAPI()) assert result["status"] == "ok" assert "workspace_evidence" not in result evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["log_counts"][0]["count"] == 2 assert result["summary"]["line_count"] == 2 def test_run_activity_events_returns_event_evidence(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "activity-events", "--cluster-id", "123", "--start", "100", "--end", "200", ] ) result = MODULE.run_activity_events(args, api=FakeAPI()) assert result["status"] == "ok" assert "workspace_evidence" not in result evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["events"][0]["displayName"] == "ScaleCluster" def test_run_topology_snapshot_combines_metrics_activity_and_current_topology(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) class SnapshotAPI(FakeAPI): def query_range(self, cluster_id: str, query: str, start: int, end: int, step: str = "60"): raise AssertionError("topology-snapshot must use instant metrics") def query_instant(self, cluster_id: str, query: str, time: int | None = None): self.instant_queries.append({"cluster_id": cluster_id, "query": query, "time": time}) if query == "tidb_server_maxprocs": value = "64" instances = ["db-tidb-0", "db-tidb-1"] elif query == "tikv_server_cpu_cores_quota": value = "32" instances = ["db-tikv-0", "db-tikv-1", "db-tikv-2"] elif query == "tiflash_proxy_tikv_server_cpu_cores_quota": value = "32" instances = ["db-tiflash-0"] else: value = "4" instances = ["db-pd-0"] return { "status": "success", "data": { "resultType": "vector", "result": [ {"metric": {"instance": instance}, "value": [time or 0, value]} for instance in instances ], }, } def activity_events( self, org_id: str, cluster_id: str, start: int, end: int, *, name: str = "", severity: int = 0, limit: int = 100, cursor: str = "", ): return { "activities": [ { "name": "scalecluster", "displayName": "ScaleCluster", "calibration_time": start + 60, "tags": {"status": "succeeded"}, "payload": {"summary": "Scale TiDB from 2 to 3 nodes"}, }, { "name": "backupcluster", "displayName": "BackupCluster", "calibration_time": start + 120, "payload": {"summary": "The auto-backup task was succeeded."}, }, ], "partial": False, "cursor": "", } parser = MODULE.build_parser() args = parser.parse_args( [ "topology-snapshot", "--cluster-id", "123", "--at", "2026-06-30T04:00:00Z", "--metric-window-minutes", "1", "--activity-before-days", "1", "--activity-after-days", "0", ] ) api = SnapshotAPI() result = MODULE.run_topology_snapshot(args, api=api) assert result["status"] == "ok" assert {call["time"] for call in api.instant_queries} == {1782792000} assert result["summary"]["metric_query_type"] == "instant" assert result["summary"]["metric_window_usage"] == "report_metadata_only_not_used_for_node_count" assert result["summary"]["historical_metrics_complete"] is True assert result["summary"]["missing_metric_components"] == [] assert result["summary"]["error_metric_components"] == [] assert result["summary"]["current_topology_usage"] == "comparison_only_not_historical_fallback" assert result["summary"]["components"][0]["component"] == "tidb" assert result["summary"]["components"][0]["series_count"] == 2 assert result["summary"]["components"][0]["instance_count"] == 2 assert result["summary"]["components"][0]["instance_count_source"] == "identity_labels" assert result["summary"]["components"][0]["identity_labels_complete"] is True assert result["summary"]["components"][0]["instances_truncated"] is False assert result["summary"]["components"][0]["historical_metric_available"] is True assert result["summary"]["components"][0]["single_node_cores"] == 64.0 assert result["summary"]["topology_event_count"] == 1 assert result["summary"]["current_topology_status"] == "ok" evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) snapshot = evidence["topology_snapshots"][0] assert evidence["time_window"] == {"start": 1782791940, "end": 1782792060, "step": "60"} assert snapshot["metric_query"]["type"] == "instant" assert snapshot["metric_query"]["at"] == 1782792000 assert snapshot["metric_query"]["metric_window_usage"] == "report_metadata_only_not_used_for_node_count" assert snapshot["metric_query"]["historical_metrics_complete"] is True assert snapshot["metric_query"]["error_metric_components"] == [] assert snapshot["components"][0]["instance_count"] == 2 assert snapshot["components"][0]["instance_count_source"] == "identity_labels" assert snapshot["components"][0]["identity_labels_complete"] is True assert snapshot["components"][0]["instances_truncated"] is False assert snapshot["components"][0]["instances"][0]["identity"] == "instance:db-tidb-0" assert snapshot["activity_window"] == {"start": 1782705600, "end": 1782792000} assert snapshot["activity"]["topology_events"][0]["display_name"] == "ScaleCluster" assert snapshot["interpretation"]["current_topology_is_not_historical_evidence"] is True assert snapshot["interpretation"]["metric_window_is_not_range_query_evidence"] is True assert snapshot["interpretation"]["current_topology_is_not_historical_fallback_when_metrics_missing"] is True assert snapshot["interpretation"]["do_not_infer_scale_change_from_current_topology_difference_alone"] is True assert snapshot["interpretation"]["do_not_infer_historical_topology_from_no_activity_events_alone"] is True def test_component_metric_snapshot_counts_all_instances_when_sample_is_truncated(): class ManyInstanceAPI: def query_instant(self, cluster_id: str, query: str, time: int | None = None): return { "status": "success", "data": { "resultType": "vector", "result": [ {"metric": {"instance": f"db-tikv-{index}"}, "value": [time or 0, "32"]} for index in range(301) ], }, } snapshot = MODULE._component_metric_snapshot( api=ManyInstanceAPI(), cluster_id="cluster-1", component="tikv", promql="tikv_server_cpu_cores_quota", at=1782792000, ) assert snapshot["series_count"] == 301 assert snapshot["instance_count"] == 301 assert len(snapshot["instances"]) == 300 assert snapshot["instances_truncated"] is True assert snapshot["instance_count_source"] == "identity_labels" assert snapshot["identity_labels_complete"] is True assert snapshot["summary"]["instance_count"] == 301 assert snapshot["summary"]["instance_count_source"] == "identity_labels" assert snapshot["summary"]["instances_truncated"] is True def test_component_metric_snapshot_falls_back_to_series_count_without_identity_labels(): class NoIdentityAPI: def query_instant(self, cluster_id: str, query: str, time: int | None = None): return { "status": "success", "data": { "resultType": "vector", "result": [ {"metric": {"job": "tidb"}, "value": [time or 0, "64"]}, {"metric": {"job": "tidb"}, "value": [time or 0, "64"]}, ], }, } snapshot = MODULE._component_metric_snapshot( api=NoIdentityAPI(), cluster_id="cluster-1", component="tidb", promql="tidb_server_maxprocs", at=1782792000, ) assert snapshot["series_count"] == 2 assert snapshot["instance_count"] == 2 assert snapshot["instance_count_source"] == "series_count_fallback" assert snapshot["identity_labels_complete"] is False assert snapshot["instances"] == [] assert snapshot["instances_truncated"] is False assert snapshot["historical_metric_available"] is True def test_topology_snapshot_iso8601_timezone_is_parsed_by_cli_parser(): parser = MODULE.build_parser() args = parser.parse_args( [ "topology-snapshot", "--cluster-id", "123", "--at", "2026-06-30T12:00:00+08:00", ] ) assert args.at == 1782792000 def test_run_topology_snapshot_marks_missing_metrics_not_current_topology_fallback(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) class EmptyMetricSnapshotAPI(FakeAPI): def query_instant(self, cluster_id: str, query: str, time: int | None = None): return {"status": "success", "data": {"resultType": "vector", "result": []}} def activity_events( self, org_id: str, cluster_id: str, start: int, end: int, *, name: str = "", severity: int = 0, limit: int = 100, cursor: str = "", ): return {"activities": [], "partial": False, "cursor": ""} parser = MODULE.build_parser() args = parser.parse_args( [ "topology-snapshot", "--cluster-id", "123", "--at", "2026-06-30T12:00:00+08:00", ] ) result = MODULE.run_topology_snapshot(args, api=EmptyMetricSnapshotAPI()) assert result["status"] == "ok" assert result["summary"]["at"] == 1782792000 assert result["summary"]["historical_metrics_complete"] is False assert result["summary"]["missing_metric_components"] == ["tidb", "tikv", "tiflash", "pd"] assert result["summary"]["error_metric_components"] == [] assert result["summary"]["current_topology_usage"] == "comparison_only_not_historical_fallback" assert result["summary"]["components"][0]["evidence_status"] == "no_data" assert result["summary"]["components"][0]["historical_metric_available"] is False evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) snapshot = evidence["topology_snapshots"][0] assert snapshot["metric_query"]["historical_metrics_complete"] is False assert snapshot["metric_query"]["missing_metric_components"] == ["tidb", "tikv", "tiflash", "pd"] assert snapshot["metric_query"]["error_metric_components"] == [] assert snapshot["interpretation"]["current_topology_is_not_historical_fallback_when_metrics_missing"] is True assert snapshot["interpretation"]["do_not_infer_historical_topology_from_no_activity_events_alone"] is True def test_run_topology_snapshot_includes_metric_errors_in_missing_components(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) class MetricErrorSnapshotAPI(FakeAPI): def query_instant(self, cluster_id: str, query: str, time: int | None = None): if query == "tidb_server_maxprocs": return {"error": "HTTP 503 data proxy timeout"} return { "status": "success", "data": { "resultType": "vector", "result": [{"metric": {"instance": f"{query}-0"}, "value": [time or 0, "8"]}], }, } def activity_events( self, org_id: str, cluster_id: str, start: int, end: int, *, name: str = "", severity: int = 0, limit: int = 100, cursor: str = "", ): return {"activities": [], "partial": False, "cursor": ""} parser = MODULE.build_parser() args = parser.parse_args( [ "topology-snapshot", "--cluster-id", "123", "--at", "2026-06-30T12:00:00+08:00", ] ) result = MODULE.run_topology_snapshot(args, api=MetricErrorSnapshotAPI()) assert result["status"] == "ok" assert result["summary"]["historical_metrics_complete"] is False assert result["summary"]["metric_components_available"] == ["tikv", "tiflash", "pd"] assert result["summary"]["missing_metric_components"] == ["tidb"] assert result["summary"]["error_metric_components"] == ["tidb"] assert result["summary"]["components"][0]["component"] == "tidb" assert result["summary"]["components"][0]["evidence_status"] == "error" assert result["summary"]["components"][0]["historical_metric_available"] is False evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) metric_query = evidence["topology_snapshots"][0]["metric_query"] assert metric_query["historical_metrics_complete"] is False assert metric_query["metric_components_available"] == ["tikv", "tiflash", "pd"] assert metric_query["missing_metric_components"] == ["tidb"] assert metric_query["error_metric_components"] == ["tidb"] def test_run_topology_snapshot_falls_back_to_premium_parent_activity(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) class PremiumSnapshotAPI(FakeAPI): def __init__(self): super().__init__(deploy_type="premium", parent_id="parent-123") self.activity_calls = [] def query_instant(self, cluster_id: str, query: str, time: int | None = None): return { "status": "success", "data": { "resultType": "vector", "result": [{"metric": {"instance": "db-tidb-0"}, "value": [time or 0, "8"]}], }, } def activity_events( self, org_id: str, cluster_id: str, start: int, end: int, *, name: str = "", severity: int = 0, limit: int = 100, cursor: str = "", ): self.activity_calls.append(cluster_id) if cluster_id == "child-456": return {"error": "HTTP 500 query activities failed"} return { "activities": [ { "name": "scalecluster", "displayName": "ScaleCluster", "calibration_time": start + 60, "payload": {"summary": "Scale TiKV from 3 to 6 nodes for child-456"}, } ], "partial": False, "cursor": "", } parser = MODULE.build_parser() args = parser.parse_args( [ "topology-snapshot", "--cluster-id", "child-456", "--at", "1000", ] ) api = PremiumSnapshotAPI() result = MODULE.run_topology_snapshot(args, api=api) assert result["status"] == "ok" assert api.activity_calls == ["child-456", "parent-123"] assert result["summary"]["topology_event_count"] == 1 evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) activity = evidence["topology_snapshots"][0]["activity"] assert activity["fallback_used"] is True assert activity["target_cluster_id"] == "parent-123" assert activity["fallback_reason"] == "premium_activityhub_parent_target" def test_topology_activity_event_requires_action_and_component(): assert MODULE._topology_activity_event( { "displayName": "UpgradeCluster", "payload": {"summary": "Upgrade cluster version"}, } ) is None assert MODULE._topology_activity_event( { "displayName": "ScaleCluster", "payload": {"summary": "Scale TiDB from 2 to 3 nodes"}, } ) is not None def test_run_activity_events_falls_back_to_premium_parent(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) class PremiumActivityAPI(FakeAPI): def __init__(self): super().__init__(deploy_type="premium", parent_id="parent-123") self.activity_calls = [] def activity_events( self, org_id: str, cluster_id: str, start: int, end: int, *, name: str = "", severity: int = 0, limit: int = 100, cursor: str = "", ): self.activity_calls.append(cluster_id) if cluster_id == "child-456": return {"error": "HTTP 500 query activities failed"} return { "activities": [ {"displayName": "ScaleTiProxy", "targetID": cluster_id, "summary": "instance id: child-456"} ], "partial": False, "cursor": "next", } parser = MODULE.build_parser() args = parser.parse_args( [ "activity-events", "--cluster-id", "child-456", "--start", "100", "--end", "200", ] ) api = PremiumActivityAPI() result = MODULE.run_activity_events(args, api=api) assert result["status"] == "ok" assert api.activity_calls == ["child-456", "parent-123"] assert result["summary"]["fallback_used"] is True assert result["summary"]["requested_cluster_id"] == "child-456" assert result["summary"]["target_cluster_id"] == "parent-123" assert result["summary"]["fallback_reason"] == "premium_activityhub_parent_target" evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["target_cluster_id"] == "parent-123" assert evidence["requested_cluster_id"] == "child-456" assert evidence["events"][0]["displayName"] == "ScaleTiProxy" def test_run_continuous_profiling_returns_profile_evidence(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "continuous-profiling", "--cluster-id", "123", "--start", "100", "--end", "200", ] ) result = MODULE.run_continuous_profiling(args, api=FakeAPI()) assert result["status"] == "ok" assert "workspace_evidence" not in result evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["profiles"][0]["state"] == "finished" def test_byoc_continuous_profiling_resolves_physical_inventory_target(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "continuous-profiling", "--cluster-id", "byoc-logical", "--start", "100", "--end", "200", ] ) api = FakeBYOCAPI() result = MODULE.run_continuous_profiling(args, api=api) assert result["status"] == "ok" assert result["summary"]["target_cluster_id"] == "byoc-host-1" assert result["summary"]["inventory_only"] is True assert ("profile-list", "byoc-host-1", 100, 200) in api.byoc_calls def test_byoc_continuous_profiling_splits_ranges_at_api_limit(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "continuous-profiling", "--cluster-id", "byoc-logical", "--start", "100", "--end", "21700", ] ) api = FakeBYOCAPI() result = MODULE.run_continuous_profiling(args, api=api) assert result["status"] == "ok" profile_calls = [call for call in api.byoc_calls if call[0] == "profile-list"] assert profile_calls == [ ("profile-list", "byoc-host-1", 100, 10899), ("profile-list", "byoc-host-1", 10900, 21699), ("profile-list", "byoc-host-1", 21700, 21700), ] def test_byoc_continuous_profiling_download_is_explicitly_unsupported(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "continuous-profiling-download", "--cluster-id", "byoc-logical", "--ts", "100", "--profile-type", "cpu", "--component", "tidb", "--address", "tidb-0:10080", ] ) result = MODULE.run_continuous_profiling_download(args, api=FakeBYOCAPI()) assert result["status"] == "error" assert "inventory-only" in result["error"] def test_run_diagnostic_files_returns_file_evidence(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "diagnostic-files", "--cluster-id", "123", "--file-type", "plan_replayer", "--start", "100", "--end", "200", ] ) result = MODULE.run_diagnostic_files(args, api=FakeAPI()) assert result["status"] == "ok" assert "workspace_evidence" not in result evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert evidence["files"][0]["kind"] == "plan_replayer" def test_byoc_diagnostic_files_are_explicitly_unsupported(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "diagnostic-files", "--cluster-id", "byoc-logical", "--file-type", "plan_replayer", "--start", "100", "--end", "200", ] ) result = MODULE.run_diagnostic_files(args, api=FakeBYOCAPI()) assert result["status"] == "error" assert "not supported for BYOC" in result["error"] def test_run_continuous_profiling_download_writes_workspace_evidence(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "continuous-profiling-download", "--cluster-id", "123", "--ts", "100", "--profile-type", "cpu", "--component", "tidb", "--address", "10.0.0.1:10080", ] ) result = MODULE.run_continuous_profiling_download(args, api=FakeAPI()) assert result["status"] == "ok" assert result["persisted_evidence"]["evidence_file_path"].endswith("profiles.json") assert "payload" not in result evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) profile = evidence["profiles"][0] output_path = Path(profile["file_path"]) assert output_path.exists() assert tmp_path / "ws-test" / "evidences" in output_path.parents assert profile["file_size_bytes"] == len(b"profile-bytes") def test_run_diagnostic_file_download_unzips_under_workspace_evidence(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "diagnostic-file-download", "--cluster-id", "123", "--file-type", "plan_replayer", "--key", "a/b/replayer.zip", "--unzip", ] ) result = MODULE.run_diagnostic_file_download(args, api=FakeAPI()) assert result["status"] == "ok" assert result["persisted_evidence"]["evidence_file_path"].endswith("files.json") assert "payload" not in result evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) file_row = evidence["files"][0] output_path = Path(file_row["file_path"]) extracted = [Path(item) for item in file_row["extracted_files"]] assert output_path.exists() assert tmp_path / "ws-test" / "evidences" in output_path.parents assert len(extracted) == 1 assert extracted[0].name == "plan.json" assert extracted[0].exists() def test_diagnostic_downloads_with_same_filename_use_distinct_paths(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() api = FakeAPI() first = parser.parse_args( [ "diagnostic-file-download", "--cluster-id", "123", "--file-type", "plan_replayer", "--key", "a/b/replayer.zip", "--filename", "replayer.zip", ] ) second = parser.parse_args( [ "diagnostic-file-download", "--cluster-id", "123", "--file-type", "plan_replayer", "--key", "c/d/replayer.zip", "--filename", "replayer.zip", ] ) first_result = MODULE.run_diagnostic_file_download(first, api=api) second_result = MODULE.run_diagnostic_file_download(second, api=api) evidence = MODULE._json_load(Path(second_result["persisted_evidence"]["evidence_file_path"])) paths = {item["file_path"] for item in evidence["files"]} assert first_result["status"] == "ok" assert second_result["status"] == "ok" assert len(paths) == 2 assert all(Path(path).exists() for path in paths) def test_run_metrics_persists_raw_evidence_when_workspace_dir_is_set(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "metrics", "--cluster-id", "123", "--name", "TiDB Total QPS", "--query", "sum(rate(tidb_server_query_total[1m]))", "--start", "100", "--end", "200", ] ) result = MODULE.run_metrics(args, api=FakeAPI()) evidence_path = Path(result["persisted_evidence"]["evidence_file_path"]) report_path = Path(result["persisted_evidence"]["report_path"]) assert result["status"] == "ok" assert "workspace_evidence" not in result assert "payload" not in result assert evidence_path.name == "metrics.json" assert report_path.name == "workspace_evidence_report.json" assert evidence_path.exists() assert report_path.exists() assert '"metrics"' in evidence_path.read_text() def test_run_metrics_keeps_raw_chart_data_out_of_workspace_report(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "metrics", "--cluster-id", "123", "--name", "TiDB Total QPS", "--query", "sum(rate(tidb_server_query_total[1m]))", "--start", "100", "--end", "200", "--inline-chart-data", ] ) result = MODULE.run_metrics(args, api=FakeAPI()) evidence_path = Path(result["persisted_evidence"]["evidence_file_path"]) report_path = Path(result["persisted_evidence"]["report_path"]) raw_metric = MODULE._json_load(evidence_path)["metrics"][0] report_metric = MODULE._json_load(report_path)["evidence"]["metrics"][0] assert raw_metric["chart_data"]["series"] assert report_metric["id"] == raw_metric["id"] assert report_metric["group"] == raw_metric["group"] assert report_metric["ref"] == raw_metric["ref"] assert report_metric["summary"]["point_count"] == raw_metric["summary"]["point_count"] assert report_metric["chart_data_ref"] == raw_metric["chart_data_ref"] assert "chart_data" not in report_metric def test_run_metrics_persists_report_time_range_metadata(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "metrics", "--cluster-id", "123", "--name", "TiDB Total QPS", "--query", "sum(rate(tidb_server_query_total[1m]))", "--start", "100", "--end", "200", "--step", "30s", ] ) result = MODULE.run_metrics(args, api=FakeAPI()) report = MODULE._json_load(Path(result["persisted_evidence"]["report_path"])) assert report["metadata"]["time_range"] == {"start": 100, "end": 200, "step": "30s"} def test_persist_raw_evidence_keeps_established_scope_window(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) first = MODULE._persist_raw_evidence( "logs", "log", "123", {"time_range": {"start": 100, "end": 500}, "logs": []} ) narrower = MODULE._persist_raw_evidence( "logs", "log", "123", {"time_range": {"start": 200, "end": 260}, "logs": []} ) report = MODULE._json_load(Path(narrower["report_path"])) # The investigation scope seeded first must survive a narrower sampling # slice; the slice window is recorded on last_raw_evidence instead. assert report["metadata"]["time_range"] == {"start": 100, "end": 500} assert report["metadata"]["last_raw_evidence"]["time_range"] == {"start": 200, "end": 260} assert first["report_path"] == narrower["report_path"] def test_persist_raw_evidence_reseeds_narrow_window_when_report_scope_is_missing(tmp_path, monkeypatch): workspace_dir = _set_workspace_dir(tmp_path, monkeypatch) first = MODULE._persist_raw_evidence( "metrics", "metric", "123", { "time_range": {"start": 100, "end": 500}, "metrics": [ { "id": "metric-full", "name": "full window", "metric_group_id": "full", "time_window": {"start": 100, "end": 500, "step": "60"}, } ], }, ) report_path = Path(first["report_path"]) report = MODULE._json_load(report_path) report["metadata"].pop("time_range", None) _write_json(report_path, report) narrower = MODULE._persist_raw_evidence( "metrics", "metric", "123", { "time_range": {"start": 200, "end": 260}, "metrics": [ { "id": "metric-narrow", "name": "narrow window", "metric_group_id": "narrow", "time_window": {"start": 200, "end": 260, "step": "60"}, } ], }, ) recovered = MODULE._json_load(Path(narrower["report_path"])) assert recovered["metadata"]["time_range"] == {"start": 100, "end": 500, "step": "60"} assert recovered["metadata"]["last_raw_evidence"]["time_range"] == {"start": 200, "end": 260} assert {item["id"] for item in recovered["evidence"]["metrics"]} == { "metric-full", "metric-narrow", } assert Path(narrower["report_path"]).is_relative_to(workspace_dir) def test_persist_raw_evidence_falls_back_to_current_window_without_recoverable_history(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) report_path = Path(MODULE._workspace_evidence_dir()) / "workspace_evidence_report.json" _write_json( report_path, { "type": "workspace_evidence_report", "metadata": {"cluster_id": "123"}, "analysis": {}, "evidence": {"metrics": [], "queries": [], "logs": [], "events": [], "profiles": [], "files": []}, "links": {}, "gate_checks": {}, }, ) narrower = MODULE._persist_raw_evidence( "metrics", "metric", "123", { "time_range": {"start": 200, "end": 260}, "metrics": [{"id": "metric-narrow", "name": "narrow window", "metric_group_id": "narrow"}], }, ) recovered = MODULE._json_load(Path(narrower["report_path"])) assert recovered["metadata"]["time_range"] == {"start": 200, "end": 260} def test_persist_raw_evidence_recovers_scope_from_raw_log_file_when_report_evidence_has_no_window(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) MODULE._persist_raw_evidence( "logs", "log", "123", { "time_range": {"start": 100, "end": 500}, "log_counts": [{"component": "tikv", "message": "full-window-log", "count": 1}], }, ) report_path = Path(MODULE._workspace_evidence_dir()) / "workspace_evidence_report.json" report = MODULE._json_load(report_path) report["metadata"].pop("time_range", None) report["evidence"]["logs"] = [] _write_json(report_path, report) narrower = MODULE._persist_raw_evidence( "metrics", "metric", "123", { "time_range": {"start": 200, "end": 260}, "metrics": [{"id": "metric-narrow", "name": "narrow window", "metric_group_id": "narrow"}], }, ) recovered = MODULE._json_load(Path(narrower["report_path"])) assert recovered["metadata"]["time_range"] == {"start": 100, "end": 500} def test_persist_raw_evidence_serializes_concurrent_workspace_writes(tmp_path, monkeypatch): workspace_dir = _set_workspace_dir(tmp_path, monkeypatch) metric_calls = [ ( "metrics", "metric", "123", { "metrics": [ { "id": f"metric-{index}", "name": f"metric {index}", "metric_group_id": f"group-{index}", } ] }, ) for index in range(16) ] log_calls = [ ( "logs", "log", "123", { "log_counts": [ { "component": "tidb", "level": "ERROR", "message": f"error pattern {index}", "count": 1, } ] }, ) for index in range(8) ] _run_concurrent_persist_processes(tmp_path, workspace_dir, [*metric_calls, *log_calls]) evidence_dir = workspace_dir / "evidences" raw_metrics = MODULE._json_load(evidence_dir / "metrics.json")["metrics"] raw_logs = MODULE._json_load(evidence_dir / "logs.json")["log_counts"] report = MODULE._json_load(evidence_dir / "workspace_evidence_report.json") assert {item["id"] for item in raw_metrics} == {f"metric-{index}" for index in range(16)} assert {item["message"] for item in raw_logs} == {f"error pattern {index}" for index in range(8)} assert {item["id"] for item in report["evidence"]["metrics"]} == { f"metric-{index}" for index in range(16) } assert {item["message"] for item in report["evidence"]["logs"]} == { f"error pattern {index}" for index in range(8) } def test_run_cluster_metadata_persists_meta_when_workspace_dir_is_set(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args(["cluster-metadata", "--cluster-id", "123", "--include-components"]) result = MODULE.run_cluster_metadata(args, api=FakeAPI()) evidence_path = Path(result["persisted_evidence"]["evidence_file_path"]) assert result["status"] == "ok" assert "payload" not in result assert evidence_path.name == "cluster_meta.json" assert evidence_path.exists() def test_data_command_requires_workspace_dir(monkeypatch): monkeypatch.delenv("WORKSPACE_EVIDENCE_DIR", raising=False) monkeypatch.delenv("INVESTIBOOK_WORKSPACE_DIR", raising=False) parser = MODULE.build_parser() args = parser.parse_args( [ "metrics", "--cluster-id", "123", "--name", "TiDB Total QPS", "--query", "sum(rate(tidb_server_query_total[1m]))", "--start", "100", "--end", "200", ] ) try: MODULE.run_metrics(args, api=FakeAPI()) except RuntimeError as exc: assert "INVESTIBOOK_WORKSPACE_DIR" in str(exc) else: raise AssertionError("expected missing INVESTIBOOK_WORKSPACE_DIR to fail") def test_workspace_root_ignores_agent_supplied_issue_evidence_dir(tmp_path, monkeypatch): workspace_dir = tmp_path / "ws-test" issue_dir = workspace_dir / "issues" / "O11Y-1" monkeypatch.setenv("INVESTIBOOK_WORKSPACE_DIR", str(workspace_dir)) monkeypatch.setenv("WORKSPACE_EVIDENCE_DIR", str(issue_dir)) parser = MODULE.build_parser() args = parser.parse_args( [ "metrics", "--cluster-id", "123", "--name", "TiDB Total QPS", "--query", "sum(rate(tidb_server_query_total[1m]))", "--start", "100", "--end", "200", ] ) result = MODULE.run_metrics(args, api=FakeAPI()) evidence_path = Path(result["persisted_evidence"]["evidence_file_path"]) assert evidence_path == workspace_dir / "evidences" / "metrics.json" assert not (issue_dir / "metrics.json").exists() def test_ad_hoc_slow_query_top_uses_date_partitions_and_epoch_time(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "ad-hoc", "slow-query-top", "--cluster-id", "123", "--start", "2026-04-22T06:24:00Z", "--end", "2026-04-22T07:24:00Z", "--order-by", "total_process_time", ] ) api = FakeAPI() result = MODULE.run_ad_hoc_slow_query_top(args, api=api) assert result["status"] == "ok" sql = api.sql_queries[0] assert "date IN ('20260422')" in sql assert "time >= 1776839040" in sql assert "time <= 1776842640" in sql assert "ORDER BY total_process_time DESC" in sql def test_ad_hoc_ttl_rows_uses_server_ttl_counters(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "ad-hoc", "ttl-rows", "--cluster-id", "123", "--start", "100", "--end", "200", ] ) api = FakeAPI() result = MODULE.run_ad_hoc_ttl_rows(args, api=api) assert result["status"] == "ok" assert result["summary"]["evidence_sufficient"] is True assert result["summary"]["ttl_activity_present"] is True queries = [item["query"] for item in api.instant_queries] assert "sum(increase(tidb_server_ttl_insert_rows[100s]))" in queries assert "sum(increase(tidb_server_ttl_processed_expired_rows[100s]))" in queries evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert {item["metric_group_id"] for item in evidence["metrics"]} == { "ttl_insert_rows_delta", "ttl_processed_expired_rows_delta", } def test_ad_hoc_topsql_links_builds_links_from_read_iops(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "ad-hoc", "topsql-links", "--cluster-id", "123", "--start", "100", "--end", "200", "--top-n", "2", ] ) api = FakeAPI() result = MODULE.run_ad_hoc_topsql_links(args, api=api) assert result["status"] == "ok" assert result["summary"]["evidence_sufficient"] is True assert [item["tikv_node"] for item in result["summary"]["topsql_links"]] == ["db-tikv-1", "db-tikv-2"] assert "node_disk_reads_completed_total" in api.range_join_queries[0]["query"] assert "node_disk_writes_completed_total" not in api.range_join_queries[0]["query"] assert "instance=db-tikv-1.db-tikv-peer.tidb123.svc%3A20160" in result["summary"]["topsql_links"][0]["url"] class P99FakeAPI(FakeAPI): def query_range(self, cluster_id: str, query: str, start: int, end: int, step: str = "60"): if "tidb_server_handle_query_duration_seconds_bucket" in query: rows = [ {"metric": {"instance": "db-tidb-0"}, "values": [[start, "0.02"], [end, "0.04"]]}, {"metric": {"instance": "db-tidb-1"}, "values": [[start, "0.01"], [end, "0.02"]]}, ] elif "tikv_grpc_msg_duration_seconds_bucket" in query: rows = [ {"metric": {"instance": "db-tikv-0"}, "values": [[start, "0.007"], [end, "0.009"]]}, ] elif "pd_server_handle_tso_duration_seconds_bucket" in query: rows = [ {"metric": {"instance": "db-pd-0"}, "values": [[start, "0.005"], [end, "0.007"]]}, ] else: rows = [ {"metric": {"instance": "db-tidb-0"}, "values": [[start, "0.003"], [end, "0.005"]]}, ] return {"status": "success", "data": {"resultType": "matrix", "result": rows}} def test_ad_hoc_tidb_p99_breakdown_returns_required_summary(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "ad-hoc", "tidb-p99-breakdown", "--cluster-id", "123", "--start", "100", "--end", "200", "--top-n", "1", ] ) result = MODULE.run_ad_hoc_tidb_p99_breakdown(args, api=P99FakeAPI()) assert result["status"] == "ok" assert result["summary"]["evidence_sufficient"] is True assert result["summary"]["top_nodes_by_avg_p99_latency"][0]["node"] == "db-tidb-0" assert result["summary"]["top_nodes_by_avg_p99_latency"][0]["avg_gc_stw_pause_p99_s"] == 0.004 assert result["summary"]["cluster_avg_tikv_grpc_request_handle_p99_s"] == 0.008 evidence = MODULE._json_load(Path(result["persisted_evidence"]["evidence_file_path"])) assert "tidb_query_p99_latency" in {item["metric_group_id"] for item in evidence["metrics"]} class CPUOverloadFakeAPI(FakeAPI): def query_range(self, cluster_id: str, query: str, start: int, end: int, step: str = "60"): def rows(component: str, values: list[tuple[str, str]]): return [ {"metric": {"pod": pod, "instance": pod}, "values": [[start, value], [end, value]]} for pod, value in values ] if "process_cpu_seconds_total" in query and 'component="tidb"' in query: result = rows("tidb", [("db-tidb-1", "8"), ("db-tidb-2", "4")]) elif "process_cpu_seconds_total" in query and 'component="tikv"' in query: result = rows("tikv", [("db-tikv-3", "16"), ("db-tikv-4", "8")]) elif "process_cpu_seconds_total" in query and 'component="pd"' in query: result = rows("pd", [("db-pd-0", "3"), ("db-pd-1", "1")]) elif "tidb_server_maxprocs" in query: result = rows("tidb", [("db-tidb-1", "16"), ("db-tidb-2", "16")]) elif "tikv_server_cpu_cores_quota" in query: result = rows("tikv", [("db-tikv-3", "32"), ("db-tikv-4", "32")]) elif "pd_service_maxprocs" in query: result = rows("pd", [("db-pd-0", "8"), ("db-pd-1", "8")]) else: result = [] return {"status": "success", "data": {"resultType": "matrix", "result": result}} def loki_query_range( self, cluster_id: str, query: str, start_ns: int, end_ns: int, *, limit: int = 200, direction: str = "backward", ): node = "" for candidate in ("db-tidb-1", "db-tikv-3", "db-pd-0"): if candidate in query: node = candidate break values = [] if node: values = [[str(start_ns), f"2026-04-27T12:00:00Z [WARN] {node} test warning"]] return { "status": "success", "data": { "resultType": "streams", "result": [{"stream": {"instance": node}, "values": values}], }, } def test_ad_hoc_cpu_overload_logs_ranks_components_and_counts_logs(tmp_path, monkeypatch): _set_workspace_dir(tmp_path, monkeypatch) parser = MODULE.build_parser() args = parser.parse_args( [ "ad-hoc", "cpu-overload-logs", "--cluster-id", "123", "--start", "100", "--end", "200", "--top-n-per-component", "1", ] ) result = MODULE.run_ad_hoc_cpu_overload_logs(args, api=CPUOverloadFakeAPI()) assert result["status"] == "ok" assert result["summary"]["evidence_sufficient"] is True assert [ (item["component"], item["node"], item["error_warn_log_count"]) for item in result["summary"]["ranked_nodes"] ] == [ ("tidb", "db-tidb-1", 2), ("tikv", "db-tikv-3", 2), ("pd", "db-pd-0", 2), ] metrics_path = Path(result["persisted_evidence"]["metrics"]["evidence_file_path"]) logs_path = Path(result["persisted_evidence"]["logs"]["evidence_file_path"]) assert metrics_path.name == "metrics.json" assert logs_path.name == "logs.json" def test_clinic_api_log_stats_helpers_build_log_stats_sql(): api = API_MODULE.ClinicAPI.__new__(API_MODULE.ClinicAPI) captured = {} def fake_query_sql(cluster_id: str, sql: str, timeout: int = 60): captured["cluster_id"] = cluster_id captured["sql"] = sql captured["timeout"] = timeout return { "columns": ["level", "component", "file", "line", "msg_digest", "sample", "count"], "rows": [ ["WARN", "tidb", "sqlkiller.go", "61", "can't connect", "sample", "2"], ["WARN", "tidb", "sqlkiller.go", "61", "can't connect", "sample", "3"], ], } api.query_sql = fake_query_sql result = api.log_stats_search("c1", "20260618", "can't connect", limit=7, timeout=33) assert result == { "columns": ["level", "component", "file", "line", "msg_digest", "sample", "total_count"], "rows": [["WARN", "tidb", "sqlkiller.go", "61", "can't connect", "sample", 5]], } assert captured["cluster_id"] == "c1" assert captured["timeout"] == 33 assert "FROM log_stats" in captured["sql"] assert "WHERE date = 20260618" in captured["sql"] assert "LIKE '%can''t connect%'" in captured["sql"] assert "LIMIT 1000" in captured["sql"] def test_clinic_api_log_stats_helpers_reject_invalid_date(): api = API_MODULE.ClinicAPI.__new__(API_MODULE.ClinicAPI) try: api.log_stats_top("c1", "2026-06-18") except ValueError as exc: assert "YYYYMMDD" in str(exc) else: raise AssertionError("invalid LogStats date should fail fast")