from __future__ import annotations import unittest from datetime import UTC, datetime from unittest.mock import patch from govoplan_files.backend.storage.connector_browse import browse_connector_profile from govoplan_files.backend.storage.connector_browse import ConnectorBrowseUnsupported from govoplan_files.backend.storage.connector_imports import read_connector_file from govoplan_files.backend.storage.connector_profiles import ConnectorProfile, connector_profiles_from_payload from govoplan_files.backend.storage.connector_providers import connector_provider_descriptors class FakeBody: def __init__(self, data: bytes) -> None: self.data = data def read(self, _limit: int) -> bytes: return self.data class FakeS3Client: def __init__(self) -> None: self.list_objects_request: dict[str, object] | None = None self.head_request: dict[str, object] | None = None self.get_request: dict[str, object] | None = None def list_objects_v2(self, **kwargs: object) -> dict[str, object]: self.list_objects_request = dict(kwargs) return { "CommonPrefixes": [{"Prefix": "root/reports/"}], "Contents": [ { "Key": "root/report.xlsx", "Size": 123, "LastModified": datetime(2026, 7, 12, 10, 0, tzinfo=UTC), "ETag": '"etag-1"', "StorageClass": "STANDARD", } ], "NextContinuationToken": "next-page", } def list_buckets(self) -> dict[str, object]: return {"Buckets": [{"Name": "archive", "CreationDate": datetime(2026, 7, 12, 9, 0, tzinfo=UTC)}]} def head_object(self, **kwargs: object) -> dict[str, object]: self.head_request = dict(kwargs) return { "ContentLength": 13, "ContentType": "text/plain", "ETag": '"etag-2"', "VersionId": "version-1", "ChecksumSHA256": "checksum", } def get_object(self, **kwargs: object) -> dict[str, object]: self.get_request = dict(kwargs) return {"Body": FakeBody(b"hello s3\n"), "ContentType": "text/plain", "ETag": '"etag-2"'} def s3_profile(**overrides: object) -> ConnectorProfile: values = { "id": "dev-s3", "label": "Dev S3", "provider": "s3", "endpoint_url": "http://127.0.0.1:9000", "base_path": "root", "credential_mode": "basic", "username": "access-key", "password_value": "secret-key", "metadata": {"bucket": "govoplan", "region": "eu-central-1", "path_style": True}, } values.update(overrides) return ConnectorProfile(**values) class ConnectorProviderTests(unittest.TestCase): def test_provider_descriptors_include_s3_and_reserved_microsoft_providers(self) -> None: descriptors = {descriptor.provider: descriptor for descriptor in connector_provider_descriptors()} self.assertTrue(descriptors["s3"].implemented) self.assertTrue(descriptors["s3"].browse_supported) self.assertEqual("boto3", descriptors["s3"].optional_dependency) self.assertFalse(descriptors["sharepoint"].implemented) self.assertFalse(descriptors["onedrive"].implemented) def test_profile_payload_accepts_new_providers_and_redacts_secret_metadata(self) -> None: profiles = connector_profiles_from_payload( { "profiles": [ { "id": "dev-s3", "label": "Dev S3", "provider": "s3", "username": "access-key", "password": "secret-key", "metadata": {"bucket": "govoplan", "secret_access_key": "do-not-return"}, }, {"id": "sharepoint", "provider": "sharepoint"}, {"id": "onedrive", "provider": "onedrive"}, ] } ) by_id = {profile.id: profile for profile in profiles} self.assertEqual("s3", by_id["dev-s3"].provider) self.assertTrue(by_id["dev-s3"].credentials_configured) self.assertEqual({"bucket": "govoplan"}, dict(by_id["dev-s3"].metadata)) self.assertEqual("sharepoint", by_id["sharepoint"].provider) self.assertEqual("onedrive", by_id["onedrive"].provider) def test_s3_browse_lists_prefixes_and_objects_with_provenance_metadata(self) -> None: client = FakeS3Client() with patch("govoplan_files.backend.storage.connector_browse._s3_client", return_value=client): items = browse_connector_profile(s3_profile(), path="") self.assertEqual({"Bucket": "govoplan", "Prefix": "root/", "Delimiter": "/", "MaxKeys": 1000}, client.list_objects_request) self.assertEqual(["folder", "file"], [item.kind for item in items]) self.assertEqual("reports", items[0].path) self.assertEqual("report.xlsx", items[1].path) self.assertEqual("govoplan:root/report.xlsx", items[1].external_id) self.assertEqual("root/report.xlsx", items[1].metadata["key"]) self.assertEqual("next-page", items[1].metadata["next_continuation_token"]) def test_s3_browse_lists_buckets_when_profile_has_no_bucket(self) -> None: client = FakeS3Client() with patch("govoplan_files.backend.storage.connector_browse._s3_client", return_value=client): items = browse_connector_profile(s3_profile(metadata={}), path="") self.assertEqual(["archive"], [item.path for item in items]) def test_s3_browse_reports_missing_optional_dependency(self) -> None: with patch("govoplan_files.backend.storage.connector_browse.import_module", side_effect=ImportError): with self.assertRaisesRegex(ConnectorBrowseUnsupported, "boto3"): browse_connector_profile(s3_profile(), path="") def test_s3_import_downloads_object_and_preserves_remote_identity(self) -> None: client = FakeS3Client() with patch("govoplan_files.backend.storage.connector_imports._s3_client", return_value=client): downloaded = read_connector_file(s3_profile(), library_id="", path="report.txt", max_bytes=1024) self.assertEqual({"Bucket": "govoplan", "Key": "root/report.txt"}, client.head_request) self.assertEqual({"Bucket": "govoplan", "Key": "root/report.txt", "VersionId": "version-1"}, client.get_request) self.assertEqual("report.txt", downloaded.filename) self.assertEqual(b"hello s3\n", downloaded.data) self.assertEqual("version-1", downloaded.revision) self.assertEqual("govoplan:root/report.txt", downloaded.external_id) self.assertEqual("s3://govoplan/root/report.txt", downloaded.external_url) self.assertEqual("checksum", downloaded.metadata["checksum_sha256"]) if __name__ == "__main__": unittest.main()