# 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 fsck."""

from io import StringIO
from typing import Any
from unittest import mock

from django.utils import timezone

from debusine.db.models import File, FileStore
from debusine.django.management.tests import call_command
from debusine.server.file_backend.interface import FileBackendEntryInterface
from debusine.server.file_backend.s3 import S3FileBackendEntry
from debusine.server.management.commands.fsck import Command
from debusine.server.management.commands.tests.test_vacuum_storage import (
    VacuumStorageTestMixin,
)
from debusine.test.django import TestCase


class FsckCommandTests(VacuumStorageTestMixin, TestCase):
    """Tests for fsck command."""

    def test_operates_on_all_stores_by_default(self) -> None:
        self.create_memory_file_store()
        with (
            mock.patch.object(
                Command, "_delete_unreferenced_files"
            ) as _delete_unreferenced_files,
            mock.patch.object(
                Command, "_report_missing_files_stores_referenced_by_the_db"
            ) as _report_missing_files_stores_referenced_by_the_db,
        ):
            call_command("fsck", "--dry-run", verbosity=2)
        self.assertQuerySetEqual(
            _delete_unreferenced_files.call_args.kwargs["stores"],
            list(FileStore.objects.all()),
            ordered=False,
        )
        self.assertQuerySetEqual(
            _report_missing_files_stores_referenced_by_the_db.call_args.kwargs[
                "stores"
            ],
            list(FileStore.objects.all()),
            ordered=False,
        )

    def test_operates_on_specified_stores_only(self) -> None:
        memory_store = self.create_memory_file_store()
        with (
            mock.patch.object(
                Command, "_delete_unreferenced_files"
            ) as _delete_unreferenced_files,
            mock.patch.object(
                Command, "_report_missing_files_stores_referenced_by_the_db"
            ) as _report_missing_files_stores_referenced_by_the_db,
        ):
            call_command("fsck", "--store=memory", "--dry-run", verbosity=2)
        self.assertQuerySetEqual(
            _delete_unreferenced_files.call_args.kwargs["stores"],
            [memory_store],
        )
        self.assertQuerySetEqual(
            _report_missing_files_stores_referenced_by_the_db.call_args.kwargs[
                "stores"
            ],
            [memory_store],
        )

    def test_unknown_store_prints_an_error(self) -> None:
        _stdout, stderr, exit_code = call_command(
            "fsck", "--store=unknown", "--dry-run", verbosity=2
        )
        self.assertEqual(stderr, "FileStore 'unknown' does not exist.\n")
        self.assertEqual(exit_code, 1)

    def test_delete_file_in_disk_not_in_store_dry_run(self) -> None:
        """
        Test fsck find file in disk not in DB.

        Also test that is not finding a file part of an artifact.
        """
        self.playground.create_artifact(paths=["README"], create_files=True)

        orphan_file = self.create_temporary_file(
            directory=self.debusine_store_directory
        )
        self.set_modified_time_two_days_ago(orphan_file)

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

        # The file was not deleted: dry-run was used
        self.assertTrue(orphan_file.exists())

        self.assertEqual(
            stdout,
            (
                "Checking orphan files from stores\n"
                f"Deleted {orphan_file}\n"
                "Checking missing files from stores\n"
            ),
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_delete_file_in_disk_not_in_db(self) -> None:
        """
        Test fsck find and delete file that is in disk but not in DB.

        Also test that file part of an artifact is not deleted.
        """
        path_in_artifact = "README"
        artifact, _ = self.playground.create_artifact(
            paths=[path_in_artifact], create_files=True
        )

        temporary_file = self.create_temporary_file(
            directory=self.debusine_store_directory
        )
        self.set_modified_time_two_days_ago(temporary_file)
        stdout, stderr, exitcode = call_command("fsck")

        # The file was deleted
        self.assertFalse(temporary_file.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.exists())

        # No verbosity, no output
        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    def test_file_not_deleted_too_new(self) -> None:
        """Do not delete the file: it's too new."""
        temporary_file = self.create_temporary_file(
            directory=self.debusine_store_directory
        )
        call_command("fsck")

        # The file was not deleted
        self.assertTrue(temporary_file.exists())

    def test_file_in_store_db_missing_from_storage(self) -> None:
        """File is referenced by the DB but not in storage."""
        path_in_artifact = "README"
        artifact, _ = self.playground.create_artifact(
            paths=[path_in_artifact], create_files=True
        )

        file_path = self.get_local_file_path(artifact, path_in_artifact)
        hash_hex = File.calculate_hash(file_path).hex()

        memory_file_store = self.create_memory_file_store()
        memory_file = memory_file_store.get_backend_object().add_file(file_path)

        file_path.unlink()
        memory_file_store.get_backend_object().get_entry(memory_file).remove()

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

        self.assertEqual(stdout, "")
        self.assertEqual(
            stderr,
            f"File in FileStore {self.file_store.name} but not in storage: "
            f"{file_path}\n"
            f"File in FileStore {memory_file_store.name} but not in storage: "
            f"{hash_hex}\n",
        )
        self.assertEqual(exitcode, 1)

    def test_report_missing_files_stores_referenced_by_the_db(self) -> None:
        """
        Verify method does not report a missing file.

        The possible missing file is added during the execution of
        _report_missing_files_stores_referenced_by_the_db.
        """
        path_in_artifact = "README"
        artifact, _ = self.playground.create_artifact(
            paths=[path_in_artifact], create_files=True
        )
        local_path = self.get_local_file_path(artifact, path_in_artifact)

        # Create a Memory store
        memory_store = self.create_memory_file_store()
        memory_backend = memory_store.get_backend_object()

        initial_number_files = len(list(memory_backend.list_entries()))

        def add_file_side_effect(
            store: FileStore,
        ) -> set[FileBackendEntryInterface[Any, Any]]:
            assert store == memory_store
            memory_backend.add_file(local_path)

            return Command._entries_from_files_in_file_store_db(store)

        stdout = StringIO()
        stderr = StringIO()
        cmd = Command(stdout=stdout, stderr=stderr, no_color=True)
        cmd.verbosity = 2

        with mock.patch.object(
            cmd,
            "_entries_from_files_in_file_store_db",
            autospec=True,
            side_effect=add_file_side_effect,
        ):
            ok = cmd._report_missing_files_stores_referenced_by_the_db(
                stores=FileStore.objects.filter(id=memory_store.id)
            )

        # One file was added during the execution of
        # Command._report_missing_files_store_referenced_by_the_db
        self.assertEqual(
            len(list(memory_backend.list_entries())), initial_number_files + 1
        )

        # No files were missing, even if a file was added half-way the operation
        self.assertTrue(ok)

        # Output does not indicate any file
        self.assertEqual(
            stdout.getvalue(), "Checking missing files from stores\n"
        )
        self.assertEqual(stderr.getvalue(), "")

    def test_entries_from_files_in_file_store_db_queries(self) -> None:
        memory_file_store = self.create_memory_file_store()
        files = [
            self.playground.create_file(contents=b"\0" * length)
            for length in (1, 2, 3)
        ]
        memory_file_store.files.add(*files)
        expected_identifiers = {file.hash_digest.hex() for file in files}

        with self.assertNumQueries(1):
            self.assertEqual(
                {
                    str(entry)
                    for entry in Command._entries_from_files_in_file_store_db(
                        memory_file_store
                    )
                },
                expected_identifiers,
            )

    def test_delete_unreferenced_race(self) -> None:
        """
        Attempt to cleanup an unreferenced file that got deleted already.

        Reproduce a race in _delete_unreferenced_race when the file is deleted
        between listing the store and statting the mtime of the file.

        This wouldn't happen in the S3 store (as mtimes are remembered from the
        file listing), but the S3 store is convenient to mock for testing.
        """
        s3_file_store = self.create_s3_file_store()
        stdout = StringIO()
        stderr = StringIO()
        cmd = Command(stdout=stdout, stderr=stderr, no_color=True)
        cmd.verbosity = 2

        with (
            self.stub_s3_file_backend(s3_file_store) as stubber,
            mock.patch.object(
                S3FileBackendEntry, "get_mtime", return_value=None
            ),
        ):
            stubber.add_response(
                "list_objects_v2",
                {
                    "Contents": [
                        {
                            "Key": "orphan-file",
                            # The stub needs a non-None value, but we patch
                            # get_mtime() to return None
                            "LastModified": timezone.now(),
                        }
                    ]
                },
                {"Bucket": "test-bucket"},
            )
            cmd._delete_unreferenced_files(
                stores=FileStore.objects.filter(id=s3_file_store.id),
                dry_run=False,
            )

        self.assertEqual(
            stdout.getvalue(), "Checking orphan files from stores\n"
        )
        self.assertEqual(stderr.getvalue(), "")
