127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import platform
|
|
import shutil
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
ENTRY_FIXTURES = PROJECT_ROOT / "fixtures" / "entries"
|
|
VAULT_FIXTURES = PROJECT_ROOT / "fixtures" / "vaults"
|
|
MANIFEST_PATH = VAULT_FIXTURES / "manifest.json"
|
|
FIXTURE_PASSWORD = "fixture-pass-123"
|
|
WRONG_PASSWORD = "fixture-pass-wrong"
|
|
|
|
|
|
def _sha256_bytes(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def _sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
while True:
|
|
chunk = handle.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _load_encrypt_data():
|
|
import sys
|
|
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
from journal.core.encryption import encrypt_data
|
|
|
|
return encrypt_data
|
|
|
|
|
|
def _group_entries_by_month(entries: list[Path]) -> dict[str, list[Path]]:
|
|
grouped: dict[str, list[Path]] = {}
|
|
for entry in entries:
|
|
month_key = entry.stem[:7]
|
|
grouped.setdefault(month_key, []).append(entry)
|
|
return grouped
|
|
|
|
|
|
def _build_zip_payload(month_entries: list[Path]) -> bytes:
|
|
staging_root = VAULT_FIXTURES / f".staging-{uuid4().hex}"
|
|
staging_root.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
for source in sorted(month_entries, key=lambda p: p.name):
|
|
shutil.copy2(source, staging_root / source.name)
|
|
zip_path = staging_root / "payload.zip"
|
|
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as archive:
|
|
for staged in sorted(staging_root.glob("*.md"), key=lambda p: p.name):
|
|
archive.write(staged, arcname=staged.name)
|
|
return zip_path.read_bytes()
|
|
finally:
|
|
shutil.rmtree(staging_root, ignore_errors=True)
|
|
|
|
|
|
def generate() -> None:
|
|
encrypt_data = _load_encrypt_data()
|
|
VAULT_FIXTURES.mkdir(parents=True, exist_ok=True)
|
|
|
|
for old_vault in VAULT_FIXTURES.glob("*.vault"):
|
|
old_vault.unlink()
|
|
|
|
entries = sorted(ENTRY_FIXTURES.glob("*.md"), key=lambda p: p.name)
|
|
if not entries:
|
|
raise RuntimeError("No entry fixtures found under fixtures/entries.")
|
|
|
|
grouped = _group_entries_by_month(entries)
|
|
|
|
manifest: dict[str, object] = {
|
|
"format_version": 1,
|
|
"generated_at_utc": datetime.now(timezone.utc).isoformat(),
|
|
"generator": "fixtures/vaults/generate_vault_fixtures.py",
|
|
"python_version": platform.python_version(),
|
|
"password": FIXTURE_PASSWORD,
|
|
"wrong_password": WRONG_PASSWORD,
|
|
"vaults": [],
|
|
}
|
|
|
|
vault_rows: list[dict[str, object]] = []
|
|
for month_key in sorted(grouped.keys()):
|
|
month_entries = grouped[month_key]
|
|
zip_payload = _build_zip_payload(month_entries)
|
|
encrypted = encrypt_data(zip_payload, FIXTURE_PASSWORD)
|
|
|
|
vault_name = f"{month_key}.vault"
|
|
vault_path = VAULT_FIXTURES / vault_name
|
|
vault_path.write_bytes(encrypted)
|
|
|
|
expected_entries = []
|
|
for entry in sorted(month_entries, key=lambda p: p.name):
|
|
expected_entries.append(
|
|
{
|
|
"file_name": entry.name,
|
|
"sha256": _sha256_file(entry),
|
|
}
|
|
)
|
|
|
|
vault_rows.append(
|
|
{
|
|
"vault_file": vault_name,
|
|
"sha256": _sha256_file(vault_path),
|
|
"entry_count": len(expected_entries),
|
|
"expected_entries": expected_entries,
|
|
}
|
|
)
|
|
|
|
manifest["vaults"] = vault_rows
|
|
MANIFEST_PATH.write_text(json.dumps(manifest, indent=2) + os.linesep, encoding="utf-8")
|
|
print(f"Generated {len(vault_rows)} vault fixture(s) at {VAULT_FIXTURES}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
generate()
|