refactor(api): split files workflow routers
This commit is contained in:
@@ -12,7 +12,8 @@ from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
||||
from govoplan_core.db.base import Base
|
||||
from govoplan_core.security.secrets import encrypt_secret
|
||||
from govoplan_files.backend.db.models import FileConnectorCredential, FileConnectorProfile
|
||||
from govoplan_files.backend.router import deactivate_connector_credential, deactivate_connector_profile
|
||||
from govoplan_files.backend.routes.connector_profiles import deactivate_connector_profile
|
||||
from govoplan_files.backend.routes.connector_settings import deactivate_connector_credential
|
||||
|
||||
|
||||
class Principal:
|
||||
|
||||
@@ -10,7 +10,7 @@ from unittest.mock import MagicMock
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
from govoplan_files.backend.router import discover_connector_endpoint
|
||||
from govoplan_files.backend.routes.connector_settings import discover_connector_endpoint
|
||||
from govoplan_files.backend.schemas import FileConnectorDiscoveryRequest
|
||||
from govoplan_files.backend.storage.connector_browse import ConnectorBrowseError, _profile_password, _s3_verify
|
||||
from govoplan_files.backend.storage.connector_deployment import (
|
||||
@@ -182,7 +182,7 @@ class ConnectorDiscoveryBoundaryTests(unittest.TestCase):
|
||||
credential_mode="basic",
|
||||
credentials={"username": "admin", "password_env": "MASTER_KEY_B64"},
|
||||
)
|
||||
with patch("govoplan_files.backend.router.browse_connector_profile") as browse, self.assertRaises(
|
||||
with patch("govoplan_files.backend.routes.connector_settings.browse_connector_profile") as browse, self.assertRaises(
|
||||
HTTPException
|
||||
) as raised:
|
||||
discover_connector_endpoint(payload, session=object(), principal=self._principal()) # type: ignore[arg-type]
|
||||
@@ -212,10 +212,10 @@ class ConnectorDiscoveryBoundaryTests(unittest.TestCase):
|
||||
events.append("io")
|
||||
return []
|
||||
|
||||
with patch("govoplan_files.backend.router._ensure_connector_configuration_allowed", side_effect=ensure), patch(
|
||||
"govoplan_files.backend.router._audit_connector_discovery_attempt",
|
||||
with patch("govoplan_files.backend.routes.connector_settings._ensure_connector_configuration_allowed", side_effect=ensure), patch(
|
||||
"govoplan_files.backend.routes.connector_settings._audit_connector_discovery_attempt",
|
||||
side_effect=audit,
|
||||
), patch("govoplan_files.backend.router.browse_connector_profile", side_effect=browse):
|
||||
), patch("govoplan_files.backend.routes.connector_settings.browse_connector_profile", side_effect=browse):
|
||||
response = discover_connector_endpoint(payload, session=object(), principal=self._principal()) # type: ignore[arg-type]
|
||||
|
||||
self.assertEqual("usable", response.status)
|
||||
|
||||
@@ -1,11 +1,56 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from collections import Counter
|
||||
|
||||
from govoplan_files.backend.router import router
|
||||
from govoplan_files.backend.routes.assets import router as assets_router
|
||||
from govoplan_files.backend.routes.connector_io import router as connector_io_router
|
||||
from govoplan_files.backend.routes.connector_profiles import router as connector_profiles_router
|
||||
from govoplan_files.backend.routes.connector_settings import router as connector_settings_router
|
||||
from govoplan_files.backend.routes.folders import router as folders_router
|
||||
from govoplan_files.backend.routes.listing import router as listing_router
|
||||
from govoplan_files.backend.routes.shares import router as shares_router
|
||||
from govoplan_files.backend.routes.spaces import router as spaces_router
|
||||
from govoplan_files.backend.routes.transfers import router as transfers_router
|
||||
from govoplan_files.backend.routes.uploads import router as uploads_router
|
||||
|
||||
|
||||
class FilesRouterContractTests(unittest.TestCase):
|
||||
@staticmethod
|
||||
def _operation_keys(candidate_router) -> list[tuple[str, str]]:
|
||||
return [
|
||||
(method, route.path)
|
||||
for route in candidate_router.routes
|
||||
for method in sorted(route.methods or ())
|
||||
]
|
||||
|
||||
def test_composed_router_contains_every_workflow_operation_once(self) -> None:
|
||||
workflow_routers = (
|
||||
spaces_router,
|
||||
folders_router,
|
||||
listing_router,
|
||||
uploads_router,
|
||||
connector_settings_router,
|
||||
connector_io_router,
|
||||
connector_profiles_router,
|
||||
assets_router,
|
||||
shares_router,
|
||||
transfers_router,
|
||||
)
|
||||
expected = [
|
||||
operation
|
||||
for workflow_router in workflow_routers
|
||||
for operation in self._operation_keys(workflow_router)
|
||||
]
|
||||
actual = self._operation_keys(router)
|
||||
|
||||
self.assertEqual(expected, actual)
|
||||
self.assertEqual(41, len(actual))
|
||||
self.assertFalse(
|
||||
[operation for operation, count in Counter(actual).items() if count > 1]
|
||||
)
|
||||
|
||||
def test_connector_routes_keep_existing_api_paths(self) -> None:
|
||||
routes = {(tuple(sorted(route.methods or ())), route.path) for route in router.routes}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user