from __future__ import annotations

import hashlib
import io
import json
import stat
import tarfile
from pathlib import Path

import pytest

from muse.基础设施 import 数据归档
from muse.基础设施.备份恢复 import 备份恢复错误, 解包备份
from muse.基础设施.数据归档 import 创建文件归档, 归档错误, 恢复文件归档, 核对文件归档


def _模式(路径: Path) -> int:
    return stat.S_IMODE(路径.stat().st_mode)


def _建立基本来源(tmp_path: Path) -> tuple[Path, Path]:
    raw = tmp_path / "raw"
    drafts = tmp_path / "drafts"
    (raw / "证据").mkdir(parents=True)
    drafts.mkdir()
    (raw / "证据" / "原字节.bin").write_bytes(b"\x00\xff\nraw\x80")
    (drafts / "第一章 草稿.txt").write_text("风从旧城来。\n", encoding="utf-8")
    return raw, drafts


@pytest.mark.case_id(
    "NC-w29-29a001",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["原字节_unicode_布局和私有权限往返"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_原字节_unicode_布局和私有权限往返__29a001(tmp_path: Path) -> None:
    raw, drafts = _建立基本来源(tmp_path)
    (raw / "证据" / "原字节.bin").chmod(0o644)
    归档 = tmp_path / "backup"

    清单 = 创建文件归档({"raw": raw, "drafts": drafts}, 归档)
    恢复raw = tmp_path / "restored-raw"
    恢复drafts = tmp_path / "restored-drafts"
    恢复结果 = 恢复文件归档(归档, {"raw": 恢复raw, "drafts": 恢复drafts})

    assert 恢复结果 == 清单 == 核对文件归档(归档)
    assert (恢复raw / "证据" / "原字节.bin").read_bytes() == b"\x00\xff\nraw\x80"
    assert (恢复drafts / "第一章 草稿.txt").read_text(encoding="utf-8") == "风从旧城来。\n"
    assert 清单["file_count"] == 2
    assert 清单["total_bytes"] == len(b"\x00\xff\nraw\x80") + len("风从旧城来。\n".encode())
    assert _模式(归档) == _模式(恢复raw) == _模式(恢复drafts) == 0o700
    assert _模式(归档 / "manifest.json") == 0o600
    assert _模式(恢复raw / "证据" / "原字节.bin") == 0o600
    清单正文 = (归档 / "manifest.json").read_text(encoding="utf-8")
    assert str(raw) not in 清单正文
    assert "风从旧城来" not in 清单正文


@pytest.mark.case_id(
    "NC-w29-29a002",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["空根和缺失根保持区别"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_空根和缺失根保持区别__29a002(tmp_path: Path) -> None:
    raw = tmp_path / "raw"
    (raw / "空子目录").mkdir(parents=True)
    drafts = tmp_path / "missing-drafts"
    归档 = tmp_path / "backup"

    清单 = 创建文件归档({"raw": raw, "drafts": drafts}, 归档)
    恢复raw = tmp_path / "restored-raw"
    恢复drafts = tmp_path / "restored-drafts"
    恢复文件归档(归档, {"raw": 恢复raw, "drafts": 恢复drafts})

    assert 清单["roots"]["raw"] == {
        "exists": True,
        "directories": ["空子目录"],
        "files": [],
    }
    assert 清单["roots"]["drafts"] == {"exists": False, "directories": [], "files": []}
    assert (恢复raw / "空子目录").is_dir()
    assert not 恢复drafts.exists()


@pytest.mark.case_id(
    "NC-w29-29a003",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["归档文件被篡改时拒绝且不恢复"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_归档文件被篡改时拒绝且不恢复__29a003(tmp_path: Path) -> None:
    raw, drafts = _建立基本来源(tmp_path)
    归档 = tmp_path / "backup"
    创建文件归档({"raw": raw, "drafts": drafts}, 归档)
    (归档 / "data" / "raw" / "证据" / "原字节.bin").write_bytes(b"tampered")
    恢复raw = tmp_path / "restored-raw"
    恢复drafts = tmp_path / "restored-drafts"

    with pytest.raises(归档错误, match="^归档文件大小或哈希与清单不一致$"):
        恢复文件归档(归档, {"raw": 恢复raw, "drafts": 恢复drafts})

    assert not 恢复raw.exists()
    assert not 恢复drafts.exists()


@pytest.mark.case_id(
    "NC-w29-29a004",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["清单路径逃逸在恢复前拒绝"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_清单路径逃逸在恢复前拒绝__29a004(tmp_path: Path) -> None:
    raw, drafts = _建立基本来源(tmp_path)
    归档 = tmp_path / "backup"
    创建文件归档({"raw": raw, "drafts": drafts}, 归档)
    清单路径 = 归档 / "manifest.json"
    清单 = json.loads(清单路径.read_text(encoding="utf-8"))
    清单["roots"]["raw"]["files"][0]["path"] = "../escape.bin"
    清单路径.write_text(json.dumps(清单), encoding="utf-8")
    恢复raw = tmp_path / "restored-raw"
    恢复drafts = tmp_path / "restored-drafts"

    with pytest.raises(归档错误, match="^归档文件路径不安全或重复$"):
        恢复文件归档(归档, {"raw": 恢复raw, "drafts": 恢复drafts})

    assert not (tmp_path / "escape.bin").exists()
    assert not 恢复raw.exists()
    assert not 恢复drafts.exists()


@pytest.mark.case_id(
    "NC-w29-29a005",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["来源符号链接被拒绝并清理归档"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_来源符号链接被拒绝并清理归档__29a005(tmp_path: Path) -> None:
    raw = tmp_path / "raw"
    drafts = tmp_path / "drafts"
    raw.mkdir()
    drafts.mkdir()
    外部 = tmp_path / "outside.txt"
    外部.write_text("不可跟随", encoding="utf-8")
    (raw / "link.txt").symlink_to(外部)
    归档 = tmp_path / "backup"

    with pytest.raises(归档错误, match="^归档根含符号链接或非普通文件$"):
        创建文件归档({"raw": raw, "drafts": drafts}, 归档)

    assert not 归档.exists()
    assert 外部.read_text(encoding="utf-8") == "不可跟随"


@pytest.mark.case_id(
    "NC-w29-29a006",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["已有恢复目标不覆盖且其他根不创建"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_已有恢复目标不覆盖且其他根不创建__29a006(tmp_path: Path) -> None:
    raw, drafts = _建立基本来源(tmp_path)
    归档 = tmp_path / "backup"
    创建文件归档({"raw": raw, "drafts": drafts}, 归档)
    恢复raw = tmp_path / "restored-raw"
    恢复raw.mkdir()
    标记 = 恢复raw / "keep.txt"
    标记.write_text("保留", encoding="utf-8")
    恢复drafts = tmp_path / "restored-drafts"

    with pytest.raises(归档错误, match="^恢复目标目录必须全新$"):
        恢复文件归档(归档, {"raw": 恢复raw, "drafts": 恢复drafts})

    assert 标记.read_text(encoding="utf-8") == "保留"
    assert not 恢复drafts.exists()


@pytest.mark.case_id(
    "NC-w29-29a007",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["多根恢复中断清理本次半成品"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_多根恢复中断清理本次半成品__29a007(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    raw, drafts = _建立基本来源(tmp_path)
    归档 = tmp_path / "backup"
    创建文件归档({"raw": raw, "drafts": drafts}, 归档)
    原复制 = 数据归档._复制归档文件到目标
    次数 = 0

    def 中断(根句柄: int, 根名: str, 条目: dict, 目标根: Path) -> None:
        nonlocal 次数
        次数 += 1
        if 次数 == 2:
            raise OSError("synthetic interruption")
        原复制(根句柄, 根名, 条目, 目标根)

    monkeypatch.setattr(数据归档, "_复制归档文件到目标", 中断)
    恢复raw = tmp_path / "restored-raw"
    恢复drafts = tmp_path / "restored-drafts"

    with pytest.raises(归档错误, match="^文件归档恢复失败$") as 捕获:
        恢复文件归档(归档, {"raw": 恢复raw, "drafts": 恢复drafts})

    assert 捕获.value.上下文 == {"原因类型": "OSError"}
    assert 次数 == 2
    assert not 恢复raw.exists()
    assert not 恢复drafts.exists()


@pytest.mark.case_id(
    "NC-w29-29a008",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["未知根和已有输出目录均被拒绝"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_未知根和已有输出目录均被拒绝__29a008(tmp_path: Path) -> None:
    raw = tmp_path / "raw"
    raw.mkdir()
    未创建归档 = tmp_path / "not-created"

    with pytest.raises(归档错误, match="^文件根名称只允许raw和drafts$"):
        创建文件归档({"assets": raw}, 未创建归档)
    assert not 未创建归档.exists()

    已有归档 = tmp_path / "existing"
    已有归档.mkdir()
    标记 = 已有归档 / "keep.txt"
    标记.write_text("保留", encoding="utf-8")
    with pytest.raises(归档错误, match="^文件归档输出目录必须全新$"):
        创建文件归档({"raw": raw}, 已有归档)
    assert 标记.read_text(encoding="utf-8") == "保留"


@pytest.mark.case_id(
    "NC-w29-29a009",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["清单外文件不能静默夹带"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
def test_清单外文件不能静默夹带__29a009(tmp_path: Path) -> None:
    raw, drafts = _建立基本来源(tmp_path)
    归档 = tmp_path / "backup"
    创建文件归档({"raw": raw, "drafts": drafts}, 归档)
    夹带 = 归档 / "data" / "raw" / "extra.bin"
    夹带.write_bytes(b"extra")
    夹带.chmod(0o600)

    with pytest.raises(归档错误, match="^归档含清单外内容或缺少已登记内容$"):
        核对文件归档(归档)


def _合成完整备份(tmp_path: Path) -> tuple[Path, str]:
    """合法离线封装；PGDMP 仅为合成头，不调用数据库客户端。"""
    raw, drafts = _建立基本来源(tmp_path)
    目录 = tmp_path / "valid-backup"
    目录.mkdir(mode=0o700)
    文件清单 = 创建文件归档({"raw": raw, "drafts": drafts}, 目录 / "files")
    dump = b"PGDMP-synthetic-offline-only"
    (目录 / "database.dump").write_bytes(dump)
    (目录 / "database.dump").chmod(0o600)
    清单 = {
        "schema_version": "muse-full-backup-v1",
        "backup_id": "",
        "resource_release": "synthetic",
        "resource_build_id": "1" * 64,
        "database": {
            "archive": "database.dump",
            "size": len(dump),
            "sha256": hashlib.sha256(dump).hexdigest(),
            "source_fingerprint": "2" * 64,
            "fingerprint_scope": "cluster-database-v1",
            "server_version_num": "160000",
            "migrations": [{"version": 1, "name": "V0001__synthetic.sql", "checksum": "3" * 64}],
            "summary": {
                key: []
                for key in (
                    "schemas",
                    "relations",
                    "table_data",
                    "constraints",
                    "sequences",
                    "default_privileges",
                )
            },
        },
        "file_archive": {
            "directory": "files",
            "manifest_sha256": hashlib.sha256(
                (目录 / "files/manifest.json").read_bytes()
            ).hexdigest(),
            "roots": ["drafts", "raw"],
            "file_count": 文件清单["file_count"],
            "total_bytes": 文件清单["total_bytes"],
        },
    }
    正文 = json.dumps(清单, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
    清单["backup_id"] = "bkp-" + hashlib.sha256(正文).hexdigest()[:40]
    (目录 / "manifest.json").write_text(json.dumps(清单), encoding="utf-8")
    (目录 / "manifest.json").chmod(0o600)
    return 目录, 清单["backup_id"]


@pytest.mark.case_id(
    "NC-w29-29a010",
    environment="离线",
    given="显式隔离配置、合成数据与已知发布构建",
    when="经真实维护、业务或浏览器入口执行并读取持久结果",
    then=["备份解包拒绝越界链接重复与额外路径且不碰既有文件"],
    contract="docs/系统架构/新版设计/迁移与验收.md#备份与恢复入口",
)
@pytest.mark.parametrize("kind", ["escape", "symlink", "duplicate", "extra"])
def test_备份解包拒绝越界链接重复与额外路径且不碰既有文件__29a010(tmp_path, kind):
    outside = tmp_path / "outside"
    outside.write_bytes(b"keep")
    source, backup_id = _合成完整备份(tmp_path)
    valid = tmp_path / "valid.tar"
    with tarfile.open(valid, "w") as archive:
        for entry in sorted(source.iterdir()):
            archive.add(entry, arcname=entry.name)
    # 同一合法底包先走公开入口成功，再只添加一种攻击条目。
    assert 解包备份(valid, tmp_path / "valid-output", backup_id)["backup_id"] == backup_id
    packet = tmp_path / "bad.tar"
    packet.write_bytes(valid.read_bytes())
    with tarfile.open(packet, "a") as archive:
        name = {
            "escape": "../outside",
            "extra": "unlisted.txt",
            "symlink": "files/link",
            "duplicate": "database.dump",
        }[kind]
        entry = tarfile.TarInfo(name)
        if kind == "symlink":
            entry.type = tarfile.SYMTYPE
            entry.linkname = str(outside)
            archive.addfile(entry)
        else:
            entry.size = 3
            archive.addfile(entry, io.BytesIO(b"bad"))
    before = {p.relative_to(tmp_path) for p in tmp_path.rglob("*")}
    target = tmp_path / "new-output"
    with pytest.raises(备份恢复错误, match="^备份包含越界、重复或非普通文件条目$"):
        解包备份(packet, target, backup_id)
    assert not target.exists() and outside.read_bytes() == b"keep"
    assert {p.relative_to(tmp_path) for p in tmp_path.rglob("*")} == before