"""完整诊断、分片流、持久状态和异常恢复出口的脱敏回归。"""
import asyncio
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "apps/market-api"))
from app.native.config import Settings
from app.native.operations import Operations
from app.native.store import Store
from app.native.supervisor import Supervisor


class RedactionTests(unittest.IsolatedAsyncioTestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory(prefix="message-redaction-")
        self.root = Path(self.temporary.name)
        self.settings = Settings(data_dir=self.root, admin_token="fixture-control-admin", mcp_tokens={"a": "fixture-control-mcp-a"}, allowed_hosts=("localhost",), allowed_origins=("http://localhost",), plugin_env_allowlist={"a": ["REDACTION_FIXTURE_SECRET"]})
        self.store = Store(self.root)
        self.supervisor = Supervisor(self.settings, self.store)
        self.secret = 'fixture-中文-quoted"-secret'

    def tearDown(self):
        self.store.close()
        self.temporary.cleanup()

    async def test_every_byte_boundary_is_redacted_after_drain(self):
        raw = self.secret.encode()
        with patch.dict(os.environ, {"REDACTION_FIXTURE_SECRET": self.secret}):
            for boundary in range(1, len(raw)):
                stream = asyncio.StreamReader()
                worker = SimpleNamespace(diagnostics="")
                draining = asyncio.create_task(self.supervisor.drain("a", worker, stream))
                stream.feed_data(raw[:boundary])
                await asyncio.sleep(0)
                stream.feed_data(raw[boundary:])
                stream.feed_eof()
                await draining
                self.assertEqual(worker.diagnostics, "[REDACTED]")

    async def test_store_and_error_reconciliation_have_final_boundary(self):
        with patch.dict(os.environ, {"REDACTION_FIXTURE_SECRET": self.secret}):
            self.store.save_profile({"id": "a", "last_error": self.secret})
            self.supervisor.mutate = AsyncMock(side_effect=RuntimeError("operation: " + self.secret))
            self.supervisor.refresh = AsyncMock(side_effect=RuntimeError("refresh: " + self.secret + " " + self.settings.admin_token))
            operations = Operations(self.store, self.supervisor)
            queued = operations.submit("a", "restart", {}, None)
            await operations.tasks[queued["id"]]
            completed = self.store.operation(queued["id"])
            self.assertEqual(completed["status"], "failed")
            self.assertNotIn(self.secret, json.dumps(completed, ensure_ascii=False))
            self.assertIn("[REDACTED]", completed["error"])
            self.assertIn("[REDACTED]", self.store.profile("a")["last_error"])
            # 即使未来调用点遗漏过滤，持久化入口也不能存入原始诊断值。
            completed.update(error=self.secret, result={"diagnostic": self.secret, "token": self.settings.mcp_tokens["a"]})
            self.store.save_operation(completed)
            for forbidden in (self.secret, self.settings.admin_token, self.settings.mcp_tokens["a"]):
                for path in self.root.glob("control.sqlite3*"):
                    self.assertNotIn(forbidden.encode(), path.read_bytes())

    async def test_cached_values_and_json_escaping_are_redacted(self):
        with patch.dict(os.environ, {"REDACTION_FIXTURE_SECRET": self.secret}):
            self.supervisor.redact("a", "")
        for value in (self.secret, json.dumps(self.secret, ensure_ascii=True)[1:-1], json.dumps(self.secret, ensure_ascii=False)[1:-1]):
            self.assertEqual(self.supervisor.redact("a", value), "[REDACTED]")
        self.assertEqual(self.supervisor.redact("b", self.secret), self.secret)
        self.assertEqual(self.supervisor.redact(None, self.secret), "[REDACTED]")