add-ideas/govoplan-files
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

intermittent commit

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-14 13:22:11 +02:00
parent b8b395e8b5
commit f3210234d3
23 changed files with 2027 additions and 555 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
tests/test_connector_providers.py Normal file
View File

@@ -0,0 +1,157 @@
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()

36
tests/test_router_contract.py Normal file
View File

@@ -0,0 +1,36 @@
from __future__ import annotations
import unittest
from govoplan_files.backend.router import router
class FilesRouterContractTests(unittest.TestCase):
def test_connector_routes_keep_existing_api_paths(self) -> None:
routes = {(tuple(sorted(route.methods or ())), route.path) for route in router.routes}
expected = {
(("GET",), "/files/connectors/providers"),
(("GET",), "/files/connectors/settings/delta"),
(("GET",), "/files/connectors/profiles"),
(("POST",), "/files/connectors/profiles"),
(("GET",), "/files/connectors/profiles/{profile_id}/browse"),
(("POST",), "/files/connectors/profiles/{profile_id}/import"),
(("POST",), "/files/connectors/profiles/{profile_id}/sync"),
(("GET",), "/files/connectors/credentials"),
(("POST",), "/files/connectors/credentials"),
(("GET",), "/files/connector-spaces"),
(("POST",), "/files/connector-spaces"),
}
self.assertTrue(expected.issubset(routes))
def test_bulk_organize_routes_keep_existing_api_paths(self) -> None:
routes = {(tuple(sorted(route.methods or ())), route.path) for route in router.routes}
self.assertIn((("POST",), "/files/bulk-rename"), routes)
self.assertIn((("POST",), "/files/transfer"), routes)
if __name__ == "__main__":
unittest.main()

67
tests/test_transfer_helpers.py Normal file
View File

@@ -0,0 +1,67 @@
from __future__ import annotations
import unittest
from govoplan_files.backend.storage.common import FileConflictResolution, FileStorageError
from govoplan_files.backend.storage.transfers import _normalize_rename_request, _normalize_transfer_request
class TransferHelperTests(unittest.TestCase):
def test_transfer_normalization_deduplicates_folder_paths_and_conflicts(self) -> None:
normalized = _normalize_transfer_request(
operation=" COPY ",
source_owner_type=" USER ",
target_owner_type=" GROUP ",
folder_paths=["Reports", "Reports/2026", "", "./Archive"],
target_folder=" Target ",
conflict_strategy="rename",
conflict_resolutions=[FileConflictResolution(target_path="Target/a.txt", action="skip")],
)
self.assertEqual("copy", normalized.operation)
self.assertEqual("user", normalized.source_owner_type)
self.assertEqual("group", normalized.target_owner_type)
self.assertEqual(["Reports", "Reports/2026", "Archive"], normalized.source_folder_paths)
self.assertEqual("Target", normalized.target_folder)
self.assertEqual("rename", normalized.conflict_strategy)
self.assertEqual("skip", normalized.conflict_resolution_map["Target/a.txt"].action)
def test_transfer_normalization_rejects_unknown_operation(self) -> None:
with self.assertRaisesRegex(FileStorageError, "Unsupported transfer operation"):
_normalize_transfer_request(
operation="link",
source_owner_type="user",
target_owner_type="group",
folder_paths=[],
target_folder="",
conflict_strategy="reject",
conflict_resolutions=None,
)
def test_rename_normalization_collapses_nested_folder_roots(self) -> None:
normalized = _normalize_rename_request(
file_ids=["file-1", "file-1", "file-2"],
folder_paths=["Reports", "Reports/2026", "Archive"],
owner_type=" USER ",
owner_id="owner-1",
mode=" prefix ",
)
self.assertEqual("prefix", normalized.mode)
self.assertEqual(["file-1", "file-2"], normalized.selected_file_ids)
self.assertEqual(["Archive", "Reports"], normalized.selected_folder_roots)
self.assertEqual("user", normalized.owner_type)
def test_rename_normalization_rejects_invalid_direct_multi_select(self) -> None:
with self.assertRaisesRegex(FileStorageError, "Direct rename requires exactly one selected item"):
_normalize_rename_request(
file_ids=["file-1", "file-2"],
folder_paths=[],
owner_type="user",
owner_id="owner-1",
mode="direct",
)
if __name__ == "__main__":
unittest.main()