1597 lines
95 KiB
Python
1597 lines
95 KiB
Python
#!/usr/bin/env python3
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare [C:4] [TYPE Script] [SEMANTICS api,scenario,catalog,chart-identity]
|
|
# @PURPOSE Create the ordinary owned scenario and bind an explicitly selected catalog head for native chart capture.
|
|
# @PRE The Git release is published and validated in PREPROD; the fixture seed, release report, target descriptor, and private env are explicit.
|
|
# @POST A server-owned scenario and explicit catalog selector are retained for native browser capture.
|
|
# @INVARIANT Dashboard and chart identities resolve through current Superset/API observations; no catalog entry or result is synthesized.
|
|
# @REJECTED Reading repository IDs from the database or assuming a target dashboard ID from DEV was rejected.
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode, urlsplit
|
|
from uuid import UUID, uuid4
|
|
|
|
import httpx
|
|
|
|
_workspace_root = Path(__file__).resolve().parents[2]
|
|
_backend_candidates = (Path("/app/backend"), _workspace_root / "backend")
|
|
BACKEND_ROOT = next((path for path in _backend_candidates
|
|
if (path / "src/schemas/dashboard_testing/native_capture.py").is_file()), None)
|
|
if BACKEND_ROOT is None:
|
|
raise RuntimeError("NATIVE_BACKEND_PACKAGE_UNAVAILABLE")
|
|
sys.path.insert(0, str(BACKEND_ROOT))
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "full_flow"))
|
|
from verify import Driver # noqa: E402
|
|
from common import Blocked # noqa: E402
|
|
from mcp_client import McpClient # noqa: E402
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "full_flow_finance"))
|
|
from release import private_environment # noqa: E402
|
|
|
|
|
|
class SafeNativeDriver(Driver):
|
|
"""Keep this flow's persisted HTTP evidence to bounded identity/status facts."""
|
|
|
|
_SAFE_FACTS = {
|
|
"http_status", "method", "path", "content_type", "error_code", "reason",
|
|
"dashboard_id", "source_dashboard_id", "target_dashboard_id", "chart_id", "dataset_id",
|
|
"scenario_id", "revision_id", "catalog_id", "catalog_revision_id", "release_id",
|
|
"operation_id", "commit_sha", "query_model_fingerprint", "filters_hash",
|
|
"candidate_count", "artifact_count", "control_count", "key_present",
|
|
"publication_state", "state", "capture_count", "environment_id", "filter_count",
|
|
"expected_filter_count", "coordinate_count", "filter_ids_match", "filters_hash_match",
|
|
"context_present", "context_error_present", "context_schema_version_match",
|
|
"pass_run_id", "failure_run_id", "restored_pass_run_id", "restore_verified",
|
|
}
|
|
|
|
def record(self, name: str, status: str, **facts) -> None:
|
|
safe: dict[str, object] = {}
|
|
for key, value in facts.items():
|
|
if key not in self._SAFE_FACTS:
|
|
continue
|
|
if key in {"http_status", "dashboard_id", "source_dashboard_id", "target_dashboard_id",
|
|
"chart_id", "dataset_id", "candidate_count", "artifact_count", "control_count",
|
|
"filter_count", "expected_filter_count", "coordinate_count"}:
|
|
if type(value) is int:
|
|
safe[key] = value
|
|
elif key in {"key_present", "filter_ids_match", "filters_hash_match",
|
|
"context_present", "context_error_present", "context_schema_version_match",
|
|
"restore_verified"}:
|
|
if type(value) is bool:
|
|
safe[key] = value
|
|
elif key == "method":
|
|
if value in {"GET", "POST", "PUT", "DELETE"}:
|
|
safe[key] = value
|
|
elif key == "path":
|
|
if type(value) is str and value.startswith("/") and "?" not in value:
|
|
safe[key] = value[:240]
|
|
elif key == "content_type":
|
|
if type(value) is str and re.fullmatch(r"[A-Za-z0-9.+-]+/[A-Za-z0-9.+-]+", value):
|
|
safe[key] = value
|
|
elif key in {"error_code", "reason"}:
|
|
if type(value) is str and re.fullmatch(r"[A-Z][A-Z0-9_]{1,95}", value):
|
|
safe[key] = value
|
|
elif key in {"query_model_fingerprint", "filters_hash"}:
|
|
if type(value) is str and re.fullmatch(r"sha256:[a-f0-9]{64}", value):
|
|
safe[key] = value
|
|
elif key == "commit_sha":
|
|
if type(value) is str and re.fullmatch(r"[a-f0-9]{40,64}", value):
|
|
safe[key] = value
|
|
elif key == "environment_id":
|
|
if type(value) is str and re.fullmatch(r"[A-Za-z0-9_.-]{1,100}", value):
|
|
safe[key] = value
|
|
elif key in {"scenario_id", "revision_id", "catalog_id", "catalog_revision_id",
|
|
"release_id", "operation_id"}:
|
|
if type(value) is str and re.fullmatch(r"[A-Za-z0-9_.:-]{1,120}", value):
|
|
safe[key] = value
|
|
elif key in {"pass_run_id", "failure_run_id", "restored_pass_run_id"}:
|
|
if type(value) is str and re.fullmatch(r"[0-9a-f-]{36}", value):
|
|
safe[key] = value
|
|
elif key in {"publication_state", "state", "capture_count"}:
|
|
if type(value) in {str, int} and len(str(value)) <= 64:
|
|
safe[key] = value
|
|
super().record(name, status, **safe)
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.CurrentIdentity [C:3] [TYPE Function]
|
|
# @POST The PREPROD dashboard ID is resolved by its seeded slug/UUID, then its current model confirms the same chart UUID and dataset semantics.
|
|
def validate_target_descriptor(target: dict, seed: dict, release: dict) -> None:
|
|
if not isinstance(target, dict) or not isinstance(seed, dict) or not isinstance(release, dict):
|
|
raise Blocked("NATIVE_TARGET_DESCRIPTOR_INVALID")
|
|
required = ("target_mode", "backend_url", "source_environment_id", "target_environment_id", "target_stage",
|
|
"superset_url", "source_dashboard_id", "source_dashboard_uuid",
|
|
"target_dashboard_uuid", "chart_uuid", "catalog_mode", "backend_username",
|
|
"superset_username", "backend_container", "chart_contract", "capture_scenario",
|
|
"reference_url", "reference_origin", "filter_context_path",
|
|
"catalog_repository", "catalog_branch", "expected_filter_context", "expected_filters_hash")
|
|
if any(key not in target or target[key] is None for key in required):
|
|
raise Blocked("NATIVE_TARGET_DESCRIPTOR_INCOMPLETE")
|
|
for key in ("backend_url", "superset_url"):
|
|
parsed = urlsplit(target[key])
|
|
if (parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password
|
|
or parsed.query or parsed.fragment or parsed.path not in {"", "/"}):
|
|
raise Blocked("NATIVE_TARGET_ENDPOINT_INVALID")
|
|
if urlsplit(target["backend_url"]).hostname not in {"127.0.0.1", "localhost"}:
|
|
raise Blocked("NATIVE_BACKEND_ENDPOINT_NOT_LOCAL")
|
|
if urlsplit(target["superset_url"]).hostname not in {
|
|
"127.0.0.1", "localhost", "superset-dev", "superset-preprod", "superset-prod"
|
|
}:
|
|
raise Blocked("NATIVE_SUPERSET_ENDPOINT_NOT_ALLOWED")
|
|
for key in ("source_environment_id", "target_environment_id", "target_stage",
|
|
"source_dashboard_uuid", "target_dashboard_uuid", "chart_uuid",
|
|
"backend_username", "superset_username", "backend_container", "reference_url"):
|
|
if type(target.get(key)) is not str or not target[key].strip():
|
|
raise Blocked("NATIVE_TARGET_IDENTITY_INVALID")
|
|
for key in ("reference_origin", "catalog_repository", "catalog_branch"):
|
|
if type(target.get(key)) is not str or not target[key].strip():
|
|
raise Blocked("NATIVE_TARGET_IDENTITY_INVALID")
|
|
if type(target.get("filter_context_path")) is not str or not target["filter_context_path"].strip():
|
|
raise Blocked("NATIVE_TARGET_IDENTITY_INVALID")
|
|
if target["target_stage"] != "preprod":
|
|
raise Blocked("NATIVE_TARGET_STAGE_UNSUPPORTED")
|
|
if target["target_mode"] not in {"clean_fixture", "lab"}:
|
|
raise Blocked("NATIVE_TARGET_MODE_UNSUPPORTED")
|
|
capture_scenario = target["capture_scenario"]
|
|
if not isinstance(capture_scenario, dict):
|
|
raise Blocked("NATIVE_CAPTURE_SCENARIO_REQUIRED")
|
|
validate_lab_bootstrap_mode(target["target_mode"], target["catalog_mode"], capture_scenario.get("mode"))
|
|
if capture_scenario.get("mode") == "metric_baseline":
|
|
metric_name = capture_scenario.get("metric_name")
|
|
if type(metric_name) is not str or not metric_name.strip():
|
|
raise Blocked("NATIVE_M01_METRIC_IDENTITY_REQUIRED")
|
|
if metric_name not in target["chart_contract"]["metric_names"]:
|
|
raise Blocked("NATIVE_M01_METRIC_OUTSIDE_CHART_CONTRACT")
|
|
selection = target.get("catalog_selection")
|
|
if (target.get("catalog_mode") != "exact_revision" or not isinstance(selection, dict)
|
|
or selection.get("kind") != "revision"
|
|
or not selection.get("catalog_id") or not selection.get("catalog_revision_id")):
|
|
raise Blocked("NATIVE_M01_EXACT_CATALOG_REVISION_REQUIRED")
|
|
elif capture_scenario.get("mode") in {"compile_structural", "mcp_bootstrap"}:
|
|
cases = capture_scenario.get("selected_case_ids")
|
|
if (not isinstance(cases, list) or not cases
|
|
or any(type(case) is not str or not case.strip() for case in cases)):
|
|
raise Blocked("NATIVE_CAPTURE_SCENARIO_CASES_REQUIRED")
|
|
elif capture_scenario.get("mode") == "existing_mcp_revision":
|
|
validate_existing_mcp_revision(capture_scenario)
|
|
validate_published_resume_target(target)
|
|
else:
|
|
raise Blocked("NATIVE_CAPTURE_SCENARIO_MODE_UNSUPPORTED")
|
|
numeric = target.get("numeric_verification")
|
|
numeric_fields = {"query_index", "row_key", "snapshot_date", "filter_from", "filter_to",
|
|
"filter_inclusive", "params", "evaluation_recipe_id", "evaluation_provider_id",
|
|
"pin_mode", "finance_fixture_seed_path", "mutator_container"}
|
|
if (not isinstance(numeric, dict)
|
|
or set(numeric) != numeric_fields
|
|
and set(numeric) != numeric_fields | {"expected_pin_sha256"}):
|
|
raise Blocked("NATIVE_NUMERIC_VERIFICATION_TARGET_INCOMPLETE")
|
|
existing_pin = numeric.get("pin_mode") == "existing"
|
|
fresh_pin = numeric.get("pin_mode") == "fresh_published"
|
|
if (type(numeric.get("query_index")) is not int or numeric["query_index"] < 0
|
|
or type(numeric.get("row_key")) is not str or not numeric["row_key"].strip()
|
|
or any(type(numeric.get(key)) is not str or not numeric[key].strip()
|
|
for key in ("snapshot_date", "filter_from", "filter_to"))
|
|
or type(numeric.get("filter_inclusive")) is not bool
|
|
or numeric["filter_inclusive"] is not False
|
|
or numeric.get("params") != {}
|
|
or numeric.get("evaluation_recipe_id") != "visual_single_attempt_v1"
|
|
or type(numeric.get("evaluation_provider_id")) is not str
|
|
or not numeric["evaluation_provider_id"].strip()
|
|
or not (existing_pin or fresh_pin)
|
|
or (existing_pin and (type(numeric.get("expected_pin_sha256")) is not str
|
|
or re.fullmatch(r"[a-f0-9]{64}", numeric["expected_pin_sha256"]) is None))
|
|
or (fresh_pin and "expected_pin_sha256" in numeric)
|
|
or type(numeric.get("finance_fixture_seed_path")) is not str
|
|
or not Path(numeric["finance_fixture_seed_path"]).is_absolute()
|
|
or type(numeric.get("mutator_container")) is not str
|
|
or re.fullmatch(r"[A-Za-z0-9_.-]{1,128}", numeric["mutator_container"]) is None):
|
|
raise Blocked("NATIVE_NUMERIC_VERIFICATION_TARGET_INVALID")
|
|
if (numeric["snapshot_date"] != "2026-09-30"
|
|
or numeric["filter_from"] != "2026-09-30T00:00:00"
|
|
or numeric["filter_to"] != "2026-10-01T00:00:00"):
|
|
raise Blocked("NATIVE_NUMERIC_VERIFICATION_DATE_SCOPE_UNSUPPORTED")
|
|
filter_context = target.get("expected_filter_context")
|
|
if not isinstance(filter_context, list):
|
|
raise Blocked("NATIVE_EXPECTED_FILTER_CONTEXT_REQUIRED")
|
|
if (type(target.get("expected_filters_hash")) is not str
|
|
or not re.fullmatch(r"sha256:[a-f0-9]{64}", target["expected_filters_hash"])):
|
|
raise Blocked("NATIVE_EXPECTED_FILTER_HASH_INVALID")
|
|
if type(target.get("source_dashboard_id")) is not int or target["source_dashboard_id"] <= 0:
|
|
raise Blocked("NATIVE_SOURCE_DASHBOARD_ID_INVALID")
|
|
if target["source_dashboard_id"] != seed.get("dashboard_id"):
|
|
raise Blocked("NATIVE_SOURCE_DASHBOARD_ID_MISMATCH")
|
|
if target["source_dashboard_uuid"] != seed.get("dashboard_uuid"):
|
|
raise Blocked("NATIVE_SOURCE_DASHBOARD_UUID_MISMATCH")
|
|
if target["chart_uuid"] != seed.get("acceptance_chart", {}).get("chart_uuid"):
|
|
raise Blocked("NATIVE_SOURCE_CHART_UUID_MISMATCH")
|
|
contract = target.get("chart_contract")
|
|
if not isinstance(contract, dict):
|
|
raise Blocked("NATIVE_CHART_CONTRACT_INVALID")
|
|
for key in ("dataset_uuid", "viz_type", "row_key_column"):
|
|
if type(contract.get(key)) is not str or not contract[key].strip():
|
|
raise Blocked("NATIVE_CHART_CONTRACT_INVALID")
|
|
for key in ("group_by_columns", "key_columns", "metric_names"):
|
|
values = contract.get(key)
|
|
if (not isinstance(values, list) or not values
|
|
or any(type(value) is not str or not value.strip() for value in values)
|
|
or len(set(values)) != len(values)):
|
|
raise Blocked("NATIVE_CHART_CONTRACT_INVALID")
|
|
if contract["row_key_column"] not in contract["group_by_columns"]:
|
|
raise Blocked("NATIVE_CHART_CONTRACT_INVALID")
|
|
policy_intent = contract.get("comparison_policy_intent")
|
|
if not isinstance(policy_intent, dict):
|
|
raise Blocked("NATIVE_CHART_POLICY_INTENT_REQUIRED")
|
|
try:
|
|
from src.schemas.dashboard_testing.native_capture import ChartResponsePolicyIntent
|
|
|
|
typed_policy = ChartResponsePolicyIntent.model_validate(policy_intent)
|
|
except Exception as exc:
|
|
raise Blocked("NATIVE_CHART_POLICY_INTENT_INVALID") from exc
|
|
expected_columns = set(contract["key_columns"]) | set(contract["metric_names"])
|
|
if any(set(query.per_column) != expected_columns or query.key_columns != contract["key_columns"]
|
|
for query in typed_policy.query_policies):
|
|
raise Blocked("NATIVE_CHART_POLICY_SCOPE_MISMATCH")
|
|
release_identity = target.get("release_identity")
|
|
if not isinstance(release_identity, dict) or any(
|
|
not release_identity.get(key) for key in ("id", "version", "commit_hash", "content_hash")
|
|
):
|
|
raise Blocked("NATIVE_RELEASE_IDENTITY_MISSING")
|
|
for key in ("id", "version", "commit_hash", "content_hash"):
|
|
if release.get(key) != release_identity.get(key):
|
|
raise Blocked("NATIVE_RELEASE_IDENTITY_MISMATCH")
|
|
release_admitted = release.get("status") == "published" or (
|
|
release.get("status") == "ready_to_publish"
|
|
and bool(release.get("approved_at"))
|
|
and bool(release.get("approved_by"))
|
|
)
|
|
if not release_admitted:
|
|
raise Blocked("NATIVE_RELEASE_APPROVAL_UNAVAILABLE")
|
|
if target["catalog_mode"] == "exact_revision":
|
|
selection = target.get("catalog_selection")
|
|
if (not isinstance(selection, dict) or selection.get("kind") != "revision"
|
|
or any(not selection.get(key) for key in ("catalog_id", "catalog_revision_id"))):
|
|
raise Blocked("NATIVE_CATALOG_SELECTION_REQUIRED")
|
|
elif target["catalog_mode"] == "current_head":
|
|
selection = target.get("catalog_selection")
|
|
if (not isinstance(selection, dict) or selection.get("kind") != "current"
|
|
or not selection.get("catalog_id")):
|
|
raise Blocked("NATIVE_CATALOG_SELECTION_REQUIRED")
|
|
elif target["catalog_mode"] == "seed_empty":
|
|
pass
|
|
else:
|
|
raise Blocked("NATIVE_CATALOG_MODE_UNSUPPORTED")
|
|
|
|
|
|
def validate_lab_bootstrap_mode(target_mode: str, catalog_mode: str, scenario_mode: str) -> None:
|
|
"""Require fresh lab bootstrap or an explicit exact-revision resume."""
|
|
if target_mode != "lab":
|
|
return
|
|
if (catalog_mode, scenario_mode) not in {
|
|
("seed_empty", "mcp_bootstrap"),
|
|
("exact_revision", "existing_mcp_revision"),
|
|
}:
|
|
raise Blocked("NATIVE_LAB_EXPLICIT_MCP_MODE_REQUIRED")
|
|
|
|
|
|
def validate_existing_mcp_revision(selection: dict) -> None:
|
|
if (not isinstance(selection, dict)
|
|
or not _is_uuid(selection.get("scenario_id"))
|
|
or not _is_uuid(selection.get("revision_id"))):
|
|
raise Blocked("NATIVE_CAPTURE_SCENARIO_REVISION_REQUIRED")
|
|
|
|
|
|
def validate_published_resume_target(target: dict) -> None:
|
|
resume = target.get("published_resume") if isinstance(target, dict) else None
|
|
selection = target.get("catalog_selection") if isinstance(target, dict) else None
|
|
if (target.get("catalog_mode") != "exact_revision"
|
|
or not isinstance(selection, dict) or selection.get("kind") != "revision"
|
|
or not _is_uuid(selection.get("catalog_id"))
|
|
or not _is_uuid(selection.get("catalog_revision_id"))
|
|
or not isinstance(resume, dict)
|
|
or set(resume) != {"capture_artifact_id", "publication_operation_id"}
|
|
or not _is_uuid(resume.get("capture_artifact_id"))
|
|
or not _is_uuid(resume.get("publication_operation_id"))):
|
|
raise Blocked("NATIVE_PUBLISHED_RESUME_INPUT_INVALID")
|
|
|
|
|
|
def _is_uuid(value: object) -> bool:
|
|
if type(value) is not str:
|
|
return False
|
|
try:
|
|
return str(UUID(value)) == value.lower()
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def current_dashboard(driver: Driver, seed: dict, target: dict) -> tuple[int, dict]:
|
|
source_dashboard_id = target["source_dashboard_id"]
|
|
release_identity = target["release_identity"]
|
|
status = driver.call("preprod-deployment-status", "GET",
|
|
f"/api/git/repositories/{source_dashboard_id}/deployment-status?env_id={target['source_environment_id']}")
|
|
stages = [row for row in status.get("environments", []) if row.get("stage") == target["target_stage"]]
|
|
driver.require("preprod-release-identity-current", len(stages) == 1
|
|
and stages[0].get("validation_status") == "validated"
|
|
and stages[0].get("drift_status") == "in_sync"
|
|
and stages[0].get("commit_hash") == release_identity.get("commit_hash")
|
|
and stages[0].get("content_hash") == release_identity.get("content_hash"))
|
|
|
|
superset_url = target["superset_url"]
|
|
superset_password = os.environ["SUPERSET_PASSWORD"]
|
|
driver.secrets.append(superset_password)
|
|
with httpx.Client(base_url=superset_url, timeout=45, trust_env=False) as superset:
|
|
login = superset.post("/api/v1/security/login", json={
|
|
"username": target["superset_username"], "password": superset_password,
|
|
"provider": "db", "refresh": True})
|
|
login.raise_for_status()
|
|
access_token = login.json().get("access_token")
|
|
if not isinstance(access_token, str) or not access_token:
|
|
raise Blocked("PREPROD_SUPERSET_AUTH_UNAVAILABLE")
|
|
driver.secrets.append(access_token)
|
|
dashboards = superset.get("/api/v1/dashboard/", params={"q": "(page:0,page_size:100)"},
|
|
headers={"Authorization": "Bearer " + access_token})
|
|
dashboards.raise_for_status()
|
|
rows = dashboards.json().get("result", [])
|
|
matches = [row for row in rows if row.get("uuid") == target["target_dashboard_uuid"]]
|
|
driver.require("preprod-dashboard-uuid-unique", len(matches) == 1
|
|
and type(matches[0].get("id")) is int and matches[0]["id"] > 0,
|
|
dashboard_match_count=len(matches))
|
|
target_dashboard_id = matches[0]["id"]
|
|
model = driver.call("preprod-current-query-model", "GET", "/api/dashboard-testing/query-model?" + urlencode({
|
|
"environment_id": target["target_environment_id"], "dashboard_id": target_dashboard_id}))
|
|
contract = target["chart_contract"]
|
|
charts = [chart for chart in model.get("charts", [])
|
|
if chart.get("chart_uuid") == target["chart_uuid"]]
|
|
datasets = [dataset for dataset in model.get("datasets", [])
|
|
if dataset.get("dataset_uuid") == contract["dataset_uuid"]]
|
|
chart_metrics = {metric.get("metric_name") for metric in charts[0].get("metrics", [])} if len(charts) == 1 else set()
|
|
driver.require("chart8-current-query-model-identity", len(charts) == 1
|
|
and charts[0].get("viz_type") == contract["viz_type"]
|
|
and len(datasets) == 1
|
|
and charts[0].get("dataset_id") == datasets[0].get("dataset_id")
|
|
and charts[0].get("group_by_columns") == contract["group_by_columns"]
|
|
and chart_metrics == set(contract["metric_names"]),
|
|
matching_chart_count=len(charts), matching_dataset_count=len(datasets),
|
|
model_chart_count=len(model.get("charts", [])))
|
|
target["resolved_target_dashboard_id"] = target_dashboard_id
|
|
return target_dashboard_id, model
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.CurrentIdentity
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.Scenario [C:3] [TYPE Function]
|
|
# @POST An actor-owned valid manual scenario exists for the capture's exact current dashboard/environment.
|
|
def register_capture_scenario(driver: Driver, dashboard_id: int, model: dict, environment_id: str,
|
|
target: dict) -> dict:
|
|
selection = target["capture_scenario"]
|
|
if selection["mode"] == "existing_mcp_revision":
|
|
return read_existing_mcp_scenario(driver, dashboard_id, environment_id, target)
|
|
if selection["mode"] not in {"compile_structural", "mcp_bootstrap"}:
|
|
raise Blocked("NATIVE_CAPTURE_MCP_BOOTSTRAP_REQUIRED")
|
|
return register_mcp_owned_scenario(
|
|
driver, dashboard_id, model, environment_id, target,
|
|
objective=selection.get("objective", "Owned chart-response capture context"),
|
|
step_prefix="capture")
|
|
|
|
|
|
def read_existing_mcp_scenario(driver: Driver, dashboard_id: int, environment_id: str,
|
|
target: dict) -> dict:
|
|
"""Resume only the caller's exact MCP-created scenario revision."""
|
|
selection = target["capture_scenario"]
|
|
scenario_id, revision_id = selection.get("scenario_id"), selection.get("revision_id")
|
|
validate_existing_mcp_revision(selection)
|
|
driver.ids.update(scenario_id=scenario_id, revision_id=revision_id)
|
|
detail = driver.call("capture-mcp-scenario-resume-readback", "GET",
|
|
f"/api/dashboard-testing/scenarios/{scenario_id}")
|
|
revisions = driver.call("capture-mcp-scenario-resume-revisions", "GET",
|
|
f"/api/dashboard-testing/scenarios/{scenario_id}/revisions")
|
|
actor = driver.call("capture-mcp-scenario-resume-actor", "GET", "/api/auth/me")
|
|
entry = detail.get("entry", {})
|
|
matches = [row for row in revisions if isinstance(row, dict)
|
|
and row.get("revision_id") == revision_id]
|
|
if (len(matches) != 1 or entry.get("scenario_id") != scenario_id
|
|
or entry.get("current_revision") != revision_id
|
|
or entry.get("dashboard_id") != dashboard_id
|
|
or environment_id not in entry.get("environment_ids", [])
|
|
or entry.get("owner_username") != target["backend_username"]
|
|
or actor.get("username") != target["backend_username"]):
|
|
raise Blocked("NATIVE_MCP_SCENARIO_RESUME_IDENTITY_MISMATCH")
|
|
driver.record("capture-mcp-scenario-exact-resume", "PASS",
|
|
scenario_id=scenario_id, revision_id=revision_id)
|
|
return {"scenario_id": scenario_id, "revision_id": revision_id,
|
|
"entry": entry, "graph": detail.get("graph")}
|
|
|
|
|
|
def register_mcp_owned_scenario(driver: Driver, dashboard_id: int, model: dict,
|
|
environment_id: str, target: dict, *,
|
|
objective: str, step_prefix: str) -> dict:
|
|
"""Create a scenario through MCP-owned run, handle, and bootstrap boundaries."""
|
|
selection = target["capture_scenario"]
|
|
cases = selection.get("selected_case_ids")
|
|
if not isinstance(cases, list) or not cases or any(
|
|
type(case) is not str or not case.strip() for case in cases
|
|
):
|
|
raise Blocked("NATIVE_CAPTURE_SCENARIO_CASES_REQUIRED")
|
|
mcp = McpClient(_SafeMcpEvidence(driver))
|
|
inspected = mcp.call(f"{step_prefix}-mcp-dashboard-inspection", "inspect_dashboard_context", {
|
|
"environment_id": environment_id, "dashboard_id": dashboard_id,
|
|
})
|
|
observed_model = inspected.get("query_model")
|
|
if (not isinstance(observed_model, dict)
|
|
or observed_model.get("dashboard_id") != dashboard_id
|
|
or observed_model.get("environment_id") != environment_id
|
|
or observed_model.get("query_model_fingerprint") != model.get("query_model_fingerprint")):
|
|
raise Blocked("NATIVE_MCP_INSPECTION_IDENTITY_MISMATCH")
|
|
run = mcp.call(f"{step_prefix}-mcp-agent-run", "create_agent_run", {
|
|
"dashboard_id": dashboard_id, "environment_id": environment_id,
|
|
"dashboard_name": model.get("title", "Native chart response"),
|
|
"idempotency_key": f"chart8-{step_prefix}-" + uuid4().hex,
|
|
})
|
|
run_id = run.get("run_id")
|
|
if not isinstance(run_id, str) or not run_id:
|
|
raise Blocked("NATIVE_MCP_AGENT_RUN_UNAVAILABLE")
|
|
compiled = mcp.call(f"{step_prefix}-mcp-scenario-compile", "inspect_scenario", {
|
|
"agent_run_id": run_id, "query_model": observed_model,
|
|
"environment_id": environment_id, "dashboard_id": dashboard_id,
|
|
"dashboard_name": model.get("title", "Native chart response"),
|
|
"objective": {"goal": objective, "selected_case_ids": cases},
|
|
"checklist_catalog_version": 1, "baseline_version": "none",
|
|
"has_dataset_fields": True,
|
|
"capabilities": {"dataset_field_read": True, "baseline": False},
|
|
"parameters": selection.get("parameters", {
|
|
"field_names": {"type": "string_list", "default": target["chart_contract"]["key_columns"]}}),
|
|
})
|
|
graph = compiled.get("scenario")
|
|
if not isinstance(graph, dict):
|
|
raise Blocked("NATIVE_MCP_COMPILED_GRAPH_UNAVAILABLE")
|
|
pack = mcp.call(f"{step_prefix}-mcp-register-draft-pack", "register_draft_pack", {
|
|
"agent_run_id": run_id, "scenario": graph,
|
|
})
|
|
if (pack.get("status") != "save_eligible"
|
|
or pack.get("context_authority") != "verified"
|
|
or not all(isinstance(pack.get(key), str) and pack[key]
|
|
for key in ("compiled_handle_id", "draft_pack_handle_id", "draft_pack_digest"))):
|
|
raise Blocked("NATIVE_MCP_DRAFT_PACK_UNAVAILABLE")
|
|
bootstrapped = mcp.call(f"{step_prefix}-mcp-bootstrap-scenario", "bootstrap_authoring_scenario", {
|
|
"idempotency_key": f"chart8-bootstrap-{step_prefix}-" + uuid4().hex,
|
|
"title": selection.get("title", "Native chart response capture"),
|
|
"dashboard_id": dashboard_id,
|
|
"allowed_environment_ids": [environment_id],
|
|
"selected_environment_id": environment_id,
|
|
"selected_case_ids": cases,
|
|
"objective": objective,
|
|
"compiled_handle_id": pack["compiled_handle_id"],
|
|
"draft_pack_id": pack["draft_pack_handle_id"],
|
|
"draft_pack_digest": pack["draft_pack_digest"],
|
|
})
|
|
scenario_id = bootstrapped.get("scenario_id")
|
|
revision_id = bootstrapped.get("revision_id")
|
|
if (bootstrapped.get("activation_status") != "current"
|
|
or not isinstance(scenario_id, str) or not scenario_id
|
|
or not isinstance(revision_id, str) or not revision_id):
|
|
raise Blocked("NATIVE_MCP_SCENARIO_BOOTSTRAP_UNAVAILABLE")
|
|
driver.ids.update(scenario_id=scenario_id, revision_id=revision_id)
|
|
detail = driver.call(f"{step_prefix}-mcp-scenario-readback", "GET",
|
|
f"/api/dashboard-testing/scenarios/{scenario_id}")
|
|
revisions = driver.call(f"{step_prefix}-mcp-scenario-revision-readback", "GET",
|
|
f"/api/dashboard-testing/scenarios/{scenario_id}/revisions")
|
|
actor = driver.call(f"{step_prefix}-mcp-actor-readback", "GET", "/api/auth/me")
|
|
entry = detail.get("entry", {})
|
|
matches = [item for item in revisions if isinstance(item, dict)
|
|
and item.get("revision_id") == revision_id]
|
|
if (len(matches) != 1 or entry.get("scenario_id") != scenario_id
|
|
or entry.get("current_revision") != revision_id
|
|
or entry.get("dashboard_id") != dashboard_id
|
|
or environment_id not in entry.get("environment_ids", [])
|
|
or entry.get("owner_username") != actor.get("username")
|
|
or actor.get("username") != target["backend_username"]):
|
|
raise Blocked("NATIVE_MCP_SCENARIO_OWNER_READBACK_MISMATCH")
|
|
return {"agent_run_id": run_id, "scenario_id": scenario_id, "revision_id": revision_id,
|
|
"entry": entry, "graph": detail.get("graph"),
|
|
"materialization_status": bootstrapped.get("materialization_status")}
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.Scenario
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.M01Scenario [C:4] [TYPE Function]
|
|
# @PRE The current actor, observed dashboard model, exact published catalog revision and protected native URL are explicit.
|
|
# @POST The registered scenario revision is server-compiled from the current M01 profile and owned by this actor.
|
|
# @INVARIANT No prior scenario owner, graph snapshot, metric value, or inferred catalog revision is reused.
|
|
def select_m01_profile_coordinate(profile: dict, target: dict, dashboard_id: int, model: dict) -> dict:
|
|
if not isinstance(profile, dict) or not isinstance(profile.get("coordinates"), list):
|
|
raise Blocked("NATIVE_M01_PROFILE_COORDINATES_UNAVAILABLE")
|
|
charts = [chart for chart in model.get("charts", [])
|
|
if chart.get("chart_uuid") == target["chart_uuid"]]
|
|
if len(charts) != 1:
|
|
raise Blocked("NATIVE_M01_CURRENT_CHART_UNAVAILABLE")
|
|
expected_dataset_id = charts[0].get("dataset_id")
|
|
expected_chart_id = charts[0].get("chart_id")
|
|
coordinates = [item for item in profile["coordinates"]
|
|
if isinstance(item, dict)
|
|
and item.get("chart_id") == expected_chart_id
|
|
and item.get("dataset_id") == expected_dataset_id
|
|
and item.get("metric_name") is None
|
|
and item.get("reference_coordinate") == {
|
|
"kind": "chart", "chart_uuid": target["chart_uuid"], "metric": None,
|
|
}]
|
|
if type(expected_chart_id) is not int or len(coordinates) != 1:
|
|
raise Blocked("NATIVE_M01_PROFILE_COORDINATE_AMBIGUOUS")
|
|
return coordinates[0]
|
|
|
|
|
|
def read_published_capture(driver: Driver, target: dict, model: dict) -> dict:
|
|
"""Resume M01 only from an explicitly named, publicly read-back publication."""
|
|
resume = target.get("published_resume")
|
|
selection = target.get("catalog_selection")
|
|
expected_capture = resume.get("capture_artifact_id") if isinstance(resume, dict) else None
|
|
operation_id = resume.get("publication_operation_id") if isinstance(resume, dict) else None
|
|
if (not isinstance(resume, dict) or set(resume) != {"capture_artifact_id", "publication_operation_id"}
|
|
or not _is_uuid(expected_capture) or not _is_uuid(operation_id)
|
|
or target.get("catalog_mode") != "exact_revision"
|
|
or not isinstance(selection, dict) or selection.get("kind") != "revision"):
|
|
raise Blocked("NATIVE_PUBLISHED_RESUME_INPUT_INVALID")
|
|
|
|
operation = driver.call("published-resume-operation-readback", "GET",
|
|
f"/api/catalog-publications/{operation_id}")
|
|
revision_response = driver.call("published-resume-revision-readback", "GET",
|
|
"/api/dashboard-testing/baselines?" + urlencode({
|
|
"catalog_id": selection["catalog_id"],
|
|
"catalog_revision_id": selection["catalog_revision_id"],
|
|
}))
|
|
revision = revision_response.get("revision", {})
|
|
if (operation.get("operation_id") != operation_id or operation.get("state") != "published"
|
|
or operation.get("catalog_revision_id") != selection["catalog_revision_id"]
|
|
or operation.get("catalog_digest") != revision.get("catalog_digest")
|
|
or not operation.get("published_receipt")
|
|
or operation.get("repository") != target["catalog_repository"]
|
|
or operation.get("branch") != target["catalog_branch"]
|
|
or revision.get("catalog_id") != selection["catalog_id"]
|
|
or revision.get("catalog_revision_id") != selection["catalog_revision_id"]
|
|
or revision.get("release_id") != target["release_identity"].get("id")):
|
|
raise Blocked("NATIVE_PUBLISHED_RESUME_IDENTITY_MISMATCH")
|
|
|
|
wrappers = revision.get("entry_revisions")
|
|
matches = [wrapper for wrapper in wrappers if isinstance(wrapper, dict)
|
|
and wrapper.get("capture_artifact_id") == expected_capture
|
|
and isinstance(wrapper.get("entry"), dict)]
|
|
if len(matches) != 1:
|
|
raise Blocked("NATIVE_PUBLISHED_RESUME_CAPTURE_MISSING")
|
|
wrapper = matches[0]
|
|
entry = wrapper["entry"]
|
|
expected = entry.get("expected") or {}
|
|
source = wrapper.get("reference_source") or {}
|
|
native = entry.get("native_result") or {}
|
|
authority = native.get("authority_record") or {}
|
|
descriptor = authority.get("descriptor") or {}
|
|
reference = descriptor.get("reference_coordinate") or {}
|
|
chart = next((item for item in model.get("charts", [])
|
|
if item.get("chart_uuid") == target["chart_uuid"]), None)
|
|
if (wrapper.get("status") != "approved" or entry.get("status") != "approved"
|
|
or expected.get("kind") != "chart_response"
|
|
or entry.get("result_key", "missing") is not None
|
|
or entry.get("chart_id") != descriptor.get("chart_id")
|
|
or entry.get("dataset_id") != descriptor.get("dataset_id")
|
|
or not isinstance(chart, dict)
|
|
or descriptor.get("chart_id") != chart.get("chart_id")
|
|
or descriptor.get("dataset_id") != chart.get("dataset_id")
|
|
or descriptor.get("kind") != "chart_response"
|
|
or authority.get("capture_artifact_id") != expected_capture
|
|
or descriptor.get("selected_metric", "missing") is not None
|
|
or reference != {"kind": "chart", "chart_uuid": target["chart_uuid"], "metric": None}
|
|
or source.get("environment_id") != target["target_environment_id"]
|
|
or source.get("dashboard_id") != target["resolved_target_dashboard_id"]
|
|
or source.get("filters_hash") != target["expected_filters_hash"]
|
|
or source.get("query_model_fingerprint") != model.get("query_model_fingerprint")
|
|
or entry.get("release_version") != target["release_identity"].get("version")
|
|
or entry.get("release_commit_hash") != target["release_identity"].get("commit_hash")):
|
|
raise Blocked("NATIVE_PUBLISHED_RESUME_CAPTURE_IDENTITY_MISMATCH")
|
|
|
|
driver.ids.update(capture_artifact_id=expected_capture,
|
|
published_catalog_revision_id=selection["catalog_revision_id"],
|
|
publication_operation_id=operation_id)
|
|
driver.record("published-capture-exact-resume", "PASS",
|
|
catalog_id=selection["catalog_id"],
|
|
catalog_revision_id=selection["catalog_revision_id"],
|
|
operation_id=operation_id,
|
|
capture_count=1, publication_state="published")
|
|
return {"capture": {"release_id": target["release_identity"]["id"],
|
|
"capture_artifact_ids": [expected_capture]},
|
|
"catalog_revision": revision,
|
|
"publication": operation,
|
|
"catalog_selection": selection,
|
|
"coordinate": {"chart_id": descriptor["chart_id"],
|
|
"dataset_id": descriptor["dataset_id"],
|
|
"chart_uuid": target["chart_uuid"],
|
|
"dataset_uuid": target["chart_contract"]["dataset_uuid"]},
|
|
"query_model_fingerprint": source["query_model_fingerprint"],
|
|
"filters_hash": source["filters_hash"]}
|
|
|
|
|
|
def register_m01_capture_scenario(driver: Driver, dashboard_id: int, model: dict,
|
|
environment_id: str, target: dict) -> dict:
|
|
"""Resolve the published chart-response source through the owner-scoped M01 MCP workflow."""
|
|
selection = target["capture_scenario"]
|
|
mcp = McpClient(_SafeMcpEvidence(driver))
|
|
agent = mcp.call("m01-mcp-agent-run", "create_agent_run", {
|
|
"dashboard_id": dashboard_id, "environment_id": environment_id,
|
|
"dashboard_name": model.get("title", "Native chart response"),
|
|
"idempotency_key": "chart8-m01-" + uuid4().hex,
|
|
})
|
|
agent_run_id = agent.get("run_id")
|
|
if not isinstance(agent_run_id, str) or not agent_run_id:
|
|
raise Blocked("NATIVE_MCP_AGENT_RUN_UNAVAILABLE")
|
|
numeric = target.get("numeric_verification")
|
|
if not isinstance(numeric, dict):
|
|
raise Blocked("NATIVE_NUMERIC_VERIFICATION_TARGET_INCOMPLETE")
|
|
proposed = mcp.call("m01-propose-metric-profile", "propose_metric_baseline_profile", {
|
|
"environment_id": environment_id, "dashboard_id": dashboard_id,
|
|
"objective": selection.get("objective", "Bind the exact published chart-response baseline for native capture"),
|
|
"evidence_recipe": numeric["evaluation_recipe_id"],
|
|
"evaluation_provider_id": numeric["evaluation_provider_id"],
|
|
})
|
|
if proposed.get("status") != "preview_only" or not isinstance(proposed.get("profile"), dict):
|
|
raise Blocked("NATIVE_M01_PROFILE_PROPOSAL_UNAVAILABLE")
|
|
profile = proposed["profile"]
|
|
coordinate = select_m01_profile_coordinate(profile, target, dashboard_id, model)
|
|
catalog_selection = target["catalog_selection"]
|
|
resolved = mcp.call("m01-resolve-published-metric-profile", "resolve_metric_baseline_profile", {
|
|
"profile_handle_id": proposed["profile_handle_id"],
|
|
"expected_profile_digest": profile["profile_digest"],
|
|
"expected_cas_version": proposed["cas_version"],
|
|
"idempotency_key": "chart8-m01-" + uuid4().hex,
|
|
"coordinate_id": coordinate["coordinate_id"],
|
|
"catalog_id": catalog_selection["catalog_id"],
|
|
"catalog_revision_id": catalog_selection["catalog_revision_id"],
|
|
"reference_url": target["reference_url"],
|
|
})
|
|
if (resolved.get("status") != "save_eligible" or not isinstance(resolved.get("profile"), dict)
|
|
or resolved["profile"].get("eligible") is not True
|
|
or resolved["profile"].get("status") != "save_eligible"):
|
|
raise Blocked("NATIVE_M01_PROFILE_RESOLUTION_UNAVAILABLE")
|
|
registered_pack = mcp.call("m01-register-metric-baseline-pack", "register_metric_baseline_pack", {
|
|
"profile_handle_id": resolved["profile_handle_id"],
|
|
"expected_profile_digest": resolved["profile"]["profile_digest"],
|
|
"expected_cas_version": resolved["cas_version"],
|
|
"agent_run_id": agent_run_id,
|
|
})
|
|
if (registered_pack.get("status") != "save_eligible"
|
|
or registered_pack.get("context_authority") != "verified"
|
|
or not all(isinstance(registered_pack.get(key), str) and registered_pack[key]
|
|
for key in ("compiled_handle_id", "draft_pack_id", "draft_pack_digest"))):
|
|
raise Blocked("NATIVE_M01_DRAFT_PACK_UNAVAILABLE")
|
|
bootstrap = mcp.call("m01-mcp-bootstrap-scenario", "bootstrap_authoring_scenario", {
|
|
"idempotency_key": "chart8-m01-bootstrap-" + uuid4().hex,
|
|
"title": selection.get("title", "Native chart response M01 binding"),
|
|
"dashboard_id": dashboard_id,
|
|
"allowed_environment_ids": [environment_id],
|
|
"selected_environment_id": environment_id,
|
|
"selected_case_ids": selection.get("selected_case_ids", ["T02"]),
|
|
"objective": selection.get("objective", "Bind the exact published chart-response metric profile"),
|
|
"compiled_handle_id": registered_pack["compiled_handle_id"],
|
|
"draft_pack_id": registered_pack["draft_pack_id"],
|
|
"draft_pack_digest": registered_pack["draft_pack_digest"],
|
|
})
|
|
scenario_id = bootstrap.get("scenario_id")
|
|
revision_id = bootstrap.get("revision_id")
|
|
if (bootstrap.get("activation_status") != "current"
|
|
or not isinstance(scenario_id, str) or not scenario_id
|
|
or not isinstance(revision_id, str) or not revision_id):
|
|
raise Blocked("NATIVE_M01_SCENARIO_REGISTRATION_UNAVAILABLE")
|
|
detail = driver.call("m01-scenario-owner-readback", "GET",
|
|
f"/api/dashboard-testing/scenarios/{scenario_id}")
|
|
revisions = driver.call("m01-scenario-revision-readback", "GET",
|
|
f"/api/dashboard-testing/scenarios/{scenario_id}/revisions")
|
|
entry = detail.get("entry", {})
|
|
matches = [item for item in revisions if isinstance(item, dict) and item.get("revision_id") == revision_id]
|
|
actor = driver.call("m01-actor-readback", "GET", "/api/auth/me")
|
|
expected_owner = actor.get("username")
|
|
graph = detail.get("graph")
|
|
metric_graph = graph.get("metric_graph") if isinstance(graph, dict) else None
|
|
recipe = metric_graph.get("metric_visual_recipe") if isinstance(metric_graph, dict) else None
|
|
driver.require("m01-scenario-owner-and-scope", len(matches) == 1
|
|
and entry.get("scenario_id") == scenario_id
|
|
and entry.get("current_revision") == revision_id
|
|
and entry.get("owner_username") == expected_owner == target["backend_username"]
|
|
and entry.get("dashboard_id") == dashboard_id
|
|
and environment_id in entry.get("environment_ids", [])
|
|
and isinstance(recipe, dict)
|
|
and recipe.get("recipe_id") == numeric["evaluation_recipe_id"]
|
|
and recipe.get("provider_id") == numeric["evaluation_provider_id"]
|
|
and bootstrap.get("materialization_status") == "materialized",
|
|
scenario_id_present=True, revision_match=len(matches) == 1,
|
|
target_dashboard_id=dashboard_id)
|
|
return {"agent_run_id": agent_run_id, "scenario_id": scenario_id,
|
|
"revision_id": revision_id, "entry": entry,
|
|
"graph": graph,
|
|
"materialization_status": bootstrap["materialization_status"],
|
|
"metric_coordinate_id": coordinate["coordinate_id"],
|
|
"catalog_id": catalog_selection["catalog_id"],
|
|
"catalog_revision_id": catalog_selection["catalog_revision_id"]}
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.M01Scenario
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.Catalog [C:3] [TYPE Function]
|
|
# @POST The repository UUID comes from an official release-bound snapshot response; seed returns an exact empty revision.
|
|
def select_catalog(driver: Driver, target: dict) -> dict:
|
|
source_dashboard_id = target["source_dashboard_id"]
|
|
release_id = target["release_identity"]["id"]
|
|
environment_id = target["target_environment_id"]
|
|
if target["catalog_mode"] == "exact_revision":
|
|
selection = target["catalog_selection"]
|
|
confirmed = driver.call("catalog-selection-readback", "GET", "/api/dashboard-testing/baselines?" + urlencode({
|
|
"catalog_id": selection["catalog_id"], "catalog_revision_id": selection["catalog_revision_id"]}))
|
|
revision = confirmed.get("revision", {})
|
|
driver.require("catalog-selection-exact-readback",
|
|
revision.get("catalog_id") == selection["catalog_id"]
|
|
and revision.get("catalog_revision_id") == selection["catalog_revision_id"])
|
|
return {"selection": {"kind": "revision", **selection}, "source_dashboard_id": source_dashboard_id}
|
|
if target["catalog_mode"] == "current_head":
|
|
selection = target["catalog_selection"]
|
|
driver.record("catalog-current-head-explicitly-selected", "PASS",
|
|
catalog_id=selection["catalog_id"],
|
|
resolution="deferred_to_native_capture")
|
|
return {"selection": {"kind": "current", "catalog_id": selection["catalog_id"]},
|
|
"source_dashboard_id": source_dashboard_id,
|
|
"resolution": "deferred_to_native_capture"}
|
|
snapshot = driver.call("release-bound-source-snapshot", "POST",
|
|
f"/api/dashboard-testing/structure-snapshot/capture?environment_id={environment_id}",
|
|
{"release_id": release_id}, expected=(201,))
|
|
driver.require("release-snapshot-exact-dashboard", snapshot.get("dashboard_id") == source_dashboard_id
|
|
and snapshot.get("environment_id") == environment_id
|
|
and isinstance(snapshot.get("repository_id"), str) and bool(snapshot["repository_id"]))
|
|
seeded = driver.call("catalog-seed-exact-repository", "POST", "/api/dashboard-testing/catalogs/seed", {
|
|
"repository_id": snapshot["repository_id"], "dashboard_id": source_dashboard_id,
|
|
"reason": "Isolated native chart-8 E2E capture catalog"}, expected=(201,),
|
|
headers={"Idempotency-Key": "chart8-seed-" + str(uuid4())})
|
|
revision = seeded.get("revision", {})
|
|
driver.require("catalog-seed-empty-exact-head", bool(revision.get("catalog_id"))
|
|
and bool(revision.get("catalog_revision_id")) and revision.get("cas_version") == 1
|
|
and revision.get("entry_revisions") == [] and revision.get("observatory_entries") == [])
|
|
exact = {"kind": "revision", "catalog_id": revision["catalog_id"],
|
|
"catalog_revision_id": revision["catalog_revision_id"]}
|
|
driver.ids.update(catalog_id=exact["catalog_id"], catalog_revision_id=exact["catalog_revision_id"])
|
|
driver.record("catalog-seed-exact-identities-retained", "PASS",
|
|
catalog_id=exact["catalog_id"], catalog_revision_id=exact["catalog_revision_id"])
|
|
confirmed = driver.call("catalog-seed-readback", "GET", "/api/dashboard-testing/baselines?" + urlencode({
|
|
"catalog_id": exact["catalog_id"], "catalog_revision_id": exact["catalog_revision_id"]}))
|
|
driver.require("catalog-seed-readback-empty", confirmed.get("revision", {}).get("catalog_revision_id")
|
|
== exact["catalog_revision_id"] and confirmed.get("revision", {}).get("entry_revisions") == [])
|
|
return {"selection": exact, "snapshot_repository_id": snapshot["repository_id"],
|
|
"source_dashboard_id": source_dashboard_id, "snapshot_path_present": bool(snapshot.get("snapshot_path"))}
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.Catalog
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.PublishedRevisionTarget [C:2] [TYPE Function]
|
|
# @PRE A publication response names one exact immutable revision of the selected catalog.
|
|
# @POST The next M01 profile request is directed to that revision without mutating the authored target.
|
|
# @INVARIANT Missing, current-head, or foreign-catalog selections cannot become the post-publication pin.
|
|
def target_for_published_revision(target: dict, publication: dict) -> dict:
|
|
selection = publication.get("catalog_selection") if isinstance(publication, dict) else None
|
|
original = target.get("catalog_selection") if isinstance(target, dict) else None
|
|
if (not isinstance(selection, dict) or selection.get("kind") != "revision"
|
|
or not isinstance(selection.get("catalog_id"), str) or not selection["catalog_id"]
|
|
or not isinstance(selection.get("catalog_revision_id"), str)
|
|
or not selection["catalog_revision_id"]
|
|
or not isinstance(original, dict) or selection["catalog_id"] != original.get("catalog_id")):
|
|
raise Blocked("NATIVE_PUBLISHED_REVISION_SELECTION_UNAVAILABLE")
|
|
return {**target, "catalog_mode": "exact_revision", "catalog_selection": {
|
|
"kind": "revision", "catalog_id": selection["catalog_id"],
|
|
"catalog_revision_id": selection["catalog_revision_id"],
|
|
}}
|
|
|
|
|
|
def bind_seeded_catalog_selection(target: dict, catalog: dict) -> dict:
|
|
"""Bind only the exact official empty-catalog seed result to this run's target."""
|
|
selection = catalog.get("selection") if isinstance(catalog, dict) else None
|
|
if (not isinstance(target, dict) or target.get("catalog_mode") != "seed_empty"
|
|
or not isinstance(selection, dict) or selection.get("kind") != "revision"
|
|
or not isinstance(selection.get("catalog_id"), str) or not selection["catalog_id"]
|
|
or not isinstance(selection.get("catalog_revision_id"), str)
|
|
or not selection["catalog_revision_id"]
|
|
or catalog.get("source_dashboard_id") != target.get("source_dashboard_id")):
|
|
raise Blocked("NATIVE_SEEDED_CATALOG_SELECTION_UNAVAILABLE")
|
|
return {**target, "catalog_selection": {
|
|
"kind": "revision", "catalog_id": selection["catalog_id"],
|
|
"catalog_revision_id": selection["catalog_revision_id"],
|
|
}}
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.PublishedRevisionTarget
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.NumericVerifierTarget [C:3] [TYPE Function]
|
|
# @PRE Native capture and catalog publication returned exact public identities; M01 scenario readback is current.
|
|
# @POST The numeric verifier receives explicit acceptance inputs and observed publication/context facts.
|
|
# @INVARIANT Pin/query hashes are not recomputed or guessed; the verifier obtains the pin and query identity from run one.
|
|
def numeric_verifier_target(target: dict, scenario: dict, captured: dict) -> dict:
|
|
numeric = target.get("numeric_verification") if isinstance(target, dict) else None
|
|
revision = captured.get("catalog_revision") if isinstance(captured, dict) else None
|
|
publication = captured.get("publication") if isinstance(captured, dict) else None
|
|
selection = captured.get("catalog_selection") if isinstance(captured, dict) else None
|
|
coordinate = captured.get("coordinate") if isinstance(captured, dict) else None
|
|
graph = scenario.get("graph") if isinstance(scenario, dict) else None
|
|
entry = scenario.get("entry") if isinstance(scenario, dict) else None
|
|
contract = target.get("chart_contract") if isinstance(target, dict) else None
|
|
release = target.get("release_identity") if isinstance(target, dict) else None
|
|
if (not isinstance(numeric, dict) or not isinstance(revision, dict)
|
|
or not isinstance(publication, dict) or not isinstance(selection, dict)
|
|
or not isinstance(coordinate, dict) or not isinstance(graph, dict)
|
|
or not isinstance(entry, dict) or not isinstance(contract, dict)
|
|
or not isinstance(release, dict)):
|
|
raise Blocked("NATIVE_NUMERIC_VERIFIER_PUBLIC_READBACK_UNAVAILABLE")
|
|
scenario_id, revision_id = scenario.get("scenario_id"), scenario.get("revision_id")
|
|
catalog_id, catalog_revision_id = selection.get("catalog_id"), selection.get("catalog_revision_id")
|
|
recipe = (graph.get("metric_graph") or {}).get("metric_visual_recipe")
|
|
if (not isinstance(scenario_id, str) or not scenario_id
|
|
or not isinstance(revision_id, str) or not revision_id
|
|
or entry.get("scenario_id") != scenario_id or entry.get("current_revision") != revision_id
|
|
or scenario.get("catalog_id") != catalog_id
|
|
or scenario.get("catalog_revision_id") != catalog_revision_id
|
|
or not isinstance(catalog_id, str) or not catalog_id
|
|
or not isinstance(catalog_revision_id, str) or not catalog_revision_id
|
|
or revision.get("catalog_id") != catalog_id
|
|
or revision.get("catalog_revision_id") != catalog_revision_id
|
|
or publication.get("catalog_revision_id") != catalog_revision_id
|
|
or publication.get("state") != "published"
|
|
or not isinstance(captured.get("capture"), dict)
|
|
or captured["capture"].get("release_id") != release.get("id")
|
|
or publication.get("catalog_id") != catalog_id
|
|
or not isinstance(recipe, dict)
|
|
or recipe.get("recipe_id") != "visual_single_attempt_v1"
|
|
or recipe.get("recipe_id") != numeric.get("evaluation_recipe_id")
|
|
or recipe.get("provider_id") != numeric.get("evaluation_provider_id")
|
|
or not isinstance(recipe.get("provider_id"), str) or not recipe["provider_id"]):
|
|
raise Blocked("NATIVE_NUMERIC_VERIFIER_IDENTITY_MISMATCH")
|
|
for key in ("chart_id", "dataset_id"):
|
|
if type(coordinate.get(key)) is not int or coordinate[key] <= 0:
|
|
raise Blocked("NATIVE_NUMERIC_VERIFIER_COORDINATE_UNAVAILABLE")
|
|
if (coordinate.get("chart_uuid") != target.get("chart_uuid")
|
|
or coordinate.get("dataset_uuid") != contract.get("dataset_uuid")
|
|
or coordinate.get("metric") is not None):
|
|
raise Blocked("NATIVE_NUMERIC_VERIFIER_COORDINATE_MISMATCH")
|
|
filters_hash = captured.get("filters_hash")
|
|
query_model_fingerprint = captured.get("query_model_fingerprint")
|
|
publication_facts = {
|
|
"operation_id": publication.get("operation_id"),
|
|
"commit_sha": publication.get("commit_sha"),
|
|
"catalog_digest": revision.get("catalog_digest"),
|
|
"render_sha256": publication.get("render_sha256"),
|
|
"release_id": release.get("id"),
|
|
"release_version": release.get("version"),
|
|
"release_commit_hash": release.get("commit_hash"),
|
|
}
|
|
if (not isinstance(filters_hash, str) or not filters_hash
|
|
or not isinstance(query_model_fingerprint, str) or not query_model_fingerprint
|
|
or filters_hash != target.get("expected_filters_hash")
|
|
or any(not isinstance(value, str) or not value for value in publication_facts.values())):
|
|
raise Blocked("NATIVE_NUMERIC_VERIFIER_READBACK_INCOMPLETE")
|
|
return {
|
|
"scenario_id": scenario_id,
|
|
"revision_id": revision_id,
|
|
"environment_id": target["target_environment_id"],
|
|
"dashboard_release_id": release["id"],
|
|
"catalog_id": catalog_id,
|
|
"catalog_revision_id": catalog_revision_id,
|
|
"publication": publication_facts,
|
|
"params": numeric["params"],
|
|
"dashboard_id": target["resolved_target_dashboard_id"],
|
|
"chart_id": coordinate["chart_id"],
|
|
"dataset_id": coordinate["dataset_id"],
|
|
"chart_uuid": coordinate["chart_uuid"],
|
|
"dataset_uuid": coordinate["dataset_uuid"],
|
|
"query_model_fingerprint": query_model_fingerprint,
|
|
"filters_hash": filters_hash,
|
|
"metric": target["capture_scenario"]["metric_name"],
|
|
"row_key": numeric["row_key"],
|
|
"query_index": numeric["query_index"],
|
|
"snapshot_date": numeric["snapshot_date"],
|
|
"filter_from": numeric["filter_from"],
|
|
"filter_to": numeric["filter_to"],
|
|
"filter_inclusive": numeric["filter_inclusive"],
|
|
"evaluation_recipe_id": recipe["recipe_id"],
|
|
"evaluation_provider_id": recipe["provider_id"],
|
|
}
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.NumericVerifierTarget
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.NumericVerifier [C:3] [TYPE Function]
|
|
# @PRE The bootstrap JSON contains only explicit target facts and exact public publication/scenario readbacks.
|
|
# @POST One verifier process owns PASS → guarded mutation/named failure → exact restore/PASS.
|
|
# @INVARIANT No extra first run, fallback selection, or success from an incomplete verifier response.
|
|
def run_numeric_verifier(driver: Driver, target: dict, numeric_target: dict,
|
|
private_env: Path, output: Path) -> dict:
|
|
verifier = Path(__file__).with_name("chart8_numeric_verification.py")
|
|
identity_path = output / "numeric-verifier-target.json"
|
|
fd = os.open(identity_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as stream:
|
|
json.dump(numeric_target, stream, sort_keys=True, separators=(",", ":"))
|
|
stream.write("\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
verification_output = output / "numeric-verification"
|
|
numeric = target["numeric_verification"]
|
|
pin_args = ( ["--expected-pin-sha256", numeric["expected_pin_sha256"]]
|
|
if numeric["pin_mode"] == "existing" else ["--resolve-new-published-pin"] )
|
|
command = [sys.executable, str(verifier),
|
|
"--expected-identity", str(identity_path),
|
|
*pin_args,
|
|
"--fixture-seed", target["numeric_verification"]["finance_fixture_seed_path"],
|
|
"--backend-url", target["backend_url"], "--auth-file", str(private_env),
|
|
"--mutator-container", target["numeric_verification"]["mutator_container"],
|
|
"--artifact-registry-container", target["backend_container"],
|
|
"--output-dir", str(verification_output)]
|
|
try:
|
|
completed = subprocess.run(command, check=False, capture_output=True, text=True)
|
|
except OSError:
|
|
driver.record("native-chart8-numeric-verifier", "BLOCKED",
|
|
error_code="NATIVE_NUMERIC_VERIFIER_LAUNCH_FAILED")
|
|
raise Blocked("NATIVE_NUMERIC_VERIFIER_LAUNCH_FAILED") from None
|
|
try:
|
|
summary = json.loads(completed.stdout)
|
|
except (TypeError, json.JSONDecodeError):
|
|
summary = None
|
|
verification_path = verification_output / "verification.json"
|
|
try:
|
|
evidence = json.loads(verification_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
evidence = None
|
|
runs = evidence.get("runs") if isinstance(evidence, dict) else None
|
|
labels = ["advisory_pass", "one_cent_named_failure", "restored_pass"]
|
|
valid_runs = (isinstance(runs, list) and len(runs) == 3
|
|
and [row.get("label") for row in runs if isinstance(row, dict)] == labels
|
|
and [row.get("status") for row in runs if isinstance(row, dict)] == ["passed", "failed", "passed"])
|
|
restored = isinstance(evidence, dict) and evidence.get("restore", {}).get("verified") is True
|
|
passed = (completed.returncode == 0 and isinstance(summary, dict)
|
|
and summary.get("status") == "passed" and summary.get("run_count") == 3
|
|
and summary.get("restore_verified") is True and valid_runs and restored)
|
|
facts = {}
|
|
if valid_runs:
|
|
facts = {"pass_run_id": runs[0].get("run_id"),
|
|
"failure_run_id": runs[1].get("run_id"),
|
|
"restored_pass_run_id": runs[2].get("run_id"),
|
|
"restore_verified": restored}
|
|
if not passed:
|
|
code = summary.get("error_code") if isinstance(summary, dict) else None
|
|
if not isinstance(code, str) or re.fullmatch(r"[A-Z][A-Z0-9_]{1,95}", code) is None:
|
|
code = "NATIVE_NUMERIC_VERIFICATION_INCOMPLETE"
|
|
driver.record("native-chart8-numeric-verifier", "BLOCKED",
|
|
error_code=code, **facts)
|
|
raise Blocked("NATIVE_NUMERIC_VERIFICATION_INCOMPLETE")
|
|
driver.record("native-chart8-numeric-verifier", "PASS", **facts)
|
|
return evidence
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.NumericVerifier
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.NativeCapture [C:4] [TYPE Function]
|
|
# @PRE Discovery is a fresh public response for the exact owned scenario and protected reference URL.
|
|
# @POST Only the whole-chart coordinate matching UUID/dataset and its explicit observed filter context is selected.
|
|
def select_native_chart_coordinate(preview: dict, target: dict, expected_model_fingerprint: str,
|
|
driver: Driver | None = None) -> tuple[dict, dict]:
|
|
if (not isinstance(preview, dict)
|
|
or preview.get("dashboard_id") != target.get("resolved_target_dashboard_id")
|
|
or preview.get("environment_id") != target.get("target_environment_id")
|
|
or preview.get("query_model_fingerprint") != expected_model_fingerprint):
|
|
raise Blocked("NATIVE_DISCOVERY_IDENTITY_MISMATCH")
|
|
contract = target["chart_contract"]
|
|
matches = [item for item in preview.get("coordinates", [])
|
|
if isinstance(item, dict)
|
|
and item.get("chart_uuid") == target["chart_uuid"]
|
|
and item.get("dataset_uuid") == contract["dataset_uuid"]
|
|
and item.get("reference_coordinate") == {"kind": "chart", "chart_uuid": target["chart_uuid"], "metric": None}]
|
|
if len(matches) != 1:
|
|
raise Blocked("NATIVE_DISCOVERY_CHART_COORDINATE_UNAVAILABLE")
|
|
coordinate = matches[0]
|
|
context = coordinate.get("native_filter_context")
|
|
expected_filters = target.get("expected_filter_context")
|
|
typed_context = None
|
|
if isinstance(context, dict):
|
|
try:
|
|
from src.schemas.dashboard_testing.filters import NormalizedFilterContext
|
|
|
|
typed_context = NormalizedFilterContext.model_validate(context)
|
|
except (ImportError, TypeError, ValueError):
|
|
typed_context = None
|
|
observed_filters = [item.model_dump(mode="json") for item in typed_context.filters] \
|
|
if typed_context is not None else None
|
|
expected_ids = [item.get("filter_id") for item in expected_filters if isinstance(item, dict)] \
|
|
if isinstance(expected_filters, list) else []
|
|
observed_ids = [item.get("filter_id") for item in observed_filters if isinstance(item, dict)] \
|
|
if isinstance(observed_filters, list) else []
|
|
filter_ids_match = (
|
|
bool(expected_ids)
|
|
and len(expected_ids) == len(expected_filters)
|
|
and len(observed_ids) == len(observed_filters)
|
|
and all(type(value) is str and value for value in expected_ids + observed_ids)
|
|
and len(set(expected_ids)) == len(expected_ids)
|
|
and len(set(observed_ids)) == len(observed_ids)
|
|
and set(expected_ids) == set(observed_ids)
|
|
)
|
|
filters_hash = context.get("filters_hash") if isinstance(context, dict) else None
|
|
filters_hash_match = (
|
|
type(filters_hash) is str
|
|
and re.fullmatch(r"sha256:[a-f0-9]{64}", filters_hash) is not None
|
|
and filters_hash == target.get("expected_filters_hash")
|
|
and coordinate.get("filters_hash") == filters_hash
|
|
)
|
|
context_schema_version_match = typed_context is not None and typed_context.schema_version == 1
|
|
context_valid = (
|
|
coordinate.get("context_error") is None
|
|
and typed_context is not None
|
|
and context_schema_version_match
|
|
and filter_ids_match
|
|
and filters_hash_match
|
|
)
|
|
if driver is not None:
|
|
driver.record("native-discovery-filter-context", "PASS" if context_valid else "BLOCKED",
|
|
chart_id=coordinate.get("chart_id"),
|
|
coordinate_count=len(matches),
|
|
context_present=isinstance(context, dict),
|
|
context_error_present=coordinate.get("context_error") is not None,
|
|
filter_count=len(observed_filters) if isinstance(observed_filters, list) else 0,
|
|
expected_filter_count=len(expected_filters) if isinstance(expected_filters, list) else 0,
|
|
filter_ids_match=filter_ids_match, filters_hash_match=filters_hash_match,
|
|
context_schema_version_match=context_schema_version_match,
|
|
filters_hash=filters_hash)
|
|
if not context_valid:
|
|
raise Blocked("NATIVE_DISCOVERY_FILTER_CONTEXT_UNAVAILABLE")
|
|
if (type(coordinate.get("chart_id")) is not int or coordinate["chart_id"] < 1
|
|
or type(coordinate.get("dataset_id")) is not int or coordinate["dataset_id"] < 1):
|
|
raise Blocked("NATIVE_DISCOVERY_NUMERIC_ID_UNAVAILABLE")
|
|
return coordinate, context
|
|
|
|
|
|
def _native_http_call(driver: Driver, name: str, method: str, path: str,
|
|
payload: dict | None = None, *, expected: tuple[int, ...] = (200,),
|
|
headers: dict | None = None) -> dict:
|
|
try:
|
|
response = driver.client.request(method, path, json=payload, headers=headers)
|
|
except httpx.HTTPError:
|
|
driver.record(name, "BLOCKED", error_code="NATIVE_HTTP_TRANSPORT_UNAVAILABLE")
|
|
raise Blocked("NATIVE_HTTP_TRANSPORT_UNAVAILABLE") from None
|
|
try:
|
|
body = response.json()
|
|
except ValueError:
|
|
body = None
|
|
if response.status_code not in expected or not isinstance(body, dict):
|
|
detail = body.get("detail") if isinstance(body, dict) else None
|
|
code = detail.get("code") if isinstance(detail, dict) else None
|
|
if not isinstance(code, str) or re.fullmatch(r"[A-Z][A-Z0-9_]{1,95}", code) is None:
|
|
code = "NATIVE_HTTP_RESPONSE_UNAVAILABLE"
|
|
driver.record(name, "BLOCKED", http_status=response.status_code, error_code=code)
|
|
raise Blocked(code)
|
|
driver.record(name, "PASS", http_status=response.status_code)
|
|
return body
|
|
|
|
|
|
def create_official_filter_state(driver: Driver, target: dict, model: dict,
|
|
context_path: Path, dashboard_id: int) -> str:
|
|
"""Create and read back an exact Superset native-filter state for the pinned context."""
|
|
try:
|
|
context = json.loads(context_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
raise Blocked("NATIVE_FILTER_CONTEXT_UNAVAILABLE") from None
|
|
if (not isinstance(context, dict)
|
|
or context.get("provenance", {}).get("dashboard_id") != target["source_dashboard_id"]
|
|
or context.get("provenance", {}).get("environment_id") != target["target_environment_id"]
|
|
or context.get("canonical_request_filters") != {
|
|
"schema_version": 1,
|
|
"filters": target["expected_filter_context"],
|
|
"filters_hash": context.get("canonical_request_filters", {}).get("filters_hash"),
|
|
}):
|
|
raise Blocked("NATIVE_FILTER_CONTEXT_IDENTITY_MISMATCH")
|
|
if context["canonical_request_filters"].get("filters_hash") != target["expected_filters_hash"]:
|
|
raise Blocked("NATIVE_FILTER_CONTEXT_HASH_MISMATCH")
|
|
model_fp = model.get("query_model_fingerprint")
|
|
if context.get("provenance", {}).get("pinned_query_model_fingerprint") != model_fp:
|
|
raise Blocked("NATIVE_FILTER_CONTEXT_MODEL_MISMATCH")
|
|
current_chart = context.get("current_query_model_chart8", {})
|
|
if (current_chart.get("chart_uuid") != target["chart_uuid"]
|
|
or current_chart.get("dataset_uuid") != target["chart_contract"]["dataset_uuid"]):
|
|
raise Blocked("NATIVE_FILTER_CONTEXT_CHART_MISMATCH")
|
|
authored = context["canonical_request_filters"]["filters"]
|
|
controls = context.get("native_filter_context")
|
|
if not isinstance(controls, list) or len(controls) != len(authored):
|
|
raise Blocked("NATIVE_FILTER_CONTROL_CONTEXT_MISMATCH")
|
|
controls_by_id = {item.get("filter_id"): item for item in controls if isinstance(item, dict)}
|
|
if len(controls_by_id) != len(controls):
|
|
raise Blocked("NATIVE_FILTER_CONTROL_CONTEXT_MISMATCH")
|
|
chart_rows = [row for row in model.get("charts", []) if row.get("chart_uuid") == target["chart_uuid"]]
|
|
if len(chart_rows) != 1 or type(chart_rows[0].get("chart_id")) is not int:
|
|
raise Blocked("NATIVE_FILTER_TARGET_CHART_UNAVAILABLE")
|
|
selected_chart_id = chart_rows[0]["chart_id"]
|
|
data_mask = {}
|
|
for item in authored:
|
|
filter_id = item.get("filter_id")
|
|
control = controls_by_id.get(filter_id)
|
|
if (not isinstance(control, dict) or control.get("filter_type") != "NATIVE_FILTER"
|
|
or not any(target_row.get("chart_id") == selected_chart_id for target_row in control.get("targets", []))):
|
|
raise Blocked("NATIVE_FILTER_CONTROL_IDENTITY_MISMATCH")
|
|
value = item.get("value")
|
|
if item.get("operator") == "TEMPORAL_RANGE" and control.get("type") == "TIME":
|
|
if not isinstance(value, dict) or value.get("state") != "selected" or value.get("inclusive") is not False:
|
|
raise Blocked("NATIVE_FILTER_TEMPORAL_CONTEXT_INVALID")
|
|
period = f"{value.get('from')} : {value.get('to')}"
|
|
data_mask[filter_id] = {
|
|
"id": filter_id, "filterState": {"value": period},
|
|
"extraFormData": {"time_range": period}, "ownState": {},
|
|
}
|
|
elif item.get("operator") == "IN" and control.get("type") == "STRING":
|
|
state = value.get("state") if isinstance(value, dict) else None
|
|
if state == "cleared":
|
|
data_mask[filter_id] = {
|
|
"id": filter_id, "filterState": {}, "extraFormData": {}, "ownState": {},
|
|
}
|
|
elif state == "selected" and isinstance(value.get("values"), list) and value["values"]:
|
|
values = value["values"]
|
|
data_mask[filter_id] = {
|
|
"id": filter_id,
|
|
"filterState": {"value": values},
|
|
"extraFormData": {"filters": [{"col": item.get("column"),
|
|
"op": "IN", "val": values}]},
|
|
"ownState": {},
|
|
}
|
|
else:
|
|
raise Blocked("NATIVE_FILTER_SELECTION_INVALID")
|
|
else:
|
|
raise Blocked("NATIVE_FILTER_OPERATOR_UNSUPPORTED")
|
|
state_json = json.dumps(data_mask, separators=(",", ":"), ensure_ascii=False)
|
|
superset_password = os.environ["SUPERSET_PASSWORD"]
|
|
driver.secrets.append(superset_password)
|
|
origin = target.get("reference_origin")
|
|
if origin not in {"http://superset-dev:8088", "http://superset-preprod:8088", "http://superset-prod:8088"}:
|
|
raise Blocked("NATIVE_SUPERSET_REFERENCE_ORIGIN_INVALID")
|
|
with httpx.Client(base_url=target["superset_url"], timeout=45, trust_env=False) as superset:
|
|
login = superset.post("/api/v1/security/login", json={
|
|
"username": target["superset_username"], "password": superset_password,
|
|
"provider": "db", "refresh": True,
|
|
})
|
|
if login.status_code != 200 or not isinstance(login.json(), dict):
|
|
raise Blocked("NATIVE_SUPERSET_AUTH_UNAVAILABLE")
|
|
access_token = login.json().get("access_token")
|
|
if not isinstance(access_token, str) or not access_token:
|
|
raise Blocked("NATIVE_SUPERSET_AUTH_UNAVAILABLE")
|
|
driver.secrets.append(access_token)
|
|
headers = {"Authorization": "Bearer " + access_token}
|
|
csrf_response = superset.get("/api/v1/security/csrf_token/", headers=headers)
|
|
csrf_body = (csrf_response.json() if csrf_response.headers.get("content-type", "").startswith(
|
|
"application/json") else None)
|
|
csrf_token = csrf_body.get("result") if isinstance(csrf_body, dict) else None
|
|
if csrf_response.status_code != 200 or not isinstance(csrf_token, str) or not csrf_token:
|
|
driver.record("native-filter-state-csrf-token", "BLOCKED", method="GET",
|
|
path="/api/v1/security/csrf_token/", http_status=csrf_response.status_code,
|
|
content_type=csrf_response.headers.get("content-type", "").split(";", 1)[0],
|
|
error_code="NATIVE_SUPERSET_CSRF_UNAVAILABLE")
|
|
raise Blocked("NATIVE_SUPERSET_CSRF_UNAVAILABLE")
|
|
driver.secrets.append(csrf_token)
|
|
headers["X-CSRFToken"] = csrf_token
|
|
driver.record("native-filter-state-csrf-token", "PASS", method="GET",
|
|
path="/api/v1/security/csrf_token/", http_status=csrf_response.status_code,
|
|
content_type=csrf_response.headers.get("content-type", "").split(";", 1)[0])
|
|
filter_state_path = f"/api/v1/dashboard/{dashboard_id}/filter_state"
|
|
created = superset.post(f"/api/v1/dashboard/{dashboard_id}/filter_state",
|
|
json={"value": state_json}, headers=headers)
|
|
created_body = created.json() if created.headers.get("content-type", "").startswith("application/json") else None
|
|
if created.status_code != 201 or not isinstance(created_body, dict):
|
|
driver.record("native-filter-state-create", "BLOCKED", method="POST",
|
|
path=filter_state_path, http_status=created.status_code,
|
|
content_type=created.headers.get("content-type", "").split(";", 1)[0],
|
|
error_code="NATIVE_FILTER_STATE_CREATE_REFUSED")
|
|
raise Blocked("NATIVE_FILTER_STATE_CREATE_UNAVAILABLE")
|
|
key = created_body.get("result", {}).get("key") if isinstance(created_body.get("result"), dict) else None
|
|
if key is None:
|
|
key = created_body.get("key")
|
|
if not isinstance(key, str) or not key:
|
|
driver.record("native-filter-state-create", "BLOCKED", method="POST",
|
|
path=filter_state_path, http_status=created.status_code,
|
|
content_type=created.headers.get("content-type", "").split(";", 1)[0],
|
|
error_code="NATIVE_FILTER_STATE_KEY_UNAVAILABLE")
|
|
raise Blocked("NATIVE_FILTER_STATE_KEY_UNAVAILABLE")
|
|
driver.record("native-filter-state-create", "PASS", method="POST", path=filter_state_path,
|
|
http_status=created.status_code,
|
|
content_type=created.headers.get("content-type", "").split(";", 1)[0],
|
|
dashboard_id=dashboard_id, key_present=True, control_count=len(data_mask))
|
|
driver.secrets.append(key)
|
|
readback = superset.get(f"/api/v1/dashboard/{dashboard_id}/filter_state/{key}", headers=headers)
|
|
readback_body = readback.json() if readback.headers.get("content-type", "").startswith("application/json") else None
|
|
if readback.status_code != 200 or not isinstance(readback_body, dict):
|
|
driver.record("native-filter-state-readback", "BLOCKED", http_status=readback.status_code,
|
|
error_code="NATIVE_FILTER_STATE_READBACK_REFUSED")
|
|
raise Blocked("NATIVE_FILTER_STATE_READBACK_UNAVAILABLE")
|
|
result = readback_body.get("result")
|
|
value = result.get("value") if isinstance(result, dict) else readback_body.get("value")
|
|
try:
|
|
observed = json.loads(value) if isinstance(value, str) else None
|
|
except json.JSONDecodeError:
|
|
observed = None
|
|
if observed != data_mask:
|
|
raise Blocked("NATIVE_FILTER_STATE_READBACK_MISMATCH")
|
|
reference_url = origin + f"/superset/dashboard/{dashboard_id}/?" + urlencode({"native_filters_key": key})
|
|
driver.secrets.append(reference_url)
|
|
target["reference_url"] = reference_url
|
|
driver.record("native-filter-state-official-roundtrip", "PASS",
|
|
dashboard_id=dashboard_id, control_count=len(data_mask), key_present=True,
|
|
filters_hash=context["canonical_request_filters"]["filters_hash"])
|
|
return reference_url
|
|
|
|
|
|
class _SafeMcpEvidence:
|
|
"""Keep the existing MCP wire owner while recording only bounded event facts."""
|
|
|
|
def __init__(self, driver: Driver):
|
|
self.client = driver.client
|
|
self.driver = driver
|
|
|
|
def record(self, name: str, status: str, **facts) -> None:
|
|
safe = {key: value for key, value in facts.items()
|
|
if key in {"http_status"} and type(value) is int}
|
|
self.driver.record(name, status, **safe)
|
|
|
|
def require(self, name: str, condition: bool, **facts) -> None:
|
|
response = facts.get("response")
|
|
code = response.get("error") if isinstance(response, dict) else None
|
|
if not isinstance(code, str) or re.fullmatch(r"[A-Z][A-Z0-9_]{1,95}", code) is None:
|
|
code = None
|
|
self.driver.record(name, "PASS" if condition else "BLOCKED",
|
|
**({"error_code": code} if code and not condition else {}))
|
|
if not condition:
|
|
raise Blocked(code or "NATIVE_MCP_DOMAIN_REFUSED")
|
|
|
|
|
|
def capture_approve_publish(driver: Driver, target: dict, scenario: dict, catalog: dict,
|
|
model: dict) -> dict:
|
|
"""Run the official native-capture, approval, consume, and publication APIs."""
|
|
selection = catalog["selection"]
|
|
if selection.get("kind") not in {"current", "revision"}:
|
|
raise Blocked("NATIVE_CAPTURE_REQUIRES_EXPLICIT_CATALOG_SELECTION")
|
|
preview = _native_http_call(driver, "native-reference-discovery", "POST",
|
|
"/api/dashboard-testing/reference-discovery",
|
|
{"scenario_id": scenario["scenario_id"], "url": target["reference_url"]})
|
|
model_fingerprint = model.get("query_model_fingerprint")
|
|
if not isinstance(model_fingerprint, str) or re.fullmatch(r"sha256:[0-9a-f]{64}", model_fingerprint) is None:
|
|
raise Blocked("NATIVE_QUERY_MODEL_FINGERPRINT_UNAVAILABLE")
|
|
coordinate, native_context = select_native_chart_coordinate(preview, target, model_fingerprint, driver)
|
|
capture_body = {
|
|
"schema_version": 1,
|
|
"kind": "chart_response",
|
|
"scope": "native_query_result",
|
|
"scenario_id": scenario["scenario_id"],
|
|
"url": target["reference_url"],
|
|
"selected": [{"kind": "chart", "chart_id": coordinate["chart_id"],
|
|
"dataset_id": coordinate["dataset_id"], "metric": None}],
|
|
"expected_query_model_fingerprint": model_fingerprint,
|
|
"expected_filters_hash": native_context["filters_hash"],
|
|
"comparison_policy_intent": target["chart_contract"]["comparison_policy_intent"],
|
|
"catalog_selection": selection,
|
|
"native_filter_context": native_context,
|
|
}
|
|
captured = _native_http_call(driver, "native-chart-response-capture", "POST",
|
|
"/api/dashboard-testing/reference-selection/chart-response-capture", capture_body,
|
|
expected=(201,))
|
|
candidate_ids = captured.get("candidate_ids")
|
|
artifact_ids = captured.get("capture_artifact_ids")
|
|
release = target["release_identity"]
|
|
if (captured.get("kind") != "chart_response" or captured.get("scope") != "native_query_result"
|
|
or captured.get("release_id") != release["id"]
|
|
or not isinstance(captured.get("agent_run_id"), str)
|
|
or not isinstance(candidate_ids, list) or len(candidate_ids) != 1
|
|
or not isinstance(artifact_ids, list) or len(artifact_ids) != 1
|
|
or captured.get("capture_state") != "pending"):
|
|
driver.record("native-capture-response-binding", "BLOCKED",
|
|
error_code="NATIVE_CAPTURE_RESPONSE_IDENTITY_MISMATCH")
|
|
raise Blocked("NATIVE_CAPTURE_RESPONSE_IDENTITY_MISMATCH")
|
|
driver.record("native-capture-response-binding", "PASS", chart_id=coordinate["chart_id"],
|
|
dataset_id=coordinate["dataset_id"], query_model_fingerprint=model_fingerprint,
|
|
filters_hash=native_context["filters_hash"], candidate_count=1, artifact_count=1,
|
|
release_id=release["id"])
|
|
|
|
candidate_id = candidate_ids[0]
|
|
gate_path = f"/api/dashboard-testing/baseline-candidates/{candidate_id}/approval-gate"
|
|
gate = _native_http_call(driver, "request-native-candidate-approval", "POST", gate_path, {
|
|
"agent_run_id": captured["agent_run_id"], "release_version": release["version"],
|
|
"release_commit_hash": release["commit_hash"],
|
|
"reason": "Review isolated synthetic chart-8 native response"}, expected=(201,))
|
|
if gate.get("candidate_id") != candidate_id or gate.get("status") != "pending":
|
|
raise Blocked("NATIVE_APPROVAL_GATE_IDENTITY_MISMATCH")
|
|
gate_id = gate.get("gate_id")
|
|
if not isinstance(gate_id, str) or not gate_id:
|
|
raise Blocked("NATIVE_APPROVAL_GATE_UNAVAILABLE")
|
|
decided = _native_http_call(driver, "confirm-native-candidate-approval", "POST",
|
|
f"{gate_path}/{gate_id}/decide", {
|
|
"decision": "confirm", "reason": "Authorized synthetic chart-8 E2E approval"})
|
|
if decided.get("status") != "confirmed" or decided.get("gate_id") != gate_id:
|
|
raise Blocked("NATIVE_APPROVAL_DECISION_UNAVAILABLE")
|
|
consumed = _native_http_call(driver, "consume-native-candidate-approval", "POST",
|
|
f"{gate_path}/{gate_id}/consume?" + urlencode({
|
|
"release_version": release["version"], "release_commit_hash": release["commit_hash"]}))
|
|
exact_consumption = (consumed.get("consumed") is True and consumed.get("status") == "consumed"
|
|
and consumed.get("gate_id") == gate_id
|
|
and consumed.get("release_version") == release["version"]
|
|
and consumed.get("release_commit_hash") == release["commit_hash"]
|
|
and consumed.get("catalog_id") == selection["catalog_id"]
|
|
and isinstance(consumed.get("catalog_revision_id"), str)
|
|
and isinstance(consumed.get("baseline_id"), str)
|
|
and consumed.get("cas_version", 0) >= 1)
|
|
if not exact_consumption:
|
|
raise Blocked("NATIVE_APPROVAL_CONSUMPTION_IDENTITY_MISMATCH")
|
|
consumed_selection = {"kind": "revision", "catalog_id": consumed["catalog_id"],
|
|
"catalog_revision_id": consumed["catalog_revision_id"]}
|
|
revision = _native_http_call(driver, "native-catalog-revision-readback", "GET",
|
|
"/api/dashboard-testing/baselines?" + urlencode({
|
|
"catalog_id": consumed_selection["catalog_id"],
|
|
"catalog_revision_id": consumed_selection["catalog_revision_id"]}))
|
|
revision_body = revision.get("revision", {})
|
|
entries = revision_body.get("entry_revisions")
|
|
bound_entry = [entry for entry in entries if entry.get("baseline_id") == consumed["baseline_id"]
|
|
and entry.get("capture_artifact_id") in artifact_ids] if isinstance(entries, list) else []
|
|
if len(bound_entry) != 1 or not bound_entry[0].get("reference_source"):
|
|
raise Blocked("NATIVE_CATALOG_CAPTURE_BINDING_UNAVAILABLE")
|
|
|
|
publication_preview = _native_http_call(driver, "native-catalog-publication-preview", "POST",
|
|
"/api/catalog-publications/preview", {"catalog_selection": consumed_selection})
|
|
destination = publication_preview.get("destination")
|
|
if (publication_preview.get("catalog_id") != consumed_selection["catalog_id"]
|
|
or publication_preview.get("catalog_revision_id") != consumed_selection["catalog_revision_id"]
|
|
or not isinstance(publication_preview.get("expected_branch_head"), str)
|
|
or not isinstance(destination, dict)
|
|
or destination.get("repository") != target["catalog_repository"]
|
|
or destination.get("branch") != target["catalog_branch"]):
|
|
raise Blocked("NATIVE_CATALOG_PUBLICATION_TARGET_MISMATCH")
|
|
operation = _native_http_call(driver, "publish-native-catalog-revision", "POST",
|
|
"/api/catalog-publications", {
|
|
"catalog_selection": consumed_selection,
|
|
"expected_branch_head": publication_preview["expected_branch_head"],
|
|
}, expected=(200,), headers={"Idempotency-Key": "chart8-" + str(uuid4())})
|
|
operation_id = operation.get("operation_id")
|
|
if not isinstance(operation_id, str) or operation.get("catalog_revision_id") != consumed_selection["catalog_revision_id"]:
|
|
raise Blocked("NATIVE_CATALOG_PUBLICATION_OPERATION_UNAVAILABLE")
|
|
deadline = time.monotonic() + 180
|
|
while operation.get("state") not in {"published", "publish_failed", "conflict"} and time.monotonic() < deadline:
|
|
time.sleep(2)
|
|
operation = _native_http_call(driver, "native-catalog-publication-readback", "GET",
|
|
f"/api/catalog-publications/{operation_id}")
|
|
if (operation.get("state") != "published"
|
|
or operation.get("catalog_revision_id") != consumed_selection["catalog_revision_id"]
|
|
or not operation.get("published_receipt")
|
|
or operation.get("commit_sha") == publication_preview["expected_branch_head"]):
|
|
raise Blocked("NATIVE_CATALOG_PUBLICATION_UNAVAILABLE")
|
|
driver.record("native-catalog-publication-identity", "PASS",
|
|
catalog_id=consumed_selection["catalog_id"],
|
|
catalog_revision_id=consumed_selection["catalog_revision_id"],
|
|
operation_id=operation_id, commit_sha=operation["commit_sha"], state=operation["state"])
|
|
return {"capture": captured, "approval": consumed, "catalog_revision": revision_body,
|
|
"publication": operation, "catalog_selection": consumed_selection,
|
|
"coordinate": {"chart_id": coordinate["chart_id"], "dataset_id": coordinate["dataset_id"],
|
|
"chart_uuid": coordinate["chart_uuid"], "dataset_uuid": coordinate["dataset_uuid"]},
|
|
"query_model_fingerprint": model_fingerprint, "filters_hash": native_context["filters_hash"]}
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.NativeCapture
|
|
|
|
|
|
# #region Tooling.DashboardTestingE2E.NativeChart8Prepare.Main [C:3] [TYPE Function]
|
|
def explicit_inputs(args: argparse.Namespace) -> tuple[Path, Path, Path, Path, Path]:
|
|
if args.profile is not None:
|
|
if any(value is not None for value in
|
|
(args.env_file, args.seed, args.published_release, args.target_config, args.output)):
|
|
raise Blocked("NATIVE_INPUT_PROFILE_CONFLICT")
|
|
try:
|
|
info = args.profile.lstat()
|
|
if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600:
|
|
raise Blocked("NATIVE_PRIVATE_PROFILE_UNAVAILABLE")
|
|
profile = json.loads(args.profile.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
raise Blocked("NATIVE_PRIVATE_PROFILE_UNAVAILABLE") from None
|
|
fields = {"private_env_file", "seed", "published_release", "target_config", "output"}
|
|
if not isinstance(profile, dict) or set(profile) != fields or any(
|
|
type(profile.get(field)) is not str or not profile[field].strip() for field in fields
|
|
):
|
|
raise Blocked("NATIVE_PRIVATE_PROFILE_INVALID")
|
|
values = {}
|
|
for field in fields:
|
|
path = Path(profile[field]).expanduser()
|
|
values[field] = path if path.is_absolute() else args.profile.parent / path
|
|
return (values["private_env_file"], values["seed"], values["published_release"],
|
|
values["target_config"], values["output"])
|
|
required = (args.env_file, args.seed, args.published_release, args.target_config, args.output)
|
|
if any(value is None for value in required):
|
|
raise Blocked("NATIVE_EXPLICIT_INPUTS_REQUIRED")
|
|
return required
|
|
|
|
|
|
def validate_numeric_fixture_path(target: dict) -> Path:
|
|
numeric = target.get("numeric_verification") if isinstance(target, dict) else None
|
|
path_value = numeric.get("finance_fixture_seed_path") if isinstance(numeric, dict) else None
|
|
if type(path_value) is not str or not Path(path_value).is_absolute():
|
|
raise Blocked("NATIVE_FINANCE_FIXTURE_SEED_UNAVAILABLE")
|
|
path = Path(path_value)
|
|
try:
|
|
info = path.lstat()
|
|
except OSError:
|
|
raise Blocked("NATIVE_FINANCE_FIXTURE_SEED_UNAVAILABLE") from None
|
|
if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600:
|
|
raise Blocked("NATIVE_FINANCE_FIXTURE_SEED_UNAVAILABLE")
|
|
return path
|
|
|
|
|
|
def write_no_go(output: Path, reason: str) -> int:
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
output.joinpath("native-setup.json").write_text(json.dumps({
|
|
"overall": "NO-GO", "reason": reason,
|
|
}) + "\n")
|
|
return 2
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--profile", type=Path, help="Mode-0600 JSON containing explicit input paths")
|
|
parser.add_argument("--private-env-file", "--env-file", dest="env_file", type=Path)
|
|
parser.add_argument("--seed", type=Path)
|
|
parser.add_argument("--published-release", type=Path)
|
|
parser.add_argument("--target-config", type=Path)
|
|
parser.add_argument("--output", type=Path)
|
|
parser.add_argument("--validate-only", action="store_true",
|
|
help="Validate explicit inputs without making API calls")
|
|
lifecycle = parser.add_mutually_exclusive_group()
|
|
lifecycle.add_argument("--run-lifecycle", action="store_true",
|
|
help="Run chart-8 capture, publication, pin, and three-run acceptance")
|
|
lifecycle.add_argument("--resume-published", action="store_true",
|
|
help="Resume only M01 and numeric acceptance from one exact published capture")
|
|
args = parser.parse_args()
|
|
try:
|
|
env_file, seed_path, release_path, target_path, output = explicit_inputs(args)
|
|
except Blocked as exc:
|
|
reason = exc.args[0] if exc.args else "NATIVE_EXPLICIT_INPUTS_REQUIRED"
|
|
print(reason, file=sys.stderr)
|
|
return 2
|
|
try:
|
|
private_environment(env_file)
|
|
if not os.environ.get("FIXTURE_PASSWORD"):
|
|
raise Blocked("NATIVE_BACKEND_CREDENTIAL_UNAVAILABLE")
|
|
if not os.environ.get("SUPERSET_PASSWORD"):
|
|
raise Blocked("NATIVE_SUPERSET_CREDENTIAL_UNAVAILABLE")
|
|
except (OSError, ValueError, Blocked):
|
|
return write_no_go(output, "NATIVE_PRIVATE_CREDENTIALS_UNAVAILABLE")
|
|
try:
|
|
seed = json.loads(seed_path.read_text(encoding="utf-8"))
|
|
release = json.loads(release_path.read_text(encoding="utf-8"))
|
|
target = json.loads(target_path.read_text(encoding="utf-8"))
|
|
validate_target_descriptor(target, seed, release)
|
|
except Blocked as exc:
|
|
return write_no_go(output, exc.args[0] if exc.args else "NATIVE_TARGET_DESCRIPTOR_UNAVAILABLE")
|
|
except (OSError, json.JSONDecodeError, TypeError):
|
|
return write_no_go(output, "NATIVE_TARGET_DESCRIPTOR_UNAVAILABLE")
|
|
if args.validate_only:
|
|
try:
|
|
validate_numeric_fixture_path(target)
|
|
except Blocked as exc:
|
|
return write_no_go(output, exc.args[0] if exc.args else "NATIVE_FINANCE_FIXTURE_SEED_UNAVAILABLE")
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
output.joinpath("native-setup.json").write_text(json.dumps({
|
|
"status": "VALIDATED", "acceptance": "NOT_RUN",
|
|
"target_mode": target["target_mode"], "catalog_mode": target["catalog_mode"],
|
|
}) + "\n")
|
|
return 0
|
|
source_dashboard_id = target["source_dashboard_id"]
|
|
try:
|
|
validate_numeric_fixture_path(target)
|
|
if os.environ.get("BACKEND_USERNAME") != target["backend_username"]:
|
|
raise Blocked("NATIVE_BACKEND_IDENTITY_MISMATCH")
|
|
except Blocked as exc:
|
|
return write_no_go(output, exc.args[0] if exc.args else "NATIVE_NUMERIC_VERIFICATION_INPUT_UNAVAILABLE")
|
|
release_id = release.get("id")
|
|
driver = SafeNativeDriver({"dashboard_id": source_dashboard_id, "backend_url": target["backend_url"],
|
|
"username": target["backend_username"]}, output)
|
|
driver.phase = "native-chart8-prepare"
|
|
driver.ids["release_id"] = release_id
|
|
acceptance_completed = False
|
|
try:
|
|
driver.login()
|
|
dashboard_id, model = current_dashboard(driver, seed, target)
|
|
catalog = select_catalog(driver, target)
|
|
if target["catalog_mode"] == "seed_empty":
|
|
target = bind_seeded_catalog_selection(target, catalog)
|
|
reference_url = create_official_filter_state(driver, target, model,
|
|
Path(target["filter_context_path"]), dashboard_id)
|
|
scenario = register_capture_scenario(driver, dashboard_id, model,
|
|
target["target_environment_id"], target)
|
|
driver.ids.update(scenario_id=scenario["scenario_id"], revision_id=scenario.get("revision_id"),
|
|
source_dashboard_id=source_dashboard_id, target_dashboard_id=dashboard_id,
|
|
catalog_id=catalog["selection"]["catalog_id"],
|
|
catalog_revision_id=catalog["selection"].get("catalog_revision_id"))
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
output.joinpath("native-setup.json").write_text(json.dumps({
|
|
"overall": "NO-GO", "acceptance": "NOT_RUN",
|
|
"source_dashboard_id": source_dashboard_id, "target_dashboard_id": dashboard_id,
|
|
"chart_uuid": target["chart_uuid"],
|
|
"query_model_fingerprint": model.get("query_model_fingerprint"),
|
|
"scenario_id": scenario["scenario_id"], "scenario_revision_id": scenario.get("revision_id"),
|
|
"catalog_id": catalog["selection"]["catalog_id"],
|
|
"catalog_revision_id": catalog["selection"].get("catalog_revision_id"),
|
|
"filter_state_created": True, "reference_url_present": bool(reference_url),
|
|
"release": {"id": release_id, "version": release.get("version"),
|
|
"commit_hash": release.get("commit_hash"), "content_hash": release.get("content_hash")},
|
|
"target": {"source_environment_id": target["source_environment_id"],
|
|
"target_environment_id": target["target_environment_id"],
|
|
"source_dashboard_id": source_dashboard_id,
|
|
"target_dashboard_id": dashboard_id,
|
|
"source_dashboard_uuid": target["source_dashboard_uuid"],
|
|
"target_dashboard_uuid": target["target_dashboard_uuid"],
|
|
"chart_uuid": target["chart_uuid"]}},
|
|
indent=2, ensure_ascii=False) + "\n")
|
|
if args.run_lifecycle or args.resume_published:
|
|
if args.resume_published:
|
|
captured = read_published_capture(driver, target, model)
|
|
pin_target = target
|
|
else:
|
|
captured = capture_approve_publish(driver, target, scenario, catalog, model)
|
|
driver.ids.update(
|
|
capture_artifact_id=(captured["capture"].get("capture_artifact_ids") or [None])[0],
|
|
published_catalog_revision_id=captured["catalog_revision"].get("catalog_revision_id"),
|
|
publication_operation_id=captured["publication"].get("operation_id"),
|
|
)
|
|
driver.record("native-chart8-capture-and-publication", "PASS",
|
|
chart_id=captured["coordinate"]["chart_id"],
|
|
dataset_id=captured["coordinate"]["dataset_id"],
|
|
query_model_fingerprint=captured["query_model_fingerprint"],
|
|
filters_hash=captured["filters_hash"], capture_count=1, publication_state="published")
|
|
if not args.resume_published:
|
|
pin_target = target_for_published_revision(target, captured)
|
|
pin_scenario = register_m01_capture_scenario(driver, dashboard_id, model,
|
|
target["target_environment_id"], pin_target)
|
|
driver.ids.update(scenario_id=pin_scenario["scenario_id"],
|
|
revision_id=pin_scenario["revision_id"],
|
|
catalog_id=captured["catalog_selection"]["catalog_id"],
|
|
catalog_revision_id=captured["catalog_selection"]["catalog_revision_id"])
|
|
driver.record("native-chart8-post-publication-m01-binding", "PASS",
|
|
scenario_id=pin_scenario["scenario_id"], revision_id=pin_scenario["revision_id"],
|
|
catalog_id=captured["catalog_selection"]["catalog_id"],
|
|
catalog_revision_id=captured["catalog_selection"]["catalog_revision_id"])
|
|
verifier_target = numeric_verifier_target(pin_target, pin_scenario, captured)
|
|
verifier_evidence = run_numeric_verifier(driver, target, verifier_target, env_file, output)
|
|
verifier_runs = verifier_evidence["runs"]
|
|
driver.record("native-chart8-three-run-acceptance", "PASS",
|
|
pass_run_id=verifier_runs[0]["run_id"],
|
|
failure_run_id=verifier_runs[1]["run_id"],
|
|
restored_pass_run_id=verifier_runs[2]["run_id"],
|
|
restore_verified=True)
|
|
setup_path = output / "native-setup.json"
|
|
setup_report = json.loads(setup_path.read_text(encoding="utf-8"))
|
|
setup_report.update(overall="PASS", acceptance="PASS",
|
|
numeric_verification="numeric-verification/verification.json",
|
|
run_ids=[row["run_id"] for row in verifier_runs])
|
|
temp_path = output / "native-setup.json.tmp"
|
|
fd = os.open(temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as stream:
|
|
json.dump(setup_report, stream, sort_keys=True, indent=2, ensure_ascii=False)
|
|
stream.write("\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temp_path, setup_path)
|
|
os.chmod(setup_path, 0o600)
|
|
acceptance_completed = True
|
|
except (Blocked, KeyError, ValueError, httpx.HTTPError) as exc:
|
|
driver.record("native-chart8-prepare-stopped", "BLOCKED",
|
|
reason=exc.args[0] if exc.args else "NATIVE_CHART8_PREPARE_UNAVAILABLE")
|
|
finally:
|
|
driver.flush()
|
|
driver.client.close()
|
|
stopped = any(step["status"] in {"BLOCKED", "NEEDS_REVIEW"} for step in driver.steps)
|
|
if args.run_lifecycle or args.resume_published:
|
|
if not stopped and acceptance_completed:
|
|
driver.flush()
|
|
return 0
|
|
driver.record("native-chart8-acceptance-unavailable", "BLOCKED",
|
|
error_code="NATIVE_CHART8_ACCEPTANCE_INCOMPLETE")
|
|
driver.flush()
|
|
return 2
|
|
if not stopped:
|
|
driver.record("native-chart8-acceptance-unavailable", "BLOCKED",
|
|
reason="NATIVE_LIFECYCLE_NOT_REQUESTED")
|
|
return int(stopped)
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare.Main
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
# #endregion Tooling.DashboardTestingE2E.NativeChart8Prepare
|