from __future__ import annotations import importlib.util import urllib.parse from pathlib import Path SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "clinic_api.py" SPEC = importlib.util.spec_from_file_location("builtin_clinic_api_byoc_contracts", SCRIPT_PATH) MODULE = importlib.util.module_from_spec(SPEC) assert SPEC and SPEC.loader SPEC.loader.exec_module(MODULE) def _encoded(params: dict) -> dict[str, list[str]]: return urllib.parse.parse_qs( urllib.parse.urlencode(params, doseq=True), keep_blank_values=True, ) class RecordingAPI(MODULE.ClinicAPI): def __init__(self): self.endpoint = "https://clinic.test" self.api_base = f"{self.endpoint}/clinic/api/v1" self.api_key = "test-key" self._canonical_org_ids = {} self.metadata: dict[str, dict] = {} self.calls: list[dict] = [] def register(self, cluster_id: str, metadata: dict | None = None) -> None: self.metadata[cluster_id] = metadata or {"orgId": "42", "deployType": "byoc"} def get_cluster_by_id(self, cluster_id: str) -> dict: return self.metadata[cluster_id] def _get(self, path: str, params: dict | None = None, extra_headers: dict | None = None) -> dict: self.calls.append({"path": path, "params": params or {}, "headers": extra_headers or {}}) return {"status": "success", "data": []} def take(self) -> dict: assert self.calls return self.calls.pop(0) def _expected_path(cluster_id: str, suffix: str) -> str: encoded = urllib.parse.quote(cluster_id, safe="") return f"/data-proxy/api/v1/orgs/42/clusters/{encoded}/{suffix}" def test_canonical_metrics_and_logs_accept_opaque_byoc_ids(): api = RecordingAPI() for cluster_id in ("logical/byoc host", "physical/byoc host"): api.register(cluster_id) api.query_range(cluster_id, "up", 1, 2, "30") call = api.take() assert call["path"] == _expected_path(cluster_id, "metrics/query-range") assert _encoded(call["params"]) == { "query": ["up"], "start": ["1"], "end": ["2"], "step": ["30"] } api.logs_query_range(cluster_id, '{container="tidb"}', 1, 2, 7, "forward") call = api.take() assert call["path"] == _expected_path(cluster_id, "logs/query-range") assert _encoded(call["params"]) == { "query": ['{container="tidb"}'], "start": ["1"], "end": ["2"], "limit": ["7"], "direction": ["forward"], } def test_canonical_log_metric_query_uses_step_without_stream_parameters(): api = RecordingAPI() cluster_id = "logical/byoc host" api.register(cluster_id) api.logs_query_range( cluster_id, 'sum(count_over_time({container="tidb"}[300s]))', 1, 2, limit=None, direction=None, step="300", ) call = api.take() assert call["path"] == _expected_path(cluster_id, "logs/query-range") assert _encoded(call["params"]) == { "query": ['sum(count_over_time({container="tidb"}[300s]))'], "start": ["1"], "end": ["2"], "step": ["300"], } def test_byoc_diagnostic_routes_and_default_parameters(): api = RecordingAPI() cluster_id = "logical/byoc host" api.register(cluster_id) cases = [ ( lambda: api.byoc_slow_query_available_fields(cluster_id), "slow-queries/available-fields", {}, ), ( lambda: api.byoc_slow_query_list(cluster_id, 1, 2), "slow-queries", { "begin_time": ["1"], "end_time": ["2"], "limit": ["100"], "orderBy": ["timestamp"], "desc": ["false"], "fields": ["query,timestamp,query_time,memory_max"], "show_internal": ["false"], }, ), ( lambda: api.byoc_slow_query_detail(cluster_id, "digest", "1.25", "99"), "slow-queries/detail", { "digest": ["digest"], "timestamp": ["1.25"], "connect_id": ["99"], "show_internal": ["false"], }, ), ( lambda: api.byoc_top_sql_instances(cluster_id, 1, 2), "top-sql/instances", {"start": ["1"], "end": ["2"]}, ), ( lambda: api.byoc_top_sql_summary(cluster_id, 1, 2, "1m", 10, "tidb-0:4000", "tidb"), "top-sql/summary", { "start": ["1"], "end": ["2"], "window": ["1m"], "top": ["10"], "instance": ["tidb-0:4000"], "instance_type": ["tidb"], }, ), ( lambda: api.byoc_statements_available_fields(cluster_id), "statements/available-fields", {}, ), ( lambda: api.byoc_statements_statement_types(cluster_id), "statements/statement-types", {}, ), ( lambda: api.byoc_statements_list(cluster_id, 1, 2), "statements", {"begin_time": ["1"], "end_time": ["2"]}, ), ( lambda: api.byoc_statements_plans(cluster_id, 1, 2, "digest"), "statements/plans", {"begin_time": ["1"], "end_time": ["2"], "digest": ["digest"]}, ), ( lambda: api.byoc_statements_plan_detail(cluster_id, 1, 2, "digest"), "statements/plan-detail", {"begin_time": ["1"], "end_time": ["2"], "digest": ["digest"]}, ), ] for invoke, suffix, expected_params in cases: invoke() call = api.take() assert call["path"] == _expected_path(cluster_id, suffix) assert _encoded(call["params"]) == expected_params def test_repeated_byoc_filters_use_repeated_query_parameters(): api = RecordingAPI() cluster_id = "logical/byoc host" api.register(cluster_id) api.byoc_slow_query_list( cluster_id, 1, 2, db=["sales & ops", "db+two"], plans=["plan+one", "plan/two"], ) assert _encoded(api.take()["params"])["db"] == ["sales & ops", "db+two"] api.byoc_statements_list( cluster_id, 1, 2, schemas=["db one", "db&two"], stmt_types=["Select", "Insert"], ) params = _encoded(api.take()["params"]) assert params["schemas"] == ["db one", "db&two"] assert params["stmt_types"] == ["Select", "Insert"] def test_byoc_physical_id_and_namespace_resolution(): api = RecordingAPI() api.register( "logical", { "orgId": "42", "deployType": "byoc", "diagnosticContext": {"parentId": "physical"}, }, ) api.register("physical", {"orgId": "42", "deployType": "byoc-host"}) assert api.byoc_physical_cluster_id("logical") == "physical" assert api.byoc_data_plane_namespace("logical") == "tidbphysical" assert api.byoc_physical_cluster_id("physical") == "physical" assert api.byoc_data_plane_namespace("physical") == "tidbphysical" def test_get_encodes_repeated_values_with_doseq(): api = object.__new__(MODULE.ClinicAPI) api.endpoint = "https://clinic.test" api.api_key = "test-key" captured = {} api._do = lambda request: captured.setdefault("url", request.full_url) or {} api._get("/test", {"plans": ["one", "two"]}) assert urllib.parse.parse_qs(urllib.parse.urlsplit(captured["url"]).query) == { "plans": ["one", "two"] }