ATOCore/tests/test_backup.py

"""Tests for runtime backup creation."""

import json
import sqlite3
from datetime import UTC, datetime

import atocore.config as config
from atocore.models.database import init_db
from atocore.ops.backup import (
    create_runtime_backup,
    list_runtime_backups,
    validate_backup,
)


def test_create_runtime_backup_copies_db_and_registry(tmp_path, monkeypatch):
    monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
    monkeypatch.setenv("ATOCORE_BACKUP_DIR", str(tmp_path / "backups"))
    monkeypatch.setenv(
        "ATOCORE_PROJECT_REGISTRY_PATH", str(tmp_path / "config" / "project-registry.json")
    )

    registry_path = tmp_path / "config" / "project-registry.json"
    registry_path.parent.mkdir(parents=True)
    registry_path.write_text('{"projects":[{"id":"p01-example","aliases":[],"ingest_roots":[{"source":"vault","subpath":"incoming/projects/p01-example"}]}]}\n', encoding="utf-8")

    original_settings = config.settings
    try:
        config.settings = config.Settings()
        init_db()
        with sqlite3.connect(str(config.settings.db_path)) as conn:
            conn.execute("INSERT INTO projects (id, name) VALUES (?, ?)", ("p01", "P01 Example"))
            conn.commit()

        result = create_runtime_backup(datetime(2026, 4, 6, 18, 0, 0, tzinfo=UTC))
    finally:
        config.settings = original_settings

    db_snapshot = tmp_path / "backups" / "snapshots" / "20260406T180000Z" / "db" / "atocore.db"
    registry_snapshot = (
        tmp_path / "backups" / "snapshots" / "20260406T180000Z" / "config" / "project-registry.json"
    )
    metadata_path = (
        tmp_path / "backups" / "snapshots" / "20260406T180000Z" / "backup-metadata.json"
    )

    assert result["db_snapshot_path"] == str(db_snapshot)
    assert db_snapshot.exists()
    assert registry_snapshot.exists()
    assert metadata_path.exists()

    with sqlite3.connect(str(db_snapshot)) as conn:
        row = conn.execute("SELECT name FROM projects WHERE id = ?", ("p01",)).fetchone()
    assert row[0] == "P01 Example"

    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    assert metadata["registry_snapshot_path"] == str(registry_snapshot)


def test_create_runtime_backup_includes_chroma_when_requested(tmp_path, monkeypatch):
    monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
    monkeypatch.setenv("ATOCORE_BACKUP_DIR", str(tmp_path / "backups"))
    monkeypatch.setenv(
        "ATOCORE_PROJECT_REGISTRY_PATH", str(tmp_path / "config" / "project-registry.json")
    )

    original_settings = config.settings
    try:
        config.settings = config.Settings()
        init_db()

        # Create a fake chroma directory tree with a couple of files.
        chroma_dir = config.settings.chroma_path
        (chroma_dir / "collection-a").mkdir(parents=True, exist_ok=True)
        (chroma_dir / "collection-a" / "data.bin").write_bytes(b"\x00\x01\x02\x03")
        (chroma_dir / "metadata.json").write_text('{"ok":true}', encoding="utf-8")

        result = create_runtime_backup(
            datetime(2026, 4, 6, 20, 0, 0, tzinfo=UTC),
            include_chroma=True,
        )
    finally:
        config.settings = original_settings

    chroma_snapshot_root = (
        tmp_path / "backups" / "snapshots" / "20260406T200000Z" / "chroma"
    )
    assert result["chroma_snapshot_included"] is True
    assert result["chroma_snapshot_path"] == str(chroma_snapshot_root)
    assert result["chroma_snapshot_files"] >= 2
    assert result["chroma_snapshot_bytes"] > 0
    assert (chroma_snapshot_root / "collection-a" / "data.bin").exists()
    assert (chroma_snapshot_root / "metadata.json").exists()


def test_list_and_validate_runtime_backups(tmp_path, monkeypatch):
    monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
    monkeypatch.setenv("ATOCORE_BACKUP_DIR", str(tmp_path / "backups"))
    monkeypatch.setenv(
        "ATOCORE_PROJECT_REGISTRY_PATH", str(tmp_path / "config" / "project-registry.json")
    )

    original_settings = config.settings
    try:
        config.settings = config.Settings()
        init_db()
        first = create_runtime_backup(datetime(2026, 4, 6, 21, 0, 0, tzinfo=UTC))
        second = create_runtime_backup(datetime(2026, 4, 6, 22, 0, 0, tzinfo=UTC))

        listing = list_runtime_backups()
        first_validation = validate_backup("20260406T210000Z")
        second_validation = validate_backup("20260406T220000Z")
        missing_validation = validate_backup("20260101T000000Z")
    finally:
        config.settings = original_settings

    assert len(listing) == 2
    assert {entry["stamp"] for entry in listing} == {
        "20260406T210000Z",
        "20260406T220000Z",
    }
    for entry in listing:
        assert entry["has_metadata"] is True
        assert entry["metadata"]["db_snapshot_path"]

    assert first_validation["valid"] is True
    assert first_validation["db_ok"] is True
    assert first_validation["errors"] == []

    assert second_validation["valid"] is True

    assert missing_validation["exists"] is False
    assert "snapshot_directory_missing" in missing_validation["errors"]

    # both metadata paths are reachable on disk
    assert json.loads(
        (tmp_path / "backups" / "snapshots" / "20260406T210000Z" / "backup-metadata.json")
        .read_text(encoding="utf-8")
    )["db_snapshot_path"] == first["db_snapshot_path"]
    assert second["db_snapshot_path"].endswith("atocore.db")


def test_create_runtime_backup_handles_missing_registry(tmp_path, monkeypatch):
    monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
    monkeypatch.setenv("ATOCORE_BACKUP_DIR", str(tmp_path / "backups"))
    monkeypatch.setenv(
        "ATOCORE_PROJECT_REGISTRY_PATH", str(tmp_path / "config" / "project-registry.json")
    )

    original_settings = config.settings
    try:
        config.settings = config.Settings()
        init_db()
        result = create_runtime_backup(datetime(2026, 4, 6, 19, 0, 0, tzinfo=UTC))
    finally:
        config.settings = original_settings

    assert result["registry_snapshot_path"] == ""