from __future__ import annotations

from collections import OrderedDict
from pathlib import Path
import hashlib
import os
import tempfile
import unittest
from unittest.mock import patch

from hibor_fast_collection.archive import create_exclusive_bytes, publish_no_replace, validate_pdf
from hibor_fast_collection.manifests import (
    MANIFEST_COLUMNS, write_bytes_create_new, write_manifest_create_new,
)
from hibor_fast_collection.models import ContractError


PDF = b"%PDF-1.4\n% synthetic metadata-only fixture\n%%EOF\n"


class ArchiveManifestTests(unittest.TestCase):
    def test_create_publish_and_duplicate(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            staging = root / "stage.part"
            final = root / "out" / "report.pdf"
            create_exclusive_bytes(staging, PDF)
            digest = hashlib.sha256(PDF).hexdigest()
            result = publish_no_replace(staging, final, expected_sha256=digest)
            self.assertFalse(result.duplicate)
            self.assertEqual(result.evidence.sha256, digest)
            staging2 = root / "stage2.part"
            create_exclusive_bytes(staging2, PDF)
            duplicate = publish_no_replace(staging2, final, expected_sha256=digest)
            self.assertTrue(duplicate.duplicate)

    def test_pdf_magic_rejected(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / "bad.pdf"
            create_exclusive_bytes(path, b"not pdf")
            with self.assertRaises(ContractError): validate_pdf(path)

    def test_publish_uses_final_directory_staging_before_no_replace_link(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            staging = root / "source-volume" / "stage.part"
            staging.parent.mkdir()
            create_exclusive_bytes(staging, PDF)
            final = root / "final-volume" / "report.pdf"
            digest = hashlib.sha256(PDF).hexdigest()
            with patch("hibor_fast_collection.archive.os.link", wraps=os.link) as linked:
                publish_no_replace(staging, final, expected_sha256=digest)
            source_arg, final_arg = map(Path, linked.call_args.args)
            self.assertEqual(source_arg.parent, final_arg.parent)
            self.assertNotEqual(source_arg, staging)

    def test_recovered_publish_staging_is_not_unlinked_by_nonowner(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            staging = root / "source" / "stage.part"
            staging.parent.mkdir()
            create_exclusive_bytes(staging, PDF)
            final = root / "final" / "report.pdf"
            final.parent.mkdir()
            digest = hashlib.sha256(PDF).hexdigest()
            recovered = final.parent / f".{final.name}.{digest[:16]}.hibor-staging"
            create_exclusive_bytes(recovered, PDF)
            result = publish_no_replace(staging, final, expected_sha256=digest)
            self.assertFalse(result.duplicate)
            self.assertTrue(recovered.is_file(), "nonowner must preserve recovery staging")
            self.assertEqual((recovered.read_bytes(), final.read_bytes()), (PDF, PDF))

    def test_delivery_bytes_exact_replay_and_drift(self):
        with tempfile.TemporaryDirectory() as tmp:
            target = Path(tmp) / "delivery.md"
            first = write_bytes_create_new(target, b"delivery-v1\n")
            second = write_bytes_create_new(target, b"delivery-v1\n")
            self.assertEqual(first, second)
            with self.assertRaises(ContractError):
                write_bytes_create_new(target, b"delivery-v2\n")

    def test_manifest_exact_48_columns(self):
        with tempfile.TemporaryDirectory() as tmp:
            row = OrderedDict((k, "") for k in MANIFEST_COLUMNS)
            row.update({
                "task_id": "T", "requested_by": "R", "review_owner": "R", "source_url": "UNKNOWN",
                "source_site": "慧博 APP（安卓模拟器本地缓存）", "title": "Title", "publisher": "Pub",
                "report_date": "2026-07-29", "downloaded_at": "2026-07-29T00:00:00Z",
                "http_status": "NOT_APPLICABLE_APP_CACHE", "content_type": "application/pdf",
                "file_name": "x.pdf", "relative_path": "x.pdf", "bytes": "10", "sha256": "a"*64,
                "download_status": "SUCCESS", "error_or_note": "NONE", "source_cache_path": "/cache/x",
                "source_file_name": "x", "android_package": "cn.com.hibor", "extension_added": "false",
                "pdf_magic_valid": "true", "remote_sha256": "a"*64, "local_sha256": "a"*64,
                "openability": "true", "page_count": "1", "encryption_status": "false",
                "schema_version": "V003", "row_id": "ROW", "handoff_id": "H", "run_id": "RUN",
                "item_id": "I", "slot_id": "S", "query": "Q", "candidate_id": "C",
                "report_identity": "RID", "analysts": "", "selection_reason": "TEST",
                "remote_bytes": "10", "local_bytes": "10", "status": "SUCCESS", "stop_code": "",
                "reused_without_new_trigger": "false", "external_evidence_hash": "a"*64,
                "manifested_at_utc": "2026-07-29T00:00:00Z",
            })
            size, digest = write_manifest_create_new(Path(tmp) / "manifest.csv", [row])
            self.assertGreater(size, 0)
            self.assertEqual(len(digest), 64)
            replay = write_manifest_create_new(Path(tmp) / "manifest.csv", [row])
            self.assertEqual(replay, (size, digest))
            changed = row.copy()
            changed["title"] = "Changed"
            with self.assertRaises(ContractError):
                write_manifest_create_new(Path(tmp) / "manifest.csv", [changed])


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