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()
|
