# 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 os
from datetime import datetime, timedelta
from pathlib import Path
from typing import TypedDict
from unittest import mock

from django.utils import timezone

from debusine.db.context import context
from debusine.db.models import (
    Artifact,
    File,
    FileInArtifact,
    FileStore,
    FileUpload,
)
from debusine.django.management.tests import call_command
from debusine.project import settings
from debusine.server.management.commands.vacuum_storage import Command
from debusine.test.django import TestCase


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

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


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

    playground_memory_file_store = False

    @context.disable_permission_checks()
    def setUp(self) -> None:
        """Set common objects and settings."""
        # Using override_settings does not work (perhaps because it's a
        # lambda calculated setting)
        self.debusine_store_directory = str(self.create_temporary_directory())
        self.original_debusine_store_directory = (
            settings.DEBUSINE_STORE_DIRECTORY
        )
        settings.DEBUSINE_STORE_DIRECTORY = 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_STORE_DIRECTORY
        settings.DEBUSINE_UPLOAD_DIRECTORY = self.debusine_upload_directory

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

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

        Also test that is not finding a file part of an artifact.
        """
        with context.disable_permission_checks():
            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(
            "vacuum_storage", "--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 empty directories\n"
            "Checking unfinished file uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from stores\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

    @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

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

        Also test that file part of an artifact is not deleted.
        """
        with context.disable_permission_checks():
            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("vacuum_storage")

        # 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("vacuum_storage")

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

    def test_delete_empty_directory_dry_run(self) -> None:
        """Report delete an empty directory (but dry-run: does not delete)."""
        with context.disable_permission_checks():
            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 orphan files from stores\n"
            "Checking empty directories\n"
            f"Deleted empty directory {empty_directory}\n"
            "Checking unfinished file uploads\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from stores\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)."""
        with context.disable_permission_checks():
            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_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)

        rmdir_patcher = mock.patch("pathlib.Path.rmdir", autospec=True)
        rmdir_mocked = rmdir_patcher.start()
        rmdir_mocked.side_effect = OSError()
        self.addCleanup(rmdir_patcher.stop)
        stdout, stderr, exitcode = call_command("vacuum_storage")

        rmdir_mocked.assert_called_once()

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

        self.assertTrue(empty_dir.is_dir())

    def test_delete_unfinished_uploads_no_delete_dry_run(self) -> None:
        """Unfinished, old FileUpload is not deleted: dry run."""
        with context.disable_permission_checks():
            artifact, _ = self.playground.create_artifact(
                paths=["README"], create_files=True
            )
            file_in_artifact = FileInArtifact.objects.get(artifact=artifact)

            temporary_file = self.create_temporary_file(
                directory=self.debusine_upload_directory
            )

            file_upload = FileUpload.objects.create(
                file_in_artifact=file_in_artifact, path=temporary_file.name
            )
            file_upload.last_activity_at = timezone.now() - timedelta(days=2)
            file_upload.save()

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

        self.assertEqual(
            stdout,
            "Checking orphan files from stores\n"
            "Checking empty directories\n"
            "Checking unfinished file uploads\n"
            f"Deleted unfinished upload {file_upload.id}\n"
            "Checking orphan files in upload directory\n"
            "Checking missing files from stores\n"
            "Checking missing files from upload\n",
        )
        self.assertEqual(stderr, "")
        self.assertEqual(exitcode, 0)

        # Check that the FileUpload row was not deleted
        self.assertQuerySetEqual(
            FileUpload.objects.filter(id=file_upload.id), [file_upload]
        )

    def test_delete_unfinished_upload_deleted(self) -> None:
        """Unfinished and old FileUpload and its file is deleted."""
        with context.disable_permission_checks():
            artifact, _ = self.playground.create_artifact(
                paths=["README"], create_files=True
            )
            file_in_artifact = FileInArtifact.objects.get(artifact=artifact)
            file_upload = FileUpload.objects.create(
                file_in_artifact=file_in_artifact, path="some-file-path"
            )
            file_upload.last_activity_at = timezone.now() - timedelta(days=2)
            file_upload.path = (
                Path(self.debusine_upload_directory) / "the-file"
            ).as_posix()
            file_upload.save()

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

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

        # Check that the FileUpload row was not deleted
        self.assertQuerySetEqual(
            FileUpload.objects.filter(id=file_upload.id), []
        )

        # The file also got deleted
        self.assertFalse(Path(file_upload.path).exists())

    def test_delete_unfinished_upload_not_deleted_recent_activity(self) -> None:
        """Unfinished FileUpload is not deleted: recent activity."""
        with context.disable_permission_checks():
            artifact, _ = self.playground.create_artifact(
                paths=["README"], create_files=True
            )
            file_in_artifact = FileInArtifact.objects.get(artifact=artifact)

            upload_temporary_file = self.create_temporary_file(
                directory=self.debusine_upload_directory
            )
            file_upload = FileUpload.objects.create(
                file_in_artifact=file_in_artifact,
                path=upload_temporary_file.name,
            )

        call_command("vacuum_storage")

        # FileUpload did not get deleted: recent activity
        self.assertQuerySetEqual(
            FileUpload.objects.filter(id=file_upload.id), [file_upload]
        )

    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 orphan files from stores\n"
            "Checking empty directories\n"
            "Checking unfinished file uploads\n"
            "Checking orphan files in upload directory\n"
            f"Deleted {orphan_file}\n"
            "Checking missing files from stores\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_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 = FileStore.objects.create(
            name="memory",
            backend=FileStore.BackendChoices.MEMORY,
            configuration={"name": "memory"},
        )
        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("vacuum_storage")

        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_file_in_upload_db_missing_from_disk(self) -> None:
        """File is referenced by the DB but not in disk for uploads."""
        with context.disable_permission_checks():
            artifact, _ = self.playground.create_artifact(
                paths=["README"], 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_list_directory_files_and_directories_raise_error(self) -> None:
        """_list_directory raise ValueError for incompatible parameters."""
        with self.assertRaisesRegex(
            ValueError,
            'Parameters "files_only" and "directories_only" are incompatible',
        ):
            Command._list_directory("/", files_only=True, directories_only=True)

    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_files_only(self) -> None:
        """_list_directory(files_only) return only files (no directories)."""
        files_and_directories = self._create_files_and_directories()

        self.assertEqual(
            Command._list_directory(
                files_and_directories["root"], files_only=True
            ),
            files_and_directories["files"],
        )

    def test_directories_only(self) -> None:
        """_list_directory(directories_only) return only directories."""
        files_and_directories = self._create_files_and_directories()

        self.assertEqual(
            Command._list_directory(
                files_and_directories["root"], directories_only=True
            ),
            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 x: False],  # noqa: U100
            ),
            set(),
        )

    def test_filter_all(self) -> None:
        """_list_directory (only files, include all filter): return files."""
        files_and_directories = self._create_files_and_directories()

        self.assertEqual(
            Command._list_directory(
                files_and_directories["root"],
                files_only=True,
                filters=[lambda x: True],  # noqa: U100,
            ),
            files_and_directories["files"],
        )

    @staticmethod
    def _set_modified_time_two_days_ago(path: Path) -> None:
        two_days_ago = (datetime.now() - timedelta(days=2)).timestamp()
        os.utime(path, (two_days_ago, two_days_ago))
