"""T-103 脱敏器测试：全部证据为合成数据，绝不读取真实 raw 目录。"""

from __future__ import annotations

from hashlib import sha256
import json
from pathlib import Path
import sys
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch
from xml.etree import ElementTree

from PIL import Image


CLIENT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(CLIENT_ROOT / "src"))

from cmbuyer_client.device.sku_evidence_sanitizer import (
    EXPECTED_GOODS_ID,
    EXPECTED_SCREENSHOT_HEIGHT,
    EXPECTED_SCREENSHOT_WIDTH,
    EXPECTED_XML_HEIGHT,
    EXPECTED_XML_WIDTH,
    HUMAN_DECLARED_STATES,
    SkuEvidenceSanitizationError,
    sanitize_sku_panel_evidence,
)


TEST_SERIAL = "synthetic-serial-never-publish"
TEST_ADDRESS = "SYNTHETIC_ADDRESS_NEVER_PUBLISH"
FULL_PHONE = "13800138000"
MASKED_PHONE = "138****0000"
SAFE_TEXT = "synthetic-safe-lower-content"


def _hash(path: Path) -> str:
    digest = sha256()
    with path.open("rb") as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _default_xml() -> str:
    return (
        "<hierarchy rotation='0'>"
        f"<node bounds='[0,0][1080,540]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE} {FULL_PHONE}' />"
        f"<node bounds='[0,540][1080,2376]' text='{SAFE_TEXT}' />"
        "</hierarchy>"
    )


def _write_raw(
    root: Path,
    *,
    state: str = "panel-opened-target-preselected",
    goods_id: str = EXPECTED_GOODS_ID,
    model: str = "PKG110",
    android_version: str = "16",
    pdd_version: str = "8.17.0",
    size: tuple[int, int] = (EXPECTED_SCREENSHOT_WIDTH, EXPECTED_SCREENSHOT_HEIGHT),
    xml: str | None = None,
) -> Path:
    raw = root / "raw"
    raw.mkdir(parents=True)
    screenshot = raw / "screenshot.png"
    image = Image.new("RGB", size, color=(0, 180, 0))
    if size == (EXPECTED_SCREENSHOT_WIDTH, EXPECTED_SCREENSHOT_HEIGHT):
        for y in range(540):
            for x in range(8):
                image.putpixel((x, y), (255, 0, 0))
    image.save(screenshot, format="PNG")
    hierarchy = raw / "hierarchy.xml"
    hierarchy.write_text(_default_xml() if xml is None else xml, encoding="utf-8")
    manifest = {
        "schema_version": 1,
        "product": {
            "goods_id": goods_id,
            "canonical_url": f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}",
        },
        "human_declared_state": state,
        "serial": TEST_SERIAL,
        "channel": "wifi",
        "device": {
            "model": model,
            "android_version": android_version,
            "pdd_package": "com.xunmeng.pinduoduo",
            "pdd_version": pdd_version,
        },
        "artifacts": [
            {"path": "screenshot.png", "sha256": _hash(screenshot)},
            {"path": "hierarchy.xml", "sha256": _hash(hierarchy)},
        ],
    }
    (raw / "manifest.json").write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8")
    return raw


class SkuEvidenceSanitizerTests(unittest.TestCase):
    def test_all_declared_states_mask_screenshot_and_xml_without_raw_metadata(self) -> None:
        for state in sorted(HUMAN_DECLARED_STATES):
            with self.subTest(state=state), TemporaryDirectory() as temporary:
                raw = _write_raw(Path(temporary), state=state)
                result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
                with Image.open(result.screenshot_path) as image:
                    self.assertEqual(image.getpixel((0, 0)), (0, 0, 0, 255))
                    self.assertEqual(image.getpixel((100, 600)), (0, 180, 0, 255))
                derived_xml = result.hierarchy_path.read_text(encoding="utf-8")
                manifest = result.manifest_path.read_text(encoding="utf-8")

                self.assertIn(SAFE_TEXT, derived_xml)
                self.assertNotIn(TEST_ADDRESS, derived_xml)
                self.assertNotIn(FULL_PHONE, derived_xml)
                self.assertNotIn(MASKED_PHONE, derived_xml)
                self.assertIn(f'"human_declared_state": "{state}"', manifest)
                self.assertIn('"privacy_tier": "SANITIZED"', manifest)
                self.assertIn('"sanitizer_version": "t103-privacy-v3"', manifest)
                self.assertIn('"screenshot_space": {', manifest)
                self.assertIn('"xml_coordinate_space": {', manifest)
                self.assertIn('"height": 2376', manifest)
                self.assertIn('"height": 2376', manifest)
                self.assertIn('"privacy_mask_rectangle": [', manifest)
                self.assertIn('"removed_nodes": 1', manifest)
                self.assertIn('"cleared_crossing_nodes": 0', manifest)
                self.assertIn('"retained_below_nodes": 1', manifest)
                self.assertIn('"max_right": 1080', manifest)
                self.assertIn('"max_bottom": 2376', manifest)
                self.assertNotIn("canonical_url", manifest)
                self.assertNotIn(TEST_SERIAL, manifest)
                self.assertNotIn("serial", manifest)
                self.assertNotIn("channel", manifest)
                self.assertNotIn(TEST_ADDRESS, manifest)
                self.assertNotIn(FULL_PHONE, manifest)

    def test_crossing_container_keeps_lower_children_but_clears_its_text(self) -> None:
        xml = (
            "<hierarchy>"
            f"<node bounds='[0,0][1080,2376]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE}'>"
            f"<node bounds='[0,0][1080,540]' text='{FULL_PHONE}' />"
            f"<node bounds='[0,540][1080,2376]' text='{SAFE_TEXT}' />"
            "</node></hierarchy>"
        )
        with TemporaryDirectory() as temporary:
            raw = _write_raw(Path(temporary), xml=xml)
            result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
            root = ElementTree.parse(result.hierarchy_path).getroot()
            crossing = root.find("node")

            self.assertIsNotNone(crossing)
            assert crossing is not None
            self.assertEqual(crossing.attrib, {"bounds": "[0,0][1080,2376]"})
            self.assertEqual(len(list(crossing)), 1)
            self.assertEqual(list(crossing)[0].get("text"), SAFE_TEXT)
            manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
            self.assertEqual(
                manifest["privacy_cleanup"],
                {
                    "removed_nodes": 1,
                    "cleared_crossing_nodes": 1,
                    "retained_below_nodes": 1,
                    "max_right": 1080,
                    "max_bottom": 2376,
                },
            )

    def test_same_raw_and_config_produce_identical_derived_files(self) -> None:
        with TemporaryDirectory() as temporary:
            root = Path(temporary)
            left_raw = _write_raw(root / "left")
            right_raw = _write_raw(root / "right")
            left = sanitize_sku_panel_evidence(left_raw, left_raw.parent / "derived")
            right = sanitize_sku_panel_evidence(right_raw, right_raw.parent / "derived")

            for left_path, right_path in (
                (left.screenshot_path, right.screenshot_path),
                (left.hierarchy_path, right.hierarchy_path),
                (left.manifest_path, right.manifest_path),
            ):
                self.assertEqual(left_path.read_bytes(), right_path.read_bytes())

    def test_manifest_records_distinct_screenshot_and_xml_coordinate_spaces(self) -> None:
        with TemporaryDirectory() as temporary:
            raw = _write_raw(Path(temporary))
            result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
            manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))

            self.assertEqual(
                manifest["screenshot_space"],
                {"width": 1080, "height": 2376, "privacy_mask_rectangle": [0, 0, 1080, 540]},
            )
            self.assertEqual(
                manifest["xml_coordinate_space"],
                {
                    "width": 1080,
                    "height": 2376,
                    "privacy_mask_rectangle": [0, 0, 1080, 540],
                    "observed_max": {"right": 1080, "bottom": 2376},
                },
            )

    def test_hash_and_metadata_mismatch_fail_closed_without_leak(self) -> None:
        scenarios = (
            ("hash", {}, "screenshot"),
            ("model", {"model": "other"}, None),
            ("android", {"android_version": "15"}, None),
            ("version", {"pdd_version": "8.17.1"}, None),
            ("goods", {"goods_id": "123"}, None),
            ("state", {"state": "guessed"}, None),
            ("old-screenshot-space", {"size": (1080, 2400)}, None),
            ("other-screenshot-space", {"size": (100, 100)}, None),
        )
        for name, kwargs, corrupt_file in scenarios:
            with self.subTest(name=name), TemporaryDirectory() as temporary:
                raw = _write_raw(Path(temporary), **kwargs)
                if corrupt_file is not None:
                    (raw / f"{corrupt_file}.png").write_bytes(b"changed")
                with self.assertRaises(SkuEvidenceSanitizationError) as raised:
                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")

                message = str(raised.exception)
                self.assertNotIn(TEST_SERIAL, message)
                self.assertNotIn(TEST_ADDRESS, message)
                self.assertNotIn(FULL_PHONE, message)
                self.assertFalse((raw.parent / "derived").exists())
                self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])

    def test_old_and_unknown_human_states_are_rejected(self) -> None:
        for state in ("initial", "one-dimension-selected", "all-dimensions-selected", "guessed"):
            with self.subTest(state=state), TemporaryDirectory() as temporary:
                raw = _write_raw(Path(temporary), state=state)
                with self.assertRaises(SkuEvidenceSanitizationError):
                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
                self.assertFalse((raw.parent / "derived").exists())

    def test_malformed_inputs_and_bounds_or_phone_residue_fail_closed(self) -> None:
        malformed = (
            ("manifest", None),
            ("png", None),
            ("xml", None),
            ("bounds", "<hierarchy><node text='missing bounds' /></hierarchy>"),
            (
                "private-parent-with-lower-child",
                "<hierarchy><node bounds='[0,0][1080,540]'><node bounds='[0,540][1080,2376]' text='x' /></node></hierarchy>",
            ),
            ("full-phone-below", f"<hierarchy><node bounds='[0,540][1080,2376]' text='{FULL_PHONE}' /></hierarchy>"),
            ("masked-phone-below", f"<hierarchy><node bounds='[0,540][1080,2376]' text='{MASKED_PHONE}' /></hierarchy>"),
        )
        for kind, xml in malformed:
            with self.subTest(kind=kind), TemporaryDirectory() as temporary:
                raw = _write_raw(Path(temporary), xml=xml)
                if kind == "manifest":
                    (raw / "manifest.json").write_text("{invalid", encoding="utf-8")
                elif kind == "png":
                    (raw / "screenshot.png").write_bytes(b"not a png")
                    manifest = json.loads((raw / "manifest.json").read_text(encoding="utf-8"))
                    manifest["artifacts"][0]["sha256"] = _hash(raw / "screenshot.png")
                    (raw / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
                elif kind == "xml":
                    (raw / "hierarchy.xml").write_text("<hierarchy>", encoding="utf-8")
                    manifest = json.loads((raw / "manifest.json").read_text(encoding="utf-8"))
                    manifest["artifacts"][1]["sha256"] = _hash(raw / "hierarchy.xml")
                    (raw / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
                with self.assertRaises(SkuEvidenceSanitizationError):
                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")

                self.assertFalse((raw.parent / "derived").exists())
                self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])

    def test_phone_recheck_rejects_separator_mask_and_cross_node_bypasses(self) -> None:
        variants = (
            "138 0013-8000",
            "138＊＊＊＊0000",
            "138••••0000",
            "138xxxx0000",
            "138XXXX0000",
        )
        for value in variants:
            xml = (
                "<hierarchy>"
                "<node bounds='[0,0][1080,540]' text='private' />"
                f"<node bounds='[0,540][1080,2376]' text='{value}' />"
                "</hierarchy>"
            )
            with self.subTest(value=value), TemporaryDirectory() as temporary:
                raw = _write_raw(Path(temporary), xml=xml)
                with self.assertRaises(SkuEvidenceSanitizationError):
                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
                self.assertFalse((raw.parent / "derived").exists())

        split_xml = (
            "<hierarchy>"
            "<node bounds='[0,0][1080,540]' text='private' />"
            "<node bounds='[0,540][1080,1000]' text='138' content-desc='0013' />"
            "<node bounds='[0,1000][1080,2376]' text='8000' />"
            "</hierarchy>"
        )
        with TemporaryDirectory() as temporary:
            raw = _write_raw(Path(temporary), xml=split_xml)
            with self.assertRaises(SkuEvidenceSanitizationError):
                sanitize_sku_panel_evidence(raw, raw.parent / "derived")
            self.assertFalse((raw.parent / "derived").exists())

    def test_privacy_geometry_requires_removed_and_retained_nodes(self) -> None:
        scenarios = (
            ("no-private", f"<hierarchy><node bounds='[0,540][1080,2376]' text='{SAFE_TEXT}' /></hierarchy>"),
            ("no-below", "<hierarchy><node bounds='[0,0][1080,540]' text='private' /></hierarchy>"),
        )
        for name, xml in scenarios:
            with self.subTest(name=name), TemporaryDirectory() as temporary:
                raw = _write_raw(Path(temporary), xml=xml)
                with self.assertRaises(SkuEvidenceSanitizationError):
                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")
                self.assertFalse((raw.parent / "derived").exists())
                self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])

    def test_xml_coordinate_space_must_have_exact_configured_maximums(self) -> None:
        scenarios = (
            (
                "short-width",
                "<hierarchy><node bounds='[0,0][1079,540]' text='private' />"
                "<node bounds='[0,540][1079,2376]' text='safe' /></hierarchy>",
                "1079x2376",
            ),
            (
                "short-height",
                "<hierarchy><node bounds='[0,0][1080,540]' text='private' />"
                "<node bounds='[0,540][1080,2375]' text='safe' /></hierarchy>",
                "1080x2375",
            ),
            (
                "wide-width",
                "<hierarchy><node bounds='[0,0][1081,540]' text='private' />"
                "<node bounds='[0,540][1081,2376]' text='safe' /></hierarchy>",
                "1081x2376",
            ),
            (
                "old-v2-xml-height",
                "<hierarchy><node bounds='[0,0][1080,540]' text='private' />"
                "<node bounds='[0,540][1080,2400]' text='safe' /></hierarchy>",
                "1080x2400",
            ),
        )
        for name, xml, observed in scenarios:
            with self.subTest(name=name), TemporaryDirectory() as temporary:
                raw = _write_raw(Path(temporary), xml=xml)
                with self.assertRaises(SkuEvidenceSanitizationError) as raised:
                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")

                self.assertIn(observed, str(raised.exception))
                self.assertNotIn(TEST_SERIAL, str(raised.exception))
                self.assertFalse((raw.parent / "derived").exists())
                self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])

    def test_manifest_schema_package_and_artifact_structure_are_required(self) -> None:
        def mutate(manifest: dict[str, object], kind: str) -> None:
            if kind == "schema":
                manifest["schema_version"] = 2
            elif kind == "package":
                manifest["device"]["pdd_package"] = "com.example.other"  # type: ignore[index]
            elif kind == "missing":
                manifest.pop("artifacts")
            elif kind == "duplicate":
                manifest["artifacts"].append(manifest["artifacts"][0])  # type: ignore[index]
            elif kind == "bad-hash":
                manifest["artifacts"][0]["sha256"] = "g" * 64  # type: ignore[index]

        for kind in ("schema", "package", "missing", "duplicate", "bad-hash"):
            with self.subTest(kind=kind), TemporaryDirectory() as temporary:
                raw = _write_raw(Path(temporary))
                manifest_path = raw / "manifest.json"
                manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
                mutate(manifest, kind)
                manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
                with self.assertRaises(SkuEvidenceSanitizationError) as raised:
                    sanitize_sku_panel_evidence(raw, raw.parent / "derived")

                self.assertNotIn(TEST_SERIAL, str(raised.exception))
                self.assertNotIn(TEST_ADDRESS, str(raised.exception))
                self.assertFalse((raw.parent / "derived").exists())

    def test_target_created_during_publish_is_preserved_and_staging_is_removed(self) -> None:
        with TemporaryDirectory() as temporary:
            raw = _write_raw(Path(temporary))
            target = raw.parent / "derived"

            def race_rename(destination: Path) -> None:
                destination.mkdir()
                (destination / "sentinel.txt").write_text("keep", encoding="utf-8")
                raise FileExistsError("simulated publish race")

            with patch("cmbuyer_client.device.sku_evidence_sanitizer.Path.rename", side_effect=race_rename):
                with self.assertRaises(SkuEvidenceSanitizationError):
                    sanitize_sku_panel_evidence(raw, target)

            self.assertEqual((target / "sentinel.txt").read_text(encoding="utf-8"), "keep")
            self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])

    def test_existing_derived_is_preserved_without_reading_or_writing_raw(self) -> None:
        with TemporaryDirectory() as temporary:
            raw = _write_raw(Path(temporary))
            target = raw.parent / "derived"
            target.mkdir()
            sentinel = target / "sentinel.txt"
            sentinel.write_text("keep", encoding="utf-8")

            with self.assertRaises(SkuEvidenceSanitizationError):
                sanitize_sku_panel_evidence(raw, target)

            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
            self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])

    def test_directory_contract_rejects_non_sibling_paths(self) -> None:
        with TemporaryDirectory() as temporary:
            root = Path(temporary)
            raw = _write_raw(root)
            with self.assertRaises(SkuEvidenceSanitizationError):
                sanitize_sku_panel_evidence(raw, root / "not-derived")
            with self.assertRaises(SkuEvidenceSanitizationError):
                sanitize_sku_panel_evidence(root / "not-raw", root / "derived")