mostovik-backend/tests/apps/parsers/test_private_sro_artifact_storage.py

"""SRO raw files stay outside every public MEDIA_ROOT serving path."""

import stat
from datetime import timedelta
from pathlib import Path
from tempfile import TemporaryDirectory
from types import ModuleType

from apps.parsers.models import ParserSourceArtifact
from apps.parsers.source_artifacts import cleanup_parser_source_artifacts
from django.conf import settings
from django.conf.urls.static import static
from django.contrib.admin.sites import AdminSite
from django.core.exceptions import ImproperlyConfigured
from django.core.files.base import ContentFile
from django.db.migrations.writer import MigrationWriter
from django.test import RequestFactory, TestCase, override_settings
from django.utils import timezone


class PrivateSroArtifactStorageTest(TestCase):
    def setUp(self):
        self.temporary = TemporaryDirectory(prefix="private-sro-artifacts-")
        self.addCleanup(self.temporary.cleanup)
        self.media = Path(self.temporary.name) / "media"
        self.private = Path(self.temporary.name) / "private"
        self.media.mkdir()
        self.overrides = override_settings(
            DEBUG=True,
            MEDIA_ROOT=self.media,
            PARSER_PRIVATE_ARTIFACT_ROOT=self.private,
        )
        self.overrides.enable()
        self.addCleanup(self.overrides.disable)
        # Same Django DEBUG media-serving helper as core.urls; no authentication.
        routes = ModuleType("private_sro_media_routes")
        routes.urlpatterns = static(
            settings.MEDIA_URL, document_root=settings.MEDIA_ROOT
        )
        self.routes = override_settings(ROOT_URLCONF=routes)
        self.routes.enable()
        self.addCleanup(self.routes.disable)

    def save(self, source="sro_membership_check"):
        artifact = ParserSourceArtifact.objects.create(
            source=source, load_batch=1, status=ParserSourceArtifact.Status.PUBLISHED
        )
        artifact.file.save("fixture.zip", ContentFile(b"synthetic-only"), save=True)
        return artifact

    def test_anonymous_media_get_cannot_read_saved_sro_raw(self):
        artifact = self.save()
        response = self.client.get(settings.MEDIA_URL + artifact.file.name)
        self.assertEqual(response.status_code, 404)
        self.assertFalse((self.media / artifact.file.name).exists())
        self.assertTrue(Path(artifact.file.path).is_relative_to(self.private.resolve()))

    def test_private_file_read_permissions_and_url_rejection(self):
        artifact = self.save()
        artifact.refresh_from_db()
        with artifact.file.open("rb") as handle:
            self.assertEqual(handle.read(), b"synthetic-only")
        self.assertEqual(artifact.file.size, len(b"synthetic-only"))
        self.assertEqual(stat.S_IMODE(Path(artifact.file.path).stat().st_mode), 0o600)
        self.assertEqual(
            stat.S_IMODE(Path(artifact.file.path).parent.stat().st_mode), 0o700
        )
        with self.assertRaisesMessage(
            ValueError, "Private parser artifacts have no public URL"
        ):
            _ = artifact.file.url

    def test_other_sources_keep_default_storage_and_public_url(self):
        artifact = self.save("budget_ubpandnubp")
        self.assertTrue(Path(artifact.file.path).is_relative_to(self.media))
        response = self.client.get(artifact.file.url)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(b"".join(response.streaming_content), b"synthetic-only")

    def test_private_root_inside_media_is_rejected_before_file_write(self):
        for root in (self.media, self.media / "private"):
            with self.subTest(root=root.name), override_settings(
                PARSER_PRIVATE_ARTIFACT_ROOT=root
            ), self.assertRaisesMessage(ImproperlyConfigured, "outside MEDIA_ROOT"):
                self.save()
        self.assertEqual(list(self.media.iterdir()), [])

    def test_symlinked_private_root_inside_media_is_rejected(self):
        alias = Path(self.temporary.name) / "alias"
        alias.symlink_to(self.media, target_is_directory=True)
        with override_settings(
            PARSER_PRIVATE_ARTIFACT_ROOT=alias
        ), self.assertRaisesMessage(ImproperlyConfigured, "outside MEDIA_ROOT"):
            self.save()
        self.assertEqual(list(self.media.iterdir()), [])

    def test_existing_retention_removes_private_file_and_model(self):
        artifact = self.save()
        path = Path(artifact.file.path)
        ParserSourceArtifact.objects.filter(pk=artifact.pk).update(
            created_at=timezone.now() - timedelta(days=100)
        )
        self.assertEqual(cleanup_parser_source_artifacts(minimum_versions=0), 1)
        self.assertFalse(path.exists())
        self.assertFalse(ParserSourceArtifact.objects.filter(pk=artifact.pk).exists())

    def test_storage_deconstruction_does_not_capture_machine_paths(self):
        storage = ParserSourceArtifact._meta.get_field("file").storage
        value, imports = MigrationWriter.serialize(storage)
        self.assertEqual(value, "apps.parsers.artifact_storage.ParserArtifactStorage()")
        self.assertEqual(imports, {"import apps.parsers.artifact_storage"})
        with override_settings(PARSER_PRIVATE_ARTIFACT_ROOT=self.private / "changed"):
            self.assertEqual(MigrationWriter.serialize(storage), (value, imports))

    def test_private_artifact_admin_uses_readonly_text_instead_of_public_link(self):
        from apps.parsers.admin import ParserSourceArtifactAdmin

        artifact = self.save()
        model_admin = ParserSourceArtifactAdmin(ParserSourceArtifact, AdminSite())
        request = RequestFactory().get("/admin/")
        fields = model_admin.get_fields(request, artifact)
        self.assertNotIn("file", fields)
        self.assertIn("private_file_name", fields)
        self.assertIn(
            "private_file_name", model_admin.get_readonly_fields(request, artifact)
        )
        self.assertEqual(model_admin.private_file_name(artifact), artifact.file.name)