98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_access.backend.db import models as access_models # noqa: F401
|
|
from govoplan_core.core.provider_governance import ExternalProviderStateContext
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_files.backend.db.models import FileConnectorProfile, FileConnectorSpace
|
|
from govoplan_files.backend.manifest import manifest
|
|
from govoplan_files.backend.provider_state import (
|
|
REMOTE_STORAGE_PROVIDER_ID,
|
|
remote_storage_provider_states,
|
|
)
|
|
|
|
|
|
class FilesProviderStateTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=(FileConnectorProfile.__table__, FileConnectorSpace.__table__),
|
|
)
|
|
self.session = sessionmaker(bind=self.engine, expire_on_commit=False)()
|
|
self.profile = FileConnectorProfile(
|
|
id="profile-1",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
label="Remote WebDAV",
|
|
provider="webdav",
|
|
endpoint_url="https://files.example.test/dav/",
|
|
enabled=True,
|
|
)
|
|
self.session.add_all(
|
|
(
|
|
self.profile,
|
|
FileConnectorSpace(
|
|
id="space-1",
|
|
tenant_id="tenant-1",
|
|
owner_type="tenant",
|
|
label="Documents",
|
|
connector_profile_id=self.profile.id,
|
|
provider="webdav",
|
|
remote_path="/documents",
|
|
read_only=True,
|
|
is_active=True,
|
|
),
|
|
)
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_state_does_not_overstate_live_health_or_expose_endpoint(self) -> None:
|
|
state = remote_storage_provider_states(
|
|
ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
|
|
)[0]
|
|
|
|
self.assertEqual("external_mirror", state.authority_mode)
|
|
self.assertEqual("unknown", state.health)
|
|
self.assertEqual("attention", state.recovery)
|
|
self.assertEqual(1, state.metrics["active_spaces"])
|
|
self.assertNotIn("files.example.test", str(state.to_dict()))
|
|
self.assertEqual(
|
|
(),
|
|
remote_storage_provider_states(
|
|
ExternalProviderStateContext(
|
|
session=self.session,
|
|
tenant_id="tenant-2",
|
|
)
|
|
),
|
|
)
|
|
|
|
def test_unimplemented_profile_is_fail_closed_and_manifest_registered(self) -> None:
|
|
self.profile.provider = "sharepoint"
|
|
self.session.flush()
|
|
|
|
state = remote_storage_provider_states(
|
|
ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
|
|
)[0]
|
|
|
|
self.assertEqual("error", state.health)
|
|
self.assertEqual("unsupported", state.recovery)
|
|
self.assertEqual(REMOTE_STORAGE_PROVIDER_ID, manifest.external_providers[0].id)
|
|
self.assertEqual(
|
|
REMOTE_STORAGE_PROVIDER_ID,
|
|
manifest.external_provider_state_providers[0].provider_id,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|