from __future__ import annotations

import json
import unittest

from ai_triage.feishu import (
    FeishuError,
    FeishuUserResolver,
    FeishuWebhookNotifier,
    resolve_owner_mentions,
)


class RecordingTransport:
    def __init__(
        self,
        status: int = 200,
        response: bytes = b'{"code":0,"msg":"ok"}',
    ) -> None:
        self.status = status
        self.response = response
        self.calls: list[tuple[str, bytes, dict[str, str], float]] = []

    def __call__(
        self,
        url: str,
        body: bytes,
        headers: dict[str, str],
        timeout: float,
    ) -> tuple[int, bytes]:
        self.calls.append((url, body, headers, timeout))
        return self.status, self.response


class SequencedTransport:
    def __init__(self, responses: list[tuple[int, bytes]]) -> None:
        self.responses = list(responses)
        self.calls: list[tuple[str, bytes, dict[str, str], float]] = []

    def __call__(
        self,
        url: str,
        body: bytes,
        headers: dict[str, str],
        timeout: float,
    ) -> tuple[int, bytes]:
        self.calls.append((url, body, headers, timeout))
        return self.responses.pop(0)


def notification(**overrides: object) -> dict[str, object]:
    value: dict[str, object] = {
        "event_id": "dedicated:101:2:201",
        "target_id": "dedicated",
        "workflow": "scenario.yaml",
        "run_id": 101,
        "run_attempt": 2,
        "run_url": "https://github.invalid/runs/101",
        "job_id": 201,
        "job_name": "gcp (case.yaml)",
        "job_url": "https://github.invalid/jobs/201",
        "incidents": [
            {
                "failed_step": "Run scenario cases",
                "triage": {
                    "category": "product_error",
                    "component": "dedicated-control-plane",
                    "problem_key": "cluster-stuck-creating",
                    "owner": "component-owner",
                    "owner_type": "component",
                    "confidence": 0.9,
                    "summary": "state did not converge",
                    "root_cause": "service remained in creating",
                    "recommendations": ["inspect trace"],
                },
                "evidence": {
                    "scenario_yaml": "ng_backups_operation.yaml",
                    "request_method": "GET",
                    "request_url": "https://api.example.com/v1/clusters/1",
                    "expected": "body.state == ACTIVE",
                    "actual_status": 200,
                    "response_body": {"state": "CREATING"},
                    "trace_id": "trace-1",
                },
            }
        ],
    }
    value.update(overrides)
    return value


class FeishuWebhookNotifierTests(unittest.TestCase):
    def test_resolves_email_owners_to_open_ids_with_app_credentials(self) -> None:
        transport = SequencedTransport(
            [
                (
                    200,
                    b'{"code":0,"tenant_access_token":"tenant-token","expire":7200}',
                ),
                (
                    200,
                    json.dumps(
                        {
                            "code": 0,
                            "data": {
                                "user_list": [
                                    {
                                        "email": "case@example.com",
                                        "user_id": "ou_case",
                                    },
                                    {
                                        "email": "infra@example.com",
                                        "user_id": "ou_infra",
                                    },
                                ]
                            },
                        }
                    ).encode(),
                ),
            ]
        )

        mentions = resolve_owner_mentions(
            "app-id",
            "app-secret",
            {
                "product": "component-name",
                "case": "case@example.com",
                "infrastructure": "infra@example.com",
            },
            transport=transport,
        )

        self.assertEqual(
            {"case": "ou_case", "infrastructure": "ou_infra"},
            mentions,
        )
        self.assertEqual(2, len(transport.calls))
        self.assertNotIn("app-secret", str(mentions))
        self.assertNotIn("tenant-token", str(mentions))

    def test_sends_card2_topic_with_columns_highlights_and_owner_mention(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            owner_mentions={"component": "ou_component"},
            transport=transport,
        )
        value = notification()
        value["incidents"][0]["triage"]["owner_type"] = "component"

        notifier.send_failure(value)

        payload = json.loads(transport.calls[0][1])
        card = payload["card"]
        serialized = json.dumps(card, ensure_ascii=False)
        self.assertEqual("interactive", payload["msg_type"])
        self.assertEqual("2.0", card["schema"])
        self.assertEqual("red", card["header"]["template"])
        self.assertNotIn("icon", card["header"])
        self.assertEqual(
            "🚨 场景 E2E 失败",
            card["header"]["title"]["content"],
        )
        column_sets = [
            element
            for element in card["body"]["elements"]
            if element.get("tag") == "column_set"
        ]
        self.assertTrue(
            any(len(element.get("columns", [])) >= 2 for element in column_sets)
        )
        self.assertIn("state did not converge", serialized)
        self.assertIn("service remained in creating", serialized)
        self.assertIn("<at id=ou_component></at>", serialized)
        self.assertTrue(
            any(
                behavior.get("type") == "open_url"
                and behavior.get("default_url")
                == "https://github.invalid/jobs/201"
                for element in _walk_card(card)
                if element.get("tag") == "button"
                for behavior in element.get("behaviors", [])
            )
        )

    def test_product_issue_mentions_live_oncall_and_shows_selected_team(self) -> None:
        transport = RecordingTransport()
        resolved_emails: list[tuple[str, ...]] = []

        def resolve(emails: tuple[str, ...]) -> dict[str, str]:
            resolved_emails.append(emails)
            return {"oncall@example.com": "ou_oncall"}

        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            email_resolver=resolve,
            transport=transport,
        )
        value = notification()
        value["incidents"][0]["triage"].update(
            {
                "owner_type": "product_oncall",
                "owner": "oncall@example.com",
                "owner_emails": ["oncall@example.com"],
                "oncall_team": "global_control",
                "oncall_team_display": "SOE - Global & Dedicated",
            }
        )

        notifier.send_failure(value)

        text = transport.calls[0][1].decode()
        self.assertEqual([("oncall@example.com",)], resolved_emails)
        self.assertIn("SOE - Global & Dedicated", text)
        self.assertIn("当前 Oncall", text)
        self.assertIn("<at id=ou_oncall></at>", text)

    def test_case_issue_labels_configured_owner_as_case_owner(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            owner_mentions={"case": "ou_case"},
            transport=transport,
        )
        value = notification()
        value["incidents"][0]["triage"].update(
            {
                "category": "case_error",
                "owner_type": "case",
                "owner": "case@example.com",
                "oncall_team": "unknown",
            }
        )

        notifier.send_failure(value)

        payload = json.loads(transport.calls[0][1])
        text = transport.calls[0][1].decode()
        self.assertEqual("orange", payload["card"]["header"]["template"])
        self.assertIn("Case Owner（配置）", text)
        self.assertIn("<at id=ou_case></at>", text)

    def test_infrastructure_issue_uses_orange_card_header(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            owner_mentions={"infrastructure": "ou_infra"},
            transport=transport,
        )
        value = notification()
        value["incidents"][0]["triage"].update(
            {
                "category": "infrastructure_error",
                "owner_type": "infrastructure",
                "owner": "infra@example.com",
                "oncall_team": "unknown",
            }
        )

        notifier.send_failure(value)

        payload = json.loads(transport.calls[0][1])
        self.assertEqual("orange", payload["card"]["header"]["template"])

    def test_partial_ai_result_keeps_existing_fields_and_marks_missing_fields_gray(
        self,
    ) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            owner_mentions={"case": "ou_case"},
            transport=transport,
        )
        value = notification()
        value["incidents"][0]["triage"].update(
            {
                "analysis_status": "ai_partial",
                "missing_fields": ["oncall_team"],
                "owner_type": "case",
                "owner": "case@example.com",
                "oncall_team": "unknown",
                "summary": "服务端返回了 HTTP 500",
                "root_cause": "产品接口处理请求时发生内部错误",
            }
        )

        notifier.send_failure(value)

        payload = json.loads(transport.calls[0][1])
        text = transport.calls[0][1].decode()
        markdown = "\n".join(
            element.get("content", "")
            for element in _walk_card(payload["card"])
            if element.get("tag") == "markdown"
        )
        self.assertIn("服务端返回了 HTTP 500", text)
        self.assertIn("产品接口处理请求时发生内部错误", text)
        self.assertIn(
            "<font color='grey'>AI 返回结果缺失字段：oncall\\_team</font>",
            markdown,
        )
        self.assertIn("Case Owner（配置）", text)
        self.assertNotIn("AI 分析结果不可用，已按保守策略归类", text)

    def test_feishu_user_resolver_caches_dynamic_email_lookups(self) -> None:
        transport = SequencedTransport(
            [
                (
                    200,
                    b'{"code":0,"tenant_access_token":"tenant-token","expire":7200}',
                ),
                (
                    200,
                    json.dumps(
                        {
                            "code": 0,
                            "data": {
                                "user_list": [
                                    {
                                        "email": "oncall@example.com",
                                        "user_id": "ou_oncall",
                                    }
                                ]
                            },
                        }
                    ).encode(),
                ),
            ]
        )
        resolver = FeishuUserResolver(
            "app-id",
            "app-secret",
            transport=transport,
        )

        first = resolver.resolve(("oncall@example.com",))
        second = resolver.resolve(("oncall@example.com",))

        self.assertEqual({"oncall@example.com": "ou_oncall"}, first)
        self.assertEqual(first, second)
        self.assertEqual(2, len(transport.calls))

    def test_card_excludes_internal_fields_and_keeps_failure_link(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )

        notifier.send_failure(notification())

        url, body, headers, _ = transport.calls[0]
        payload = json.loads(body)
        text = body.decode()
        self.assertEqual("https://secret.invalid/hook/token", url)
        self.assertEqual("application/json", headers["Content-Type"])
        self.assertEqual("interactive", payload["msg_type"])
        self.assertNotIn("dedicated:101:2:201", text)
        self.assertNotIn("product_error", text)
        self.assertNotIn("confidence", text)
        self.assertNotIn("problem_key", text)
        self.assertIn("body.state == ACTIVE", text)
        self.assertIn("CREATING", text)
        self.assertIn("https://github.invalid/jobs/201", text)
        self.assertNotIn("secret.invalid", text)

    def test_card_uses_chinese_user_facing_labels_and_defaults(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )
        value = notification()
        value["incidents"][0]["triage"].update(
            {
                "summary": "集群状态未在预期时间内收敛",
                "root_cause": "控制面持续返回创建中状态",
                "recommendations": ["根据链路追踪 ID 检查控制面日志"],
                "owner": "",
            }
        )

        notifier.send_failure(value)

        text = transport.calls[0][1].decode()
        for expected in (
            "请求：",
            "预期结果：",
            "实际状态码：",
            "实际响应：",
            "链路追踪 ID：",
            "测试用例：",
            "运行 101",
            "尝试 2",
            "未配置",
            "Dedicated 控制面",
        ):
            self.assertIn(expected, text)
        for english_label in (
            "Request:",
            "Expected:",
            "Actual status:",
            "Actual response:",
            "Trace ID:",
            "Testcase：",
            "Run 101",
            "Attempt 2",
            "unassigned",
            "not captured",
            "dedicated-control-plane",
        ):
            self.assertNotIn(english_label, text)

    def test_failure_details_do_not_repeat_top_level_diagnosis_as_phenomenon(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )

        notifier.send_failure(notification())

        text = transport.calls[0][1].decode()
        self.assertEqual(1, text.count("state did not converge"))
        self.assertEqual(1, text.count("service remained in creating"))
        self.assertNotIn("现象", text)

    def test_failure_summary_includes_related_metadata(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )
        value = notification()
        value["incidents"][0]["evidence"]["related_metadata"] = {
            "clusterId": ["cluster-1234"],
            "resourcePoolId": ["pool-9876"],
            "orgId": ["org-1", "org-2"],
            "restoreId": ["restore-1122"],
            "changefeedId": ["changefeed-3344"],
        }

        notifier.send_failure(value)

        payload = json.loads(transport.calls[0][1])
        text = transport.calls[0][1].decode()
        failure_text = next(
            element["content"]
            for element in payload["card"]["body"]["elements"]
            if element.get("tag") == "markdown"
            and "clusterId：cluster-1234" in element.get("content", "")
        )
        self.assertNotIn("相关元数据", text)
        self.assertIn("clusterId：cluster-1234", text)
        self.assertIn("resourcePoolId：pool-9876", text)
        self.assertIn("orgId：org-1、org-2", text)
        self.assertIn("restoreId：restore-1122", text)
        self.assertIn("changefeedId：changefeed-3344", text)
        self.assertIn(
            "ng_backups_operation.yaml)\n- clusterId：cluster-1234",
            failure_text,
        )
        self.assertIn(
            "changefeedId：changefeed-3344\n\n- 请求",
            failure_text,
        )

    def test_identical_diagnosis_and_root_cause_are_rendered_once(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )
        value = notification()
        value["incidents"][0]["triage"]["root_cause"] = "state did not converge"

        notifier.send_failure(value)

        text = transport.calls[0][1].decode()
        self.assertEqual(1, text.count("state did not converge"))
        self.assertNotIn("根因判断", text)

    def test_failure_details_link_to_scenario_testcase_yaml(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )

        notifier.send_failure(notification())

        text = transport.calls[0][1].decode()
        self.assertIn(
            "https://github.com/tidbcloud/APITestGen/blob/main/"
            "scenario_testcases/ng_backups_operation.yaml",
            text,
        )

    def test_testcase_link_falls_back_to_yaml_in_job_name(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )
        value = notification(job_name="nonprod (ng_manual_backup_restore.yaml)")
        value["incidents"][0]["evidence"]["scenario_yaml"] = None

        notifier.send_failure(value)

        text = transport.calls[0][1].decode()
        self.assertIn(
            "https://github.com/tidbcloud/APITestGen/blob/main/"
            "scenario_testcases/ng_manual_backup_restore.yaml",
            text,
        )

    def test_aggregated_topic_lists_every_failed_job_link(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )
        first = notification()["incidents"][0]
        second = json.loads(json.dumps(first))
        second["triage"]["summary"] = "another cluster did not converge"
        topic = {
            "target_id": "dedicated",
            "workflow": "scenario.yaml",
            "run_id": 101,
            "run_attempt": 2,
            "run_url": "https://github.invalid/runs/101",
            "event_ids": ["dedicated:101:2:201", "dedicated:101:2:202"],
            "topic_key": "topic-internal-only",
            "progress_keys": {
                "dedicated:101:2:201": ["failure-internal-only-1"],
                "dedicated:101:2:202": ["failure-internal-only-2"],
            },
            "failures": [
                {
                    "job_id": 201,
                    "job_name": "gcp (case-a.yaml)",
                    "job_url": "https://github.invalid/jobs/201",
                    **first,
                },
                {
                    "job_id": 202,
                    "job_name": "gcp (case-b.yaml)",
                    "job_url": "https://github.invalid/jobs/202",
                    **second,
                },
            ],
        }

        notifier.send_failure(topic)

        text = transport.calls[0][1].decode()
        self.assertIn("2 个失败任务", text)
        self.assertIn("https://github.invalid/jobs/201", text)
        self.assertIn("https://github.invalid/jobs/202", text)
        self.assertNotIn("dedicated:101:2:201", text)
        self.assertNotIn("topic-internal-only", text)
        self.assertNotIn("failure-internal-only", text)

    def test_aggregated_topic_button_opens_the_workflow_run(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )
        first = notification()["incidents"][0]
        topic = {
            "target_id": "dedicated",
            "workflow": "scenario.yaml",
            "run_id": 101,
            "run_attempt": 2,
            "run_url": "https://github.invalid/runs/101",
            "failures": [
                {
                    "job_id": job_id,
                    "job_name": f"gcp (case-{job_id}.yaml)",
                    "job_url": f"https://github.invalid/jobs/{job_id}",
                    **json.loads(json.dumps(first)),
                }
                for job_id in (201, 202)
            ],
        }

        notifier.send_failure(topic)

        card = json.loads(transport.calls[0][1])["card"]
        destinations = [
            behavior.get("default_url")
            for element in _walk_card(card)
            if element.get("tag") == "button"
            for behavior in element.get("behaviors", [])
            if behavior.get("type") == "open_url"
        ]
        self.assertEqual(["https://github.invalid/runs/101"], destinations)

    def test_large_group_stays_within_custom_webhook_body_limit(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            transport=transport,
            max_text_chars=20_000,
        )
        first = notification()["incidents"][0]
        topic = {
            "target_id": "dedicated",
            "workflow": "scenario.yaml",
            "run_id": 101,
            "run_attempt": 2,
            "run_url": "https://github.invalid/runs/101",
            "failures": [
                {
                    "job_id": job_id,
                    "job_name": f"gcp (case-{job_id}.yaml)",
                    "job_url": f"https://github.invalid/jobs/{job_id}",
                    **json.loads(json.dumps(first)),
                }
                for job_id in range(200, 300)
            ],
        }

        notifier.send_failure(topic)

        body = transport.calls[0][1]
        self.assertLessEqual(len(body), 20_000)
        self.assertIn(b"https://github.invalid/jobs/200", body)
        self.assertIn("100 个失败任务".encode(), body)
        self.assertNotIn("另有 95 个失败任务".encode(), body)

    def test_large_group_shows_two_details_three_more_links_and_no_empty_lists(
        self,
    ) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            transport=transport,
            max_text_chars=20_000,
        )
        first = notification()["incidents"][0]
        failures = []
        for job_id in range(200, 210):
            failure = {
                "job_id": job_id,
                "job_name": f"gcp (case-{job_id}.yaml)",
                "job_url": f"https://github.invalid/jobs/{job_id}",
                **json.loads(json.dumps(first)),
            }
            failure["evidence"] = {}
            failure["triage"]["recommendations"] = []
            failures.append(failure)
        topic = {
            "target_id": "dedicated",
            "workflow": "scenario.yaml",
            "run_id": 101,
            "run_attempt": 2,
            "run_url": "https://github.invalid/runs/101",
            "failures": failures,
        }

        notifier.send_failure(topic)

        card = json.loads(transport.calls[0][1])["card"]
        markdown = "\n".join(
            element.get("content", "")
            for element in _walk_card(card)
            if element.get("tag") == "markdown"
        )
        self.assertIn("10 个失败任务", markdown)
        self.assertIn("**失败任务 1", markdown)
        self.assertIn("**失败任务 2", markdown)
        self.assertNotIn("**失败任务 3", markdown)
        for job_id in (202, 203, 204):
            self.assertIn(f"https://github.invalid/jobs/{job_id}", markdown)
        self.assertNotIn("https://github.invalid/jobs/205", markdown)
        more_failures = next(
            element["content"]
            for element in _walk_card(card)
            if element.get("tag") == "markdown"
            and element.get("content", "").startswith("**更多失败任务**")
        )
        self.assertEqual(
            3,
            sum(line.startswith("- [") for line in more_failures.splitlines()),
        )
        self.assertNotIn("另有", more_failures)
        self.assertNotIn("建议操作", markdown)
        self.assertNotIn("\n- \n", markdown)
        self.assertNotIn("\n-\n", markdown)

    def test_empty_owner_is_unassigned_and_text_is_truncated(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            transport=transport,
            max_text_chars=220,
        )
        value = notification()
        value["incidents"][0]["triage"]["owner"] = ""
        value["incidents"][0]["triage"]["summary"] = "x" * 1000

        notifier.send_failure(value)

        text = transport.calls[0][1].decode()
        self.assertIn("未配置", text)
        self.assertIn("已截断", text)
        self.assertNotIn("truncated", text)
        self.assertLess(len(text), 2_000)

    def test_redacts_untrusted_fields_and_link_query(self) -> None:
        transport = RecordingTransport()
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token", transport=transport
        )
        value = notification(job_url="https://github.invalid/jobs/201?token=leak")
        value["incidents"][0]["triage"]["summary"] = "Authorization: Bearer should-not-leak"

        notifier.send_failure(value)

        text = transport.calls[0][1].decode()
        self.assertNotIn("should-not-leak", text)
        self.assertNotIn("token=leak", text)

    def test_http_failure_does_not_include_webhook_secret(self) -> None:
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            transport=RecordingTransport(status=500, response=b"failure"),
        )

        with self.assertRaises(FeishuError) as caught:
            notifier.send_failure(notification())

        self.assertNotIn("secret.invalid", str(caught.exception))

    def test_feishu_api_error_is_reported(self) -> None:
        notifier = FeishuWebhookNotifier(
            "https://secret.invalid/hook/token",
            transport=RecordingTransport(
                response=b'{"code":19001,"msg":"bad request"}'
            ),
        )

        with self.assertRaises(FeishuError):
            notifier.send_failure(notification())


def _walk_card(value: object):
    if isinstance(value, dict):
        yield value
        for child in value.values():
            yield from _walk_card(child)
    elif isinstance(value, list):
        for child in value:
            yield from _walk_card(child)


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