# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Tests for the management command vacuum_storage."""

import datetime as dt
import os
import tempfile
from collections.abc import Generator
from contextlib import contextmanager
from io import BytesIO
from pathlib import Path
from typing import Any, TypedDict, override
from unittest import mock

from botocore import stub
from django.conf import settings
from django.db.models import F
from django.utils import timezone

from debusine.artifacts.models import CollectionCategory
from debusine.assets.models import (
    AWSProviderAccountConfiguration,
    AWSProviderAccountCredentials,
    AWSProviderAccountData,
)
from debusine.db.models import (
    Artifact,
    ArtifactRelation,
    CollectionItem,
    File,
    FileInArtifact,
    FileStore,
    FileUpload,
)
from debusine.db.models.files import FileStoreJournal
from debusine.db.tests.utils import RunInParallelTransaction
from debusine.django.management.tests import call_command
from debusine.server.file_backend.memory import MemoryFileBackend
from debusine.server.file_backend.s3 import S3FileBackend
from debusine.server.management.commands.vacuum_storage import Command
from debusine.test.django import (
    BaseDjangoTestCase,
    TestCase,
    TransactionTestCase,
)
from debusine.test.playground import Playground


class _FilesAndDirectories(TypedDict):
    """Dictionary containing files and directories for testing."""

    root: Path
    files: set[Path]
    directories: set[Path]


class VacuumStorageTestMixin(BaseDjangoTestCase):
    """Shared setup and teardown for vacuum_storage tests."""

    playground: Playground
    playground_memory_file_store = False
    two_days_ago = timezone.now() - dt.timedelta(days=2)

    @override
    def setUp(self) -> None:
        """Set common objects and settings."""
        # Using override_settings does not work (perhaps because it's a
        # lambda calculated setting)
        super().setUp()
        self.debusine_store_directory = self.create_temporary_directory()
        self.original_debusine_store_directory = (
            settings.DEBUSINE_STORE_DIRECTORY
        )
        settings.DEBUSINE_STORE_DIRECTORY = str(self.debusine_store_directory)

        # Use the default FileStore using a new base_directory
        self.file_store = FileStore.default()
        self.file_store.configuration["base_directory"] = (
            settings.DEBUSINE_STORE_DIRECTORY
        )
        self.file_store.save()

        # Use a new directory for the uploads
        self.debusine_upload_directory = str(self.create_temporary_directory())
        self.original_upload_directory = settings.DEBUSINE_UPLOAD_DIRECTORY
        settings.DEBUSINE_UPLOAD_DIRECTORY = self.debusine_upload_directory

    @override
    def tearDown(self) -> None:
        """Restore settings."""
        settings.DEBUSINE_STORE_DIRECTORY = (
            self.original_debusine_store_directory
        )
        settings.DEBUSINE_UPLOAD_DIRECTORY = self.original_upload_directory
        super().tearDown()

    def create_s3_file_store(self, **kwargs: Any) -> FileStore:
        """Create an S3 `FileStore`."""
        provider_account = self.playground.create_cloud_provider_account_asset(
            data=AWSProviderAccountData(
                name="test",
                configuration=AWSProviderAccountConfiguration(
                    region_name="test-region"
                ),
                credentials=AWSProviderAccountCredentials(
                    access_key_id="access-key",
                    secret_access_key="secret-key",
                ),
            )
        )
        return FileStore.objects.create(
            name="s3",
            backend=FileStore.BackendChoices.S3,
            configuration={"bucket_name": "test-bucket"},
            provider_account=provider_account,
            **kwargs,
        )

    def create_memory_file_store(self) -> FileStore:
        """Create a temporary in-memory file store."""
        store = FileStore.objects.create(
            name="memory",
            backend=FileStore.BackendChoices.MEMORY,
            configuration={"name": "memory"},
        )
        backend = store.get_backend_object()
        assert isinstance(backend, MemoryFileBackend)
        self.addCleanup(backend.storages.pop, store.name)
        return store

    @contextmanager
    def stub_s3_file_backend(self, store: FileStore) -> Generator[stub.Stubber]:
        """Make a stub for an S3 file store."""
        backend = store.get_backend_object()
        assert isinstance(backend, S3FileBackend)
        stubber = stub.Stubber(backend.client)
        with (
            mock.patch.object(
                S3FileBackend, "_make_client", return_value=backend.client
            ),
            stubber,
        ):
            yield stubber

    @staticmethod
    def get_local_file_path(artifact: Artifact, path_in_artifact: str) -> Path:
        """Return the local file path for the path_in_artifact in Artifact."""
        fileobj = artifact.fileinartifact_set.get(path=path_in_artifact).file
        file_backend = FileStore.default().get_backend_object()
        path = file_backend.get_local_path(fileobj)
        assert path is not None
        return path

    @classmethod
    def set_modified_time_two_days_ago(cls, path: Path) -> None:
        two_days_ago = cls.two_days_ago.timestamp()
        os.utime(path, (two_days_ago, two_days_ago))


class VacuumStorageCommandTransactionTests(
    VacuumStorageTestMixin, TransactionTestCase
):
    """Tests for vacuum_storage commands that require transactions."""

    def create_incomplete_old_artifact(self) -> tuple[Artifact, FileUpload]:
        """Return Artifact: incomplete and old, and its FileUpload."""
        artifact, _ = self.playground.create_artifact(
            paths=["README-partial-upload", "README-no-upload"],
            skip_add_files_in_store=True,
            create_files=True,
        )
        artifact.created_at = self.two_days_ago
        artifact.save()

        # The file is not completed (skip_add_files_in_store=True
        # in create_artifact)
        file_in_artifact = artifact.fileinartifact_set.get(
            path="README-partial-upload"
        )
        assert file_in_artifact is not None
        self.assertFalse(file_in_artifact.complete)

        # Create a partial upload.
        # The last_activity_at field is the default "now",
        # but it will be deleted because the Artifact was
        # creating and still incomplete more than one day ago
        with tempfile.NamedTemporaryFile(
            prefix="missing-file-",
            dir=settings.DEBUSINE_UPLOAD_DIRECTORY,
            # Can be deleted by FileUpload.delete via its on_commit hook.
            delete=False,
        ) as temp_file:
            pass
        file_upload = FileUpload.objects.create(
            file_in_artifact=file_in_artifact, path=Path(temp_file.name).name
        )

        return artifact, file_upload

    def test_delete_old_incomplete_artifacts(self) -> None:
        """Delete old and incomplete artifacts."""
        artifact_1, file_upload_1 = self.create_incomplete_old_artifact()
        file_upload_1_path = file_upload_1.absolute_file_path()

        # artifact_2 will not be deleted: incomplete but recent
        artifact_2, _ = self.playground.create_artifact(
            paths=["README"], create_files=True
        )
        file_in_artifact_2 = artifact_2.fileinartifact_set.first()
        assert file_in_artifact_2 is not None
        file_in_artifact_2.complete = False
        file_in_artifact_2.save()
        self.playground.create_artifact_relation(artifact_1, artifact_2)

        # artifact_3 will not be deleted: old but complete
        artifact_3, _ = self.playground.create_artifact(paths=["README"])
        artifact_3.created_at = self.two_days_ago
        artifact_3.save()
        self.playground.create_artifact_relation(artifact_2, artifact_3)

        # artifact_4 will be deleted: target of an artifact relation (this
        # cannot happen normally any more:
        # ArtifactRelationsView.perform_create forbids it)
        artifact_4, file_upload_4 = self.create_incomplete_old_artifact()
        file_upload_4_path = file_upload_4.absolute_file_path()
        self.playground.create_artifact_relation(artifact_3, artifact_4)

        # artifact_5 will not be deleted: in a collection
        artifact_5, _ = self.create_incomplete_old_artifact()
        collection = self.playground.create_collection(
            "test", CollectionCategory.TEST
        )
        CollectionItem.objects.create_from_artifact(
            artifact_5,
            parent_collection=collection,
            name="test",
            data={},
            created_by_user=self.playground.get_default_user(),
        )

        stdout, stderr, exitcode = call_command("vacuum_storage", verbosity=2)

        # artifact_1 and its outbound relation are deleted, as are
        # artifact_4 and its inbound relation
        self.assertQuerySetEqual(
            Artifact.objects.all(),
            {artifact_2, artifact_3, artifact_5},
            ordered=False,
        )
        self.assertQuerySetEqual(
            ArtifactRelation.objects.values_list("artifact", "target"),
            [(artifact_2.id, artifact_3.id)],
            ordered=False,
        )

        # There is no file (deleted as part of FileUpload.delete())
        self.assertFalse(file_upload_1_path.exists())
        self.assertFalse(file_upload_4_path.exists())

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            "Checking incomplete artifacts\n"
            f"Deleted incomplete artifact {artifact_1.id}\n"
            f"Deleted incomplete artifact {artifact_4.id}\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_delete_old_incomplete_artifacts_dryrun(self) -> None:
        """Do not delete old and incomplete artifact: dry-run."""
        artifact, file_upload = self.create_incomplete_old_artifact()
        file_upload_path = file_upload.absolute_file_path()

        stdout, stderr, exitcode = call_command("vacuum_storage", "--dry-run")

        # artifact is not deleted
        self.assertQuerySetEqual(Artifact.objects.all(), {artifact})

        # file_upload and file in disk are not deleted
        self.assertQuerySetEqual(FileUpload.objects.all(), {file_upload})
        self.assertTrue(file_upload_path.exists())

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_delete_redundant_uploads(self) -> None:
        artifact, _ = self.playground.create_artifact(
            paths=[
                "old-leftover-upload",
                "recent-leftover-upload",
                "partial-upload",
            ],
            skip_add_files_in_store=True,
            create_files=True,
        )
        uploads: dict[str, FileUpload] = {}
        now = timezone.now()
        for path, last_activity_at in (
            ("old-leftover-upload", now - dt.timedelta(hours=2)),
            ("recent-leftover-upload", now - dt.timedelta(minutes=1)),
            ("partial-upload", now),
        ):
            with tempfile.NamedTemporaryFile(
                prefix="file-uploading-",
                dir=settings.DEBUSINE_UPLOAD_DIRECTORY,
                # Can be deleted by FileUpload.delete via its on_commit hook.
                delete=False,
            ) as temp_file:
                pass
            file_in_artifact = artifact.fileinartifact_set.get(path=path)
            uploads[path] = FileUpload.objects.create(
                file_in_artifact=file_in_artifact,
                path=Path(temp_file.name).name,
            )
            uploads[path].last_activity_at = last_activity_at
            uploads[path].save()
            file_in_artifact.complete = True
            file_in_artifact.save()
        for path in ("old-leftover-upload", "recent-leftover-upload"):
            uploads[path].absolute_file_path().unlink()

        stdout, stderr, exitcode = call_command("vacuum_storage", verbosity=2)

        self.assertQuerySetEqual(
            FileUpload.objects.all(),
            {uploads["recent-leftover-upload"], uploads["partial-upload"]},
            ordered=False,
        )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            f"Deleted incomplete upload {uploads['old-leftover-upload']}\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_delete_redundant_uploads_dry_run(self) -> None:
        artifact, _ = self.playground.create_artifact(
            paths=["old-leftover-upload"],
            skip_add_files_in_store=True,
            create_files=True,
        )
        with tempfile.NamedTemporaryFile(
            prefix="file-uploading-",
            dir=settings.DEBUSINE_UPLOAD_DIRECTORY,
            # Can be deleted by FileUpload.delete via its on_commit hook.
            delete=False,
        ) as temp_file:
            pass
        file_in_artifact = artifact.fileinartifact_set.get(
            path="old-leftover-upload"
        )
        upload = FileUpload.objects.create(
            file_in_artifact=file_in_artifact, path=Path(temp_file.name).name
        )
        upload.last_activity_at = timezone.now() - dt.timedelta(hours=2)
        upload.save()
        file_in_artifact.complete = True
        file_in_artifact.save()
        upload.absolute_file_path().unlink()

        stdout, stderr, exitcode = call_command(
            "vacuum_storage", "--dry-run", verbosity=2
        )

        # The file upload is not deleted.
        self.assertQuerySetEqual(FileUpload.objects.all(), {upload})

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            f"Deleted incomplete upload {upload}\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_roll_up_file_store_journal(self) -> None:
        memory_file_store = self.create_memory_file_store()
        files = [
            self.playground.create_file(contents=b"\0" * length)
            for length in (0, 4, 10, 1024, 512)
        ]
        for file in files[:3]:
            self.file_store.files.add(file)
        for file in files[2:]:
            memory_file_store.files.add(file)

        self.assertQuerySetEqual(
            FileStoreJournal.objects.order_by("id").values_list(
                "store__name", "size_change"
            ),
            [
                (self.file_store.name, 0),
                (self.file_store.name, 4),
                (self.file_store.name, 10),
                (memory_file_store.name, 10),
                (memory_file_store.name, 1024),
                (memory_file_store.name, 512),
            ],
        )
        self.assertQuerySetEqual(
            FileStore.objects.annotate_current_total_size().values_list(
                "name", "total_size", "current_total_size"
            ),
            [(self.file_store.name, 0, 14), (memory_file_store.name, 0, 1546)],
            ordered=False,
        )

        def add_file_in_thread(*args: Any, **kwargs: Any) -> Any:
            # Queries in REPEATABLE READ transactions see a snapshot as of
            # the start of the first non-transaction-control statement in
            # the transaction, so issue one to make sure that the parallel
            # transaction happens after that point.
            self.file_store.refresh_from_db()

            thread = RunInParallelTransaction(
                lambda: self.file_store.files.add(files[3])
            )
            thread.run_and_wait()
            return mock.DEFAULT

        with (
            self.spy_object(
                Command,
                "_roll_up_file_store_journal",
                side_effect=add_file_in_thread,
            ),
        ):
            stdout, stderr, exitcode = call_command("vacuum_storage")

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)
        self.assertQuerySetEqual(
            FileStoreJournal.objects.order_by("id").values_list(
                "store__name", "size_change"
            ),
            [(self.file_store.name, 1024)],
        )
        self.assertQuerySetEqual(
            FileStore.objects.annotate_current_total_size().values_list(
                "name", "total_size", "current_total_size"
            ),
            [
                (self.file_store.name, 14, 1038),
                (memory_file_store.name, 1546, 1546),
            ],
            ordered=False,
        )

    def test_roll_up_file_store_journal_dry_run(self) -> None:
        file = self.playground.create_file(contents=b"\0" * 4)
        self.file_store.files.add(file)

        self.assertQuerySetEqual(
            FileStoreJournal.objects.order_by("id").values_list(
                "store__name", "size_change"
            ),
            [(self.file_store.name, 4)],
        )
        self.assertQuerySetEqual(
            FileStore.objects.annotate_current_total_size().values_list(
                "name", "total_size", "current_total_size"
            ),
            [(self.file_store.name, 0, 4)],
        )

        stdout, stderr, exitcode = call_command("vacuum_storage", "--dry-run")

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)
        self.assertQuerySetEqual(
            FileStoreJournal.objects.order_by("id").values_list(
                "store__name", "size_change"
            ),
            [(self.file_store.name, 4)],
        )
        self.assertQuerySetEqual(
            FileStore.objects.annotate_current_total_size().values_list(
                "name", "total_size", "current_total_size"
            ),
            [(self.file_store.name, 0, 4)],
        )


class VacuumStorageCommandTests(VacuumStorageTestMixin, TestCase):
    """Tests for vacuum_storage command."""

    @override
    def setUp(self) -> None:
        super().setUp()
        self.enterContext(
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes")
        )

    def get_store_contents(self, store: FileStore) -> set[bytes]:
        """
        Return a set of the contents of all the files in a store.

        Since file stores are content-addressable, we assume this is a
        sufficient representation for convenient testing.
        """
        store_contents = set()
        for entry in store.get_backend_object().list_entries():
            with entry.get_temporary_local_path() as path:
                store_contents.add(path.read_bytes())
        return store_contents

    def test_delete_empty_directory_dry_run(self) -> None:
        """Report delete an empty directory (but dry-run: does not delete)."""
        empty_directory = self.create_temporary_directory(
            directory=self.debusine_store_directory
        )
        self.set_modified_time_two_days_ago(empty_directory)
        self.playground.create_artifact(paths=["README"], create_files=True)

        stdout, stderr, exitcode = call_command(
            "vacuum_storage", "--dry-run", verbosity=2
        )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"Deleted empty directory {empty_directory}\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        # Not deleted: --dry-run was used
        self.assertTrue(empty_directory.is_dir())

    def test_delete_empty_directory(self) -> None:
        """Delete an empty directory (and leave a non-empty directory)."""
        empty_dir = self.create_temporary_directory(
            directory=self.debusine_store_directory
        )
        self.set_modified_time_two_days_ago(empty_dir)
        path_in_artifact = "README"
        artifact, _ = self.playground.create_artifact(
            paths=[path_in_artifact],
            create_files=True,
            skip_add_files_in_store=False,
        )

        stdout, stderr, exitcode = call_command("vacuum_storage")

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        self.assertFalse(empty_dir.is_dir())

        # Directory of the file in the artifact exists
        # The file part of an artifact was not deleted
        local_file_path = self.get_local_file_path(artifact, path_in_artifact)
        self.assertTrue(local_file_path.parent.is_dir())

    def test_delete_empty_doesnt_delete_store(self) -> None:
        """The store itself isn't deleted."""
        self.set_modified_time_two_days_ago(self.debusine_store_directory)

        _, _, exitcode = call_command("vacuum_storage")
        self.assertEqual(exitcode, 0)

        self.assertTrue(self.debusine_store_directory.is_dir())

    def test_delete_empty_directory_walks_two_levels(self) -> None:
        """Deleting empty directories only walks down as far as needed."""
        (
            self.debusine_store_directory / "00" / "01" / "02" / "03" / "04"
        ).mkdir(parents=True)
        (self.debusine_store_directory / "00" / "01" / "03").mkdir(parents=True)
        for dirpath, _, _ in self.debusine_store_directory.walk(top_down=False):
            self.set_modified_time_two_days_ago(dirpath)

        with self.spy("pathlib.Path.rmdir") as rmdir:
            stdout, stderr, exitcode = call_command("vacuum_storage")

        self.assertEqual(rmdir.call_count, 4)
        rmdir.assert_has_calls(
            [
                mock.call(self.debusine_store_directory / "00"),
                mock.call(self.debusine_store_directory / "00" / "01"),
                mock.call(self.debusine_store_directory / "00" / "01" / "02"),
                mock.call(self.debusine_store_directory / "00" / "01" / "03"),
            ],
            any_order=True,
        )

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_delete_empty_directory_skips_known_nonempty(self) -> None:
        """Directories with files in the DB are presumed to be non-empty."""
        artifact, _ = self.playground.create_artifact(
            paths=["README"], create_files=True
        )
        empty_dir = self.create_temporary_directory(
            directory=self.debusine_store_directory
        )
        for dirpath, _, _ in self.debusine_store_directory.walk(top_down=False):
            self.set_modified_time_two_days_ago(dirpath)

        with self.spy("pathlib.Path.rmdir") as rmdir:
            stdout, stderr, exitcode = call_command("vacuum_storage")

        rmdir.assert_called_once_with(empty_dir)

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        self.assertTrue(
            self.get_local_file_path(artifact, "README").parent.is_dir()
        )
        self.assertFalse(empty_dir.is_dir())

    def test_delete_empty_directory_too_new(self) -> None:
        """Directory is not deleted: it's too new."""
        empty_dir = self.create_temporary_directory(
            directory=self.debusine_store_directory
        )
        call_command("vacuum_storage")

        self.assertTrue(empty_dir.is_dir())

    def test_delete_directory_fails(self) -> None:
        """directory.rmdir() fails (file was probably added after the check)."""
        empty_dir = self.create_temporary_directory(
            directory=self.debusine_store_directory
        )
        self.set_modified_time_two_days_ago(empty_dir)

        with mock.patch(
            "pathlib.Path.rmdir", autospec=True, side_effect=OSError
        ) as rmdir:
            stdout, stderr, exitcode = call_command("vacuum_storage")

        rmdir.assert_called_once_with(empty_dir)

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        self.assertTrue(empty_dir.is_dir())

    def test_apply_populate_policy_dry_run(self) -> None:
        """Populate a store: dry run."""
        artifact, _ = self.playground.create_artifact(
            paths=["1", "2", "3"], create_files=True
        )
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store()
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(
            s3_file_store, through_defaults={"populate": True}
        )

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", "--dry-run", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Populating {s3_file_store.name}\n"
            + "".join(
                f"{scope}: Would copy {file.hash_digest.hex()} from "
                f"{self.file_store.name} to {s3_file_store.name}\n"
                for file in artifact.files.order_by("sha256")
            )
            + "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_populate_policy(self) -> None:
        """Populate a store."""
        artifact, _ = self.playground.create_artifact(
            paths=["1", "2", "3"], create_files=True
        )
        files = artifact.files.order_by("sha256")
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store()
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(
            s3_file_store,
            through_defaults={"populate": True, "write_only": True},
        )

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            # _apply_populate_policy
            for file in files:
                stubber.add_response(
                    "put_object",
                    {},
                    {
                        "Bucket": "test-bucket",
                        "Key": f"{file.hash_digest.hex()}-{file.size}",
                        "Body": stub.ANY,
                        "ChecksumAlgorithm": "SHA256",
                    },
                )

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Populating {s3_file_store.name}\n"
            + "".join(
                f"{scope}: Copied {file.hash_digest.hex()} from "
                f"{self.file_store.name} to {s3_file_store.name}\n"
                for file in files
            )
            + "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_populate_policy_nothing_to_do(self) -> None:
        """If all files are already in a `populate` store, do nothing."""
        artifact, _ = self.playground.create_artifact(
            paths=["1", "2", "3"], create_files=True
        )
        files = artifact.files.order_by("sha256")
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store()
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(
            s3_file_store, through_defaults={"populate": True}
        )
        s3_file_store.files.add(*files)

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_populate_policy_file_deleted_in_parallel(self) -> None:
        """Populating a store ignores files that were deleted in parallel."""
        artifact, _ = self.playground.create_artifact(
            paths=["1", "2", "3"], create_files=True
        )
        files = list(artifact.files.order_by("sha256"))
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store()
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(
            s3_file_store, through_defaults={"populate": True}
        )
        original_refresh_from_db = File.refresh_from_db

        def mock_file_refresh_from_db(file: File) -> None:
            if file == files[1]:
                artifact.files.remove(file)
                self.file_store.files.remove(file)
                file.delete()
            return original_refresh_from_db(file)

        with (
            self.stub_s3_file_backend(s3_file_store) as stubber,
            mock.patch.object(
                File,
                "refresh_from_db",
                autospec=True,
                side_effect=mock_file_refresh_from_db,
            ),
        ):
            # _apply_populate_policy
            for file in (files[0], files[2]):
                stubber.add_response(
                    "put_object",
                    {},
                    {
                        "Bucket": "test-bucket",
                        "Key": f"{file.hash_digest.hex()}-{file.size}",
                        "Body": stub.ANY,
                        "ChecksumAlgorithm": "SHA256",
                    },
                )

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Populating {s3_file_store.name}\n"
            + "".join(
                f"{scope}: Copied {file.hash_digest.hex()} from "
                f"{self.file_store.name} to {s3_file_store.name}\n"
                for file in (files[0], files[2])
            )
            + "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_populate_policy_cannot_download(self) -> None:
        """Cannot populate files that have no other download file stores."""
        artifact, _ = self.playground.create_artifact(
            paths=["1", "2", "3"], create_files=True
        )
        files = artifact.files.order_by("sha256")
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store()
        self.assertIn(self.file_store, scope.file_stores.all())
        local_policies = scope.filestoreinscope_set.get(
            file_store=self.file_store
        )
        local_policies.write_only = True
        local_policies.save()
        scope.file_stores.add(
            s3_file_store, through_defaults={"populate": True}
        )

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Populating {s3_file_store.name}\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(
            stderr,
            "".join(
                f"{scope}: Wanted to populate {s3_file_store.name} with "
                f"{file.hash_digest.hex()}, but it has no other download file "
                f"stores\n"
                for file in files
            ),
        )
        self.assertEqual(exitcode, 1)

        stubber.assert_no_pending_responses()

    def test_apply_populate_policy_size_exceeded(self) -> None:
        """We stop populating a store when its size limits are reached."""
        artifact, _ = self.playground.create_artifact(
            paths=["1", "2", "3"], files_size=100, create_files=True
        )
        files = artifact.files.order_by("sha256")
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=150)
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(
            s3_file_store, through_defaults={"populate": True}
        )

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            # _apply_populate_policy
            stubber.add_response(
                "put_object",
                {},
                {
                    "Bucket": "test-bucket",
                    "Key": f"{files[0].hash_digest.hex()}-{files[0].size}",
                    "Body": stub.ANY,
                    "ChecksumAlgorithm": "SHA256",
                },
            )

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Populating {s3_file_store.name}\n"
            f"{scope}: Copied {files[0].hash_digest.hex()} from "
            f"{self.file_store.name} to {s3_file_store.name}\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(
            stderr,
            f"{scope}: Stopping population of {s3_file_store.name} to avoid "
            f"exceeding size limits\n",
        )
        self.assertEqual(exitcode, 1)

        stubber.assert_no_pending_responses()

    def test_apply_drain_policy_dry_run(self) -> None:
        """Drain a store: dry run."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=10)
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(s3_file_store, through_defaults={"drain": True})
        drained_artifact, _ = self.playground.create_artifact(
            paths=["1"], create_files=True
        )
        undrained_paths = {
            path: f"Test file {path}".encode() for path in ("2", "3")
        }
        undrained_artifact, _ = self.playground.create_artifact(
            paths=undrained_paths,
            create_files=True,
            skip_add_files_in_store=True,
        )
        files = File.objects.filter(
            artifact__in=(drained_artifact, undrained_artifact)
        ).order_by("sha256")
        s3_file_store.files.add(*files)

        expected_stdout = [
            "Checking empty directories\n",
            f"{scope}: Draining {s3_file_store.name}\n",
        ]
        for file in File.objects.filter(
            artifact__in=(drained_artifact, undrained_artifact)
        ).order_by("sha256"):
            if file in drained_artifact.files.all():
                expected_stdout.append(
                    f"{scope}: Would remove {file.hash_digest.hex()} from "
                    f"{s3_file_store.name}\n"
                )
            else:
                expected_stdout.append(
                    f"{scope}: Would move {file.hash_digest.hex()} from "
                    f"{s3_file_store.name} to {self.file_store.name}\n"
                )
        expected_stdout += [
            "Checking incomplete artifacts\n",
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n",
            "Checking missing files from upload\n",
        ]

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", "--dry-run", verbosity=2
            )

        self.assertEqual(stdout, "".join(expected_stdout))
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_drain_policy(self) -> None:
        """Drain a store."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=150)
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(s3_file_store, through_defaults={"drain": True})
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        drained_artifact, _ = self.playground.create_artifact(
            paths={"1": paths["1"]}, create_files=True
        )
        undrained_artifact, _ = self.playground.create_artifact(
            paths={"2": paths["2"], "3": paths["3"]},
            create_files=True,
            skip_add_files_in_store=True,
        )
        files = File.objects.filter(
            artifact__in=(drained_artifact, undrained_artifact)
        ).order_by("sha256")
        s3_file_store.files.add(*files)

        expected_stdout = [
            "Checking empty directories\n",
            f"{scope}: Draining {s3_file_store.name}\n",
        ]
        for file in files:
            if file in drained_artifact.files.all():
                expected_stdout.append(
                    f"{scope}: Removed {file.hash_digest.hex()} from "
                    f"{s3_file_store.name}\n"
                )
            else:
                expected_stdout.append(
                    f"{scope}: Moved {file.hash_digest.hex()} from "
                    f"{s3_file_store.name} to {self.file_store.name}\n"
                )
        expected_stdout += [
            "Checking incomplete artifacts\n",
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n",
            "Checking missing files from upload\n",
        ]

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            # _apply_drain_policy
            for file in files:
                path = file.fileinartifact_set.get().path
                identifier = f"{file.sha256.hex()}-{file.size}"
                if file not in drained_artifact.files.all():
                    stubber.add_response(
                        "head_object",
                        {"ContentLength": len(paths[path])},
                        {"Bucket": "test-bucket", "Key": identifier},
                    )
                    stubber.add_response(
                        "get_object",
                        {"Body": BytesIO(paths[path])},
                        {"Bucket": "test-bucket", "Key": identifier},
                    )
                stubber.add_response(
                    "delete_object",
                    {},
                    {"Bucket": "test-bucket", "Key": identifier},
                )

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(stdout, "".join(expected_stdout))
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

        self.assertEqual(
            self.get_store_contents(self.file_store), set(paths.values())
        )

    def test_apply_drain_policy_nothing_to_do(self) -> None:
        """If a `drain` store is already empty, do nothing."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=150)
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(s3_file_store, through_defaults={"drain": True})

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_drain_policy_file_deleted_in_parallel(self) -> None:
        """Draining a store ignores files that were deleted in parallel."""
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        artifact, _ = self.playground.create_artifact(
            paths=paths, create_files=True, skip_add_files_in_store=True
        )
        files = artifact.files.order_by("sha256")
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store()
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(s3_file_store, through_defaults={"drain": True})
        s3_file_store.files.add(*files)
        original_refresh_from_db = File.refresh_from_db

        def mock_file_refresh_from_db(file: File) -> None:
            if file == files[1]:
                artifact.files.remove(file)
                s3_file_store.files.remove(file)
                file.delete()
            return original_refresh_from_db(file)

        with (
            self.stub_s3_file_backend(s3_file_store) as stubber,
            mock.patch.object(
                File,
                "refresh_from_db",
                autospec=True,
                side_effect=mock_file_refresh_from_db,
            ),
        ):
            # _apply_drain_policy
            for file in (files[0], files[2]):
                path = file.fileinartifact_set.get().path
                identifier = f"{file.sha256.hex()}-{file.size}"
                stubber.add_response(
                    "head_object",
                    {"ContentLength": len(paths[path])},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                stubber.add_response(
                    "get_object",
                    {"Body": BytesIO(paths[path])},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                stubber.add_response(
                    "delete_object",
                    {},
                    {"Bucket": "test-bucket", "Key": identifier},
                )

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Draining {s3_file_store.name}\n"
            + "".join(
                f"{scope}: Moved {file.hash_digest.hex()} from "
                f"{s3_file_store.name} to {self.file_store.name}\n"
                for file in (files[0], files[2])
            )
            + "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_drain_policy_drain_to(self) -> None:
        """The `drain_to` policy selects a target store."""
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        artifact, _ = self.playground.create_artifact(
            paths=paths, create_files=True, skip_add_files_in_store=True
        )
        files = artifact.files.order_by("sha256")
        scope = self.playground.get_default_scope()
        memory_file_store = self.create_memory_file_store()
        s3_file_store = self.create_s3_file_store()
        self.assertEqual(
            scope.filestoreinscope_set.get(
                file_store=self.file_store
            ).upload_priority,
            100,
        )
        scope.file_stores.add(
            memory_file_store, through_defaults={"upload_priority": 0}
        )
        scope.file_stores.add(
            s3_file_store,
            through_defaults={"drain": True, "drain_to": "memory"},
        )
        s3_file_store.files.add(*files)

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            # _apply_drain_policy
            for file in files:
                path = file.fileinartifact_set.get().path
                identifier = f"{file.sha256.hex()}-{file.size}"
                stubber.add_response(
                    "head_object",
                    {"ContentLength": len(paths[path])},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                stubber.add_response(
                    "get_object",
                    {"Body": BytesIO(paths[path])},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                stubber.add_response(
                    "delete_object",
                    {},
                    {"Bucket": "test-bucket", "Key": identifier},
                )

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Draining {s3_file_store.name}\n"
            + "".join(
                f"{scope}: Moved {file.hash_digest.hex()} from "
                f"{s3_file_store.name} to {memory_file_store.name}\n"
                for file in files
            )
            + "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

        self.assertEqual(
            self.get_store_contents(memory_file_store), set(paths.values())
        )

    def test_apply_drain_policy_cannot_upload(self) -> None:
        """Cannot drain files that have no other upload file stores."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=150)
        local_policies = scope.filestoreinscope_set.get(
            file_store=self.file_store
        )
        local_policies.read_only = True
        local_policies.save()
        scope.file_stores.add(s3_file_store, through_defaults={"drain": True})
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        artifact, _ = self.playground.create_artifact(
            paths=paths, create_files=True, skip_add_files_in_store=True
        )
        files = artifact.files.order_by("sha256")
        s3_file_store.files.add(*files)

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Draining {s3_file_store.name}\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(
            stderr,
            "".join(
                f"{scope}: Wanted to drain {s3_file_store.name} of "
                f"{file.hash_digest.hex()}, but it has no other upload file "
                f"stores\n"
                for file in files
            ),
        )
        self.assertEqual(exitcode, 1)

        stubber.assert_no_pending_responses()

        self.assertEqual(self.get_store_contents(self.file_store), set())

    def test_apply_drain_policy_cannot_upload_drain_to(self) -> None:
        """Raise an explicit error if the `drain_to` store is ineligible."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store()
        local_policies = scope.filestoreinscope_set.get(
            file_store=self.file_store
        )
        local_policies.read_only = True
        local_policies.save()
        scope.file_stores.add(
            s3_file_store,
            through_defaults={"drain": True, "drain_to": self.file_store.name},
        )
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        artifact, _ = self.playground.create_artifact(
            paths=paths, create_files=True, skip_add_files_in_store=True
        )
        files = artifact.files.order_by("sha256")
        s3_file_store.files.add(*files)

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Draining {s3_file_store.name}\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(
            stderr,
            "".join(
                f"{scope}: Wanted to drain {s3_file_store.name} of "
                f"{file.hash_digest.hex()}, but {self.file_store.name} cannot "
                f"accept it\n"
                for file in files
            ),
        )
        self.assertEqual(exitcode, 1)

        stubber.assert_no_pending_responses()

        self.assertEqual(self.get_store_contents(self.file_store), set())

    def test_apply_max_size_limits_dry_run(self) -> None:
        """Apply maximum size limits: dry run."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=20)
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(s3_file_store)
        drained_artifact, _ = self.playground.create_artifact(
            paths=["1"], files_size=10, create_files=True
        )
        undrained_paths = {
            path: f"Test file {path}".encode() for path in ("2", "3")
        }
        undrained_artifact, _ = self.playground.create_artifact(
            paths=undrained_paths,
            create_files=True,
            skip_add_files_in_store=True,
        )
        files = File.objects.filter(
            artifact__in=(drained_artifact, undrained_artifact)
        ).order_by("sha256")
        s3_file_store.files.add(*files)
        s3_total_size = (
            FileStore.objects.annotate_current_total_size()
            .values_list("current_total_size", flat=True)
            .get(pk=s3_file_store.pk)
        )

        expected_stdout = [
            "Checking empty directories\n",
            f"{scope}: Draining up to {s3_total_size - 20} bytes "
            f"from {s3_file_store.name}\n",
        ]
        for file in File.objects.filter(
            artifact__in=(drained_artifact, undrained_artifact)
        ).order_by("artifact__created_at", F("size").desc(), "sha256")[:2]:
            if file in drained_artifact.files.all():
                expected_stdout.append(
                    f"{scope}: Would remove {file.hash_digest.hex()} from "
                    f"{s3_file_store.name}\n"
                )
            else:
                expected_stdout.append(
                    f"{scope}: Would move {file.hash_digest.hex()} from "
                    f"{s3_file_store.name} to {self.file_store.name}\n"
                )
        expected_stdout += [
            "Checking incomplete artifacts\n",
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n",
            "Checking missing files from upload\n",
        ]

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", "--dry-run", verbosity=2
            )

        self.assertEqual(stdout, "".join(expected_stdout))
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_max_size_limits(self) -> None:
        """Apply maximum size limits."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(max_size=20)
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(s3_file_store)
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        drained_artifact, _ = self.playground.create_artifact(
            paths={"1": paths["1"]}, create_files=True
        )
        undrained_artifact, _ = self.playground.create_artifact(
            paths={"2": paths["2"], "3": paths["3"]},
            create_files=True,
            skip_add_files_in_store=True,
        )
        files = File.objects.filter(
            artifact__in=(drained_artifact, undrained_artifact)
        ).order_by("artifact__created_at", F("size").desc(), "sha256")
        s3_file_store.files.add(*files)
        s3_total_size = (
            FileStore.objects.annotate_current_total_size()
            .values_list("current_total_size", flat=True)
            .get(pk=s3_file_store.pk)
        )

        expected_stdout = [
            "Checking empty directories\n",
            f"{scope}: Draining up to {s3_total_size - 20} bytes "
            f"from {s3_file_store.name}\n",
        ]
        for file in files[:2]:
            if file in drained_artifact.files.all():
                expected_stdout.append(
                    f"{scope}: Removed {file.hash_digest.hex()} from "
                    f"{s3_file_store.name}\n"
                )
            else:
                expected_stdout.append(
                    f"{scope}: Moved {file.hash_digest.hex()} from "
                    f"{s3_file_store.name} to {self.file_store.name}\n"
                )
        expected_stdout += [
            "Checking incomplete artifacts\n",
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n",
            "Checking missing files from upload\n",
        ]
        expected_local_contents = set()

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            # _apply_max_size_limits
            for file in files[:2]:
                path = file.fileinartifact_set.get().path
                identifier = f"{file.sha256.hex()}-{file.size}"
                if file not in drained_artifact.files.all():
                    stubber.add_response(
                        "head_object",
                        {"ContentLength": len(paths[path])},
                        {"Bucket": "test-bucket", "Key": identifier},
                    )
                    stubber.add_response(
                        "get_object",
                        {"Body": BytesIO(paths[path])},
                        {"Bucket": "test-bucket", "Key": identifier},
                    )
                stubber.add_response(
                    "delete_object",
                    {},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                expected_local_contents.add(paths[path])

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(stdout, "".join(expected_stdout))
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

        self.assertEqual(
            self.get_store_contents(self.file_store), expected_local_contents
        )

    def test_apply_max_size_limits_nothing_to_do(self) -> None:
        """If a store with a maximum size is already empty, do nothing."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=150)
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(s3_file_store)

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_max_size_limits_file_deleted_in_parallel(self) -> None:
        """Applying `max_size` ignores files that were deleted in parallel."""
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        artifact, _ = self.playground.create_artifact(
            paths=paths, create_files=True, skip_add_files_in_store=True
        )
        files = artifact.files.order_by(F("size").desc(), "sha256")
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=5, max_size=20)
        self.assertIn(self.file_store, scope.file_stores.all())
        scope.file_stores.add(s3_file_store)
        s3_file_store.files.add(*files)
        s3_total_size = (
            FileStore.objects.annotate_current_total_size()
            .values_list("current_total_size", flat=True)
            .get(pk=s3_file_store.pk)
        )
        original_refresh_from_db = File.refresh_from_db

        def mock_file_refresh_from_db(file: File) -> None:
            if file == files[1]:
                artifact.files.remove(file)
                s3_file_store.files.remove(file)
                file.delete()
            return original_refresh_from_db(file)

        with (
            self.stub_s3_file_backend(s3_file_store) as stubber,
            mock.patch.object(
                File,
                "refresh_from_db",
                autospec=True,
                side_effect=mock_file_refresh_from_db,
            ),
        ):
            # _apply_max_size_limits
            for file in (files[0], files[2]):
                path = file.fileinartifact_set.get().path
                identifier = f"{file.sha256.hex()}-{file.size}"
                stubber.add_response(
                    "head_object",
                    {"ContentLength": len(paths[path])},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                stubber.add_response(
                    "get_object",
                    {"Body": BytesIO(paths[path])},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                stubber.add_response(
                    "delete_object",
                    {},
                    {"Bucket": "test-bucket", "Key": identifier},
                )

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Draining up to {s3_total_size - 5} bytes "
            f"from {s3_file_store.name}\n"
            + "".join(
                f"{scope}: Moved {file.hash_digest.hex()} from "
                f"{s3_file_store.name} to {self.file_store.name}\n"
                for file in (files[0], files[2])
            )
            + "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

    def test_apply_max_size_limits_drain_to(self) -> None:
        """The `drain_to` policy selects a target store."""
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        artifact, _ = self.playground.create_artifact(
            paths=paths, create_files=True, skip_add_files_in_store=True
        )
        files = artifact.files.order_by(F("size").desc(), "sha256")
        scope = self.playground.get_default_scope()
        memory_file_store = self.create_memory_file_store()
        s3_file_store = self.create_s3_file_store(max_size=5)
        self.assertEqual(
            scope.filestoreinscope_set.get(
                file_store=self.file_store
            ).upload_priority,
            100,
        )
        scope.file_stores.add(
            memory_file_store, through_defaults={"upload_priority": 0}
        )
        scope.file_stores.add(
            s3_file_store,
            through_defaults={"drain_to": "memory"},
        )
        s3_file_store.files.add(*files)
        s3_total_size = (
            FileStore.objects.annotate_current_total_size()
            .values_list("current_total_size", flat=True)
            .get(pk=s3_file_store.pk)
        )

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            # _apply_max_size_limits
            for file in files:
                path = file.fileinartifact_set.get().path
                identifier = f"{file.sha256.hex()}-{file.size}"
                stubber.add_response(
                    "head_object",
                    {"ContentLength": len(paths[path])},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                stubber.add_response(
                    "get_object",
                    {"Body": BytesIO(paths[path])},
                    {"Bucket": "test-bucket", "Key": identifier},
                )
                stubber.add_response(
                    "delete_object",
                    {},
                    {"Bucket": "test-bucket", "Key": identifier},
                )

            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Draining up to {s3_total_size - 5} bytes "
            f"from {s3_file_store.name}\n"
            + "".join(
                f"{scope}: Moved {file.hash_digest.hex()} from "
                f"{s3_file_store.name} to {memory_file_store.name}\n"
                for file in files
            )
            + "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        stubber.assert_no_pending_responses()

        self.assertEqual(
            self.get_store_contents(memory_file_store), set(paths.values())
        )

    def test_apply_max_size_limits_cannot_upload(self) -> None:
        """Cannot drain files that have no other upload file stores."""
        scope = self.playground.get_default_scope()
        s3_file_store = self.create_s3_file_store(soft_max_size=20)
        self.assertIn(self.file_store, scope.file_stores.all())
        local_policies = scope.filestoreinscope_set.get(
            file_store=self.file_store
        )
        local_policies.read_only = True
        local_policies.save()
        scope.file_stores.add(s3_file_store)
        paths = {path: f"Test file {path}".encode() for path in ("1", "2", "3")}
        artifact, _ = self.playground.create_artifact(
            paths=paths, create_files=True, skip_add_files_in_store=True
        )
        files = artifact.files.order_by("sha256")
        s3_file_store.files.add(*files)
        s3_total_size = (
            FileStore.objects.annotate_current_total_size()
            .values_list("current_total_size", flat=True)
            .get(pk=s3_file_store.pk)
        )

        with self.stub_s3_file_backend(s3_file_store) as stubber:
            stdout, stderr, exitcode = call_command(
                "vacuum_storage", verbosity=2
            )

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            f"{scope}: Draining up to {s3_total_size - 20} bytes "
            f"from {s3_file_store.name}\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(
            stderr,
            "".join(
                f"{scope}: Wanted to drain {s3_file_store.name} of "
                f"{file.hash_digest.hex()}, but it has no other upload file "
                f"stores\n"
                for file in files
            ),
        )
        self.assertEqual(exitcode, 1)

        stubber.assert_no_pending_responses()

        self.assertEqual(self.get_store_contents(self.file_store), set())

    def test_delete_orphan_file_in_upload_directory_no_delete_dry_run(
        self,
    ) -> None:
        """File is orphan but is not deleted: --dry-run."""
        orphan_file = self.create_temporary_file(
            directory=self.debusine_upload_directory
        )
        self.set_modified_time_two_days_ago(orphan_file)
        stdout, stderr, exitcode = call_command(
            "vacuum_storage", "--dry-run", verbosity=2
        )

        self.assertTrue(orphan_file.exists())

        self.assertEqual(
            stdout,
            "Checking empty directories\n"
            "Checking incomplete artifacts\n"
            "Checking redundant uploads\n"
            "Checking orphan files in upload directory\n"
            f"Deleted {orphan_file}\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_delete_orphan_file_too_new(self) -> None:
        """File is orphan but not deleted: too new."""
        orphan_file = self.create_temporary_file(
            directory=self.debusine_upload_directory
        )
        stdout, stderr, exitcode = call_command("vacuum_storage")

        self.assertTrue(orphan_file.exists())

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_delete_orphan_file(self) -> None:
        """File is orphan and deleted."""
        orphan_file = self.create_temporary_file(
            directory=self.debusine_upload_directory
        )
        self.set_modified_time_two_days_ago(orphan_file)
        call_command("vacuum_storage")

        self.assertFalse(orphan_file.exists())

    def test_file_in_upload_db_missing_from_disk(self) -> None:
        """File is referenced by the DB but not in disk for uploads."""
        artifact, _ = self.playground.create_artifact(
            paths=["README"], skip_add_files_in_store=True, create_files=True
        )
        file_in_artifact = FileInArtifact.objects.get(artifact=artifact)

        file_path = "missing-file"
        FileUpload.objects.create(
            file_in_artifact=file_in_artifact, path=file_path
        )

        stdout, stderr, exitcode = call_command("vacuum_storage")

        debusine_upload_directory = Path(self.debusine_upload_directory)

        self.assertEqual(stdout, "")
        self.assertEqual(
            stderr,
            f"File in FileUpload but not on disk: "
            f"{debusine_upload_directory / file_path}\n",
        )
        self.assertEqual(exitcode, 1)

    def _create_files_and_directories(self) -> _FilesAndDirectories:
        root = self.create_temporary_directory()

        files = {
            self.create_temporary_file(directory=root),
            self.create_temporary_file(directory=root),
        }
        directory = self.create_temporary_directory(directory=root)

        files.add(self.create_temporary_file(directory=directory))

        return {"root": root, "files": files, "directories": {directory}}

    def test_files_and_directories(self) -> None:
        """_list_directory without filters: return all files and directories."""
        files_and_directories = self._create_files_and_directories()

        self.assertEqual(
            Command._list_directory(files_and_directories["root"]),
            files_and_directories["files"]
            | files_and_directories["directories"],
        )

    def test_filter_nothing(self) -> None:
        """_list_directory with exclude all filter: return nothing."""
        files_and_directories = self._create_files_and_directories()

        self.assertEqual(
            Command._list_directory(
                files_and_directories["root"], filters=[lambda _: False]
            ),
            set(),
        )

    def test_filter_files(self) -> None:
        """_list_directory (is_file filter): return files."""
        files_and_directories = self._create_files_and_directories()

        self.assertEqual(
            Command._list_directory(
                files_and_directories["root"], filters=[lambda x: x.is_file()]
            ),
            files_and_directories["files"],
        )
