archive-clients/tests/test_eviction.py

import tempfile
import unittest
from pathlib import Path

from archive_clients.eviction import (
    remove_qb_entry,
    safe_unlink,
    verify_and_snapshot,
)
from archive_clients.resources import NormalizedResource
from archive_clients.state import ClientStore
from archive_control.v1 import resource_pb2


def normalized(torrent_hash, paths):
    summary = resource_pb2.ResourceSummary(
        qb_torrent_id=torrent_hash,
        display_name="resource",
        total_file_count=len(paths),
        canonical_paths=True,
    )
    summary.resource_id.info_hash_v1_hex = torrent_hash
    files = []
    for index, path in enumerate(paths):
        size = len(path) + 10
        files.append(resource_pb2.TorrentFile(
            file_index=index,
            canonical_path=path,
            logical_bytes=size,
            completed_bytes=size,
            selected=True,
        ))
        summary.selected_files.ranges.add(first=index, last=index)
        summary.selected_complete_files.ranges.add(first=index, last=index)
    return NormalizedResource(summary, tuple(files), None)


class FakeQB:
    def __init__(self, evicted, remaining):
        self.evicted = evicted
        self.remaining = remaining
        self.deleted = []

    def get_resource(self, torrent_hash):
        return (
            self.evicted
            if self.evicted
            and self.evicted.summary.qb_torrent_id == torrent_hash
            else None
        )

    def delete_entry(self, torrent_hash):
        self.deleted.append(torrent_hash)
        self.evicted = None

    def list_resources(self):
        return list(self.remaining)


class EvictionTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name) / "qb"
        self.root.mkdir()
        self.store = ClientStore(Path(self.temp.name) / "state.db")
        self.store.initialize()
        self.store.save_job(
            "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING", 1, 0, False
        )

    def tearDown(self):
        self.temp.cleanup()

    def test_shared_paths_and_unknown_files_survive(self):
        evicted = normalized("a" * 40, ["tree/shared.bin", "tree/owned.bin"])
        other = normalized("b" * 40, ["tree/shared.bin"])
        for item in evicted.files:
            path = self.root / item.canonical_path
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(b"x" * item.logical_bytes)
        unknown = self.root / "tree" / "from-another-app.txt"
        unknown.write_text("keep")
        qb = FakeQB(evicted, [other])

        verify_and_snapshot(
            job_id="job-1",
            resource=evicted,
            selected_indices=[0, 1],
            qb_root=self.root,
            store=self.store,
        )
        remove_qb_entry(
            job_id="job-1",
            torrent_hash="a" * 40,
            qbittorrent=qb,
            store=self.store,
        )
        result = safe_unlink(
            job_id="job-1",
            qb_root=self.root,
            qbittorrent=qb,
            store=self.store,
        )
        self.assertTrue((self.root / "tree/shared.bin").exists())
        self.assertFalse((self.root / "tree/owned.bin").exists())
        self.assertTrue(unknown.exists())
        self.assertTrue((self.root / "tree").is_dir())
        self.assertEqual(qb.deleted, ["a" * 40])
        self.assertEqual(
            result["retained_files"],
            [{"path": "tree/shared.bin", "reason": "shared"}],
        )

    def test_changed_inode_is_not_unlinked(self):
        evicted = normalized("a" * 40, ["tree/owned.bin"])
        path = self.root / "tree/owned.bin"
        path.parent.mkdir(parents=True)
        path.write_bytes(b"x" * evicted.files[0].logical_bytes)
        qb = FakeQB(evicted, [])
        verify_and_snapshot(
            job_id="job-1",
            resource=evicted,
            selected_indices=[0],
            qb_root=self.root,
            store=self.store,
        )
        path.unlink()
        path.write_bytes(b"y" * evicted.files[0].logical_bytes)
        remove_qb_entry(
            job_id="job-1",
            torrent_hash="a" * 40,
            qbittorrent=qb,
            store=self.store,
        )
        result = safe_unlink(
            job_id="job-1",
            qb_root=self.root,
            qbittorrent=qb,
            store=self.store,
        )
        self.assertTrue(path.exists())
        self.assertEqual(
            result["retained_files"][0]["reason"], "identity-changed"
        )

    def test_retry_is_idempotent(self):
        evicted = normalized("a" * 40, ["owned.bin"])
        path = self.root / "owned.bin"
        path.write_bytes(b"x" * evicted.files[0].logical_bytes)
        qb = FakeQB(evicted, [])
        verify_and_snapshot(
            job_id="job-1",
            resource=evicted,
            selected_indices=[0],
            qb_root=self.root,
            store=self.store,
        )
        remove_qb_entry(
            job_id="job-1",
            torrent_hash="a" * 40,
            qbittorrent=qb,
            store=self.store,
        )
        first = safe_unlink(
            job_id="job-1",
            qb_root=self.root,
            qbittorrent=qb,
            store=self.store,
        )
        second = safe_unlink(
            job_id="job-1",
            qb_root=self.root,
            qbittorrent=qb,
            store=self.store,
        )
        self.assertEqual(first, second)
        self.assertEqual(qb.deleted, ["a" * 40])


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