from __future__ import annotations

import os
import subprocess
import sys
import unittest
from pathlib import Path

if __package__:
    from .dashboard_common import (
        expressions_match,
        find_graph_panel,
        find_row,
        load_dashboard,
        quota_expressions,
        row_expressions,
        target_expressions,
        thread_cpu_selectors,
    )
else:
    sys.path.insert(0, str(Path(__file__).parent.parent))
    from tests.dashboard_common import (
        expressions_match,
        find_graph_panel,
        find_row,
        load_dashboard,
        quota_expressions,
        row_expressions,
        target_expressions,
        thread_cpu_selectors,
    )

ROOT = Path(__file__).parent.parent
REPO_ROOT = ROOT.parent.parent
GENERATOR = REPO_ROOT / "scripts" / "gen-tikv-details-dashboard"
DETAILS_SOURCE = ROOT / "tikv_details.dashboard.py"
DETAILS_JSON = ROOT / "tikv_details.json"
SERVERLESS_DETAILS_JSON = ROOT / "tikv_details_serverless.json"
SERVERLESS_WORKER_JSON = ROOT / "tikv_worker_serverless.json"


def setUpModule() -> None:
    if os.environ.get("TIKV_DASHBOARD_FIXTURES_READY") == "1":
        return

    for output in (SERVERLESS_DETAILS_JSON, SERVERLESS_WORKER_JSON):
        output.unlink(missing_ok=True)
    _ = subprocess.run(
        [str(GENERATOR)],
        check=True,
        cwd=REPO_ROOT,
    )


class DashboardContractsTest(unittest.TestCase):
    def test_thread_cpu_pair_selectors_use_sanitized_thread_names(self):
        selectors = thread_cpu_selectors(DETAILS_SOURCE)
        self.assertTrue(selectors)
        self.assertFalse([selector for selector in selectors if "-" in selector])

    def test_serverless_ia_and_cluster_resources_select_tikv_container(self):
        dashboard = load_dashboard(SERVERLESS_DETAILS_JSON)
        ia_expressions = row_expressions(find_row(dashboard, "IA"))
        self.assertTrue(expressions_match(ia_expressions, ('container="tikv"',)))

        cluster_row = find_row(dashboard, "Cluster")
        for panel_title in ("CPU", "Memory", "Uptime"):
            with self.subTest(panel=panel_title):
                expressions = target_expressions(
                    find_graph_panel(cluster_row, panel_title)
                )
                self.assertTrue(
                    expressions_match(expressions, ('container="tikv"',), ("job=",))
                )

    def test_serverless_worker_dashboard_owns_worker_rows_and_configuration(self):
        details = load_dashboard(SERVERLESS_DETAILS_JSON)
        worker = load_dashboard(SERVERLESS_WORKER_JSON)
        worker_rows = (
            "Cloud Worker Service",
            "Cloud Worker - Native BR",
            "Cloud Worker - Schema Manager",
            "IA",
            "Thread CPU",
        )
        exclusive_worker_rows = {
            "Cloud Worker Service",
            "Cloud Worker - Native BR",
            "Cloud Worker - Schema Manager",
        }

        self.assertEqual(worker["__inputs"], details["__inputs"])
        self.assertCountEqual(
            [panel["title"] for panel in worker["panels"]], worker_rows
        )
        self.assertTrue(
            exclusive_worker_rows.isdisjoint(
                panel["title"] for panel in details["panels"]
            )
        )

    def test_serverless_worker_dashboard_shows_worker_thread_cpu(self):
        dashboard = load_dashboard(SERVERLESS_WORKER_JSON)
        expressions = row_expressions(find_row(dashboard, "Thread CPU"))

        self.assertEqual(len(expressions), 2)
        for metric in (
            "tikv_worker_thread_cpu_seconds_total",
            "tikv_worker_thread_cpu_detail_seconds_total",
        ):
            with self.subTest(metric=metric):
                self.assertTrue(any(metric in expression for expression in expressions))
        for expression in expressions:
            with self.subTest(expression=expression):
                self.assertIn('instance=~"${worker_instance:regex}"', expression)
                self.assertIn('container=~"tikv-worker|cop-worker"', expression)

    def test_serverless_worker_dashboard_uses_only_worker_instance_variable(self):
        dashboard = load_dashboard(SERVERLESS_WORKER_JSON)
        variable_names = {
            variable["name"] for variable in dashboard["templating"]["list"]
        }
        expressions = [
            expression
            for row in dashboard["panels"]
            for expression in row_expressions(row)
        ]

        self.assertNotIn("instance", variable_names)
        self.assertIn("worker_instance", variable_names)
        self.assertTrue(expressions)
        for expression in expressions:
            with self.subTest(expression=expression):
                self.assertTrue(
                    "${worker_instance:regex}" in expression
                    or "$worker_instance" in expression
                )
                self.assertNotIn("${instance:regex}", expression)

    def test_serverless_native_br_targets_select_worker_instances(self):
        dashboard = load_dashboard(SERVERLESS_WORKER_JSON)
        expressions = row_expressions(find_row(dashboard, "Cloud Worker - Native BR"))

        self.assertTrue(expressions)
        for expression in expressions:
            with self.subTest(expression=expression):
                self.assertIn('instance=~"${worker_instance:regex}"', expression)

    def test_serverless_remote_compaction_selects_worker_instances(self):
        dashboard = load_dashboard(SERVERLESS_DETAILS_JSON)
        compaction_row = find_row(dashboard, "Kv Engine - Compaction")
        expressions = target_expressions(
            find_graph_panel(
                compaction_row,
                "Remote Compaction Exceed Memory Limit Rate",
            )
        )

        self.assertEqual(len(expressions), 1)
        self.assertIn('instance=~"${worker_instance:regex}"', expressions[0])

    def test_cold_region_row_is_serverless_only_and_has_expected_metrics(self):
        details = load_dashboard(DETAILS_JSON)
        serverless_details = load_dashboard(SERVERLESS_DETAILS_JSON)

        self.assertNotIn("Cold Region", [panel["title"] for panel in details["panels"]])
        expressions = row_expressions(find_row(serverless_details, "Cold Region"))
        expected_metrics = (
            "rfstore_cold_regions_count",
            "rfstore_cold_region_state_transitions_total",
            "rfstore_cold_leader_probe_pending_count",
            "rfstore_cold_campaign_events_total",
        )

        self.assertEqual(len(expressions), 4)
        for metric in expected_metrics:
            with self.subTest(metric=metric):
                self.assertTrue(any(metric in expression for expression in expressions))

    def test_serverless_worker_prefetch_panel_uses_cop_worker_selectors(self):
        dashboard = load_dashboard(SERVERLESS_WORKER_JSON)
        ia_row = find_row(dashboard, "IA")
        worker_service_row = find_row(dashboard, "Cloud Worker Service")

        ia_expressions = row_expressions(ia_row)
        self.assertTrue(
            expressions_match(ia_expressions, ('container=~"tikv-worker|cop-worker"',))
        )
        self.assertNotIn(
            "Cop Worker IA Prefetch Cache Hit",
            [panel.get("title", "") for panel in ia_row["panels"]],
        )

        expressions = target_expressions(
            find_graph_panel(worker_service_row, "Cop Worker IA Prefetch Cache Hit")
        )
        self.assertTrue(
            expressions_match(
                expressions,
                ('container="cop-worker"', 'instance=~"$worker_instance"'),
            )
        )

    def test_serverless_worker_resource_panels_use_container_limits(self):
        dashboard = load_dashboard(SERVERLESS_WORKER_JSON)
        worker_service_row = find_row(dashboard, "Cloud Worker Service")
        worker_cases = (("Copr", "cop-worker"), ("TiKV", "tikv-worker"))
        resource_cases = (
            ("CPU", ("container_spec_cpu_quota", "container_spec_cpu_period", " / ")),
            ("Memory", ("container_spec_memory_limit_bytes", " > 0")),
        )

        all_expressions: list[str] = []
        quota_legends = {"quota-{{instance}}", "max-quota", "min-quota"}
        for worker, container in worker_cases:
            for resource, quota_fragments in resource_cases:
                panel_title = f"{worker} Worker {resource}"
                panel = find_graph_panel(worker_service_row, panel_title)
                expressions = target_expressions(panel)
                all_expressions.extend(expressions)
                with self.subTest(panel=panel_title):
                    self.assertTrue(
                        expressions_match(
                            expressions, (f'container="{container}"',), ("job=",)
                        )
                    )
                    quotas = quota_expressions(panel, quota_legends)
                    self.assertEqual(len(quotas), 3)
                    self.assertTrue(expressions_match(quotas, quota_fragments))

        for metric in (
            "tikv_server_cpu_cores_quota",
            "tikv_server_memory_quota_bytes",
        ):
            with self.subTest(deprecated_metric=metric):
                self.assertFalse(
                    [
                        expression
                        for expression in all_expressions
                        if metric in expression
                    ]
                )

    def test_worker_quota_targets_use_process_selectors(self):
        dashboard = load_dashboard(DETAILS_JSON)
        worker_service_row = find_row(dashboard, "Cloud Worker Service")
        quota_legends = {"quota-{{instance}}", "max-quota", "min-quota"}
        process_cases = (
            (("Copr Worker CPU", "Copr Worker Memory"), "coprocessor-worker"),
            (("TiKV Worker CPU", "TiKV Worker Memory"), "tikv-worker"),
        )

        for panel_titles, job in process_cases:
            with self.subTest(job=job):
                quotas = [
                    expression
                    for panel_title in panel_titles
                    for expression in quota_expressions(
                        find_graph_panel(worker_service_row, panel_title), quota_legends
                    )
                ]
                self.assertEqual(len(quotas), 6)
                self.assertTrue(
                    expressions_match(quotas, (f'job=~"{job}"',), ("container=",))
                )


if __name__ == "__main__":
    _ = unittest.main()