Integrate file connectors with credential envelopes
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
@@ -8,6 +9,7 @@ from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from govoplan_access.backend.db import models as access_models # noqa: F401 - resolve Files user foreign keys
|
||||
from govoplan_core.db.base import Base
|
||||
from govoplan_core.security.credential_envelopes import CredentialEnvelope
|
||||
from govoplan_core.security.secrets import decrypt_secret, encrypt_secret
|
||||
from govoplan_files.backend.db.models import FileConnectorProfile
|
||||
from govoplan_files.backend.storage.common import FileStorageError
|
||||
@@ -20,7 +22,13 @@ from govoplan_files.backend.storage.connector_profile_store import (
|
||||
class ConnectorProfileStoreUpdateTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(self.engine, tables=[FileConnectorProfile.__table__])
|
||||
Base.metadata.create_all(
|
||||
self.engine,
|
||||
tables=[
|
||||
FileConnectorProfile.__table__,
|
||||
CredentialEnvelope.__table__,
|
||||
],
|
||||
)
|
||||
self.session = sessionmaker(bind=self.engine)()
|
||||
self.row = FileConnectorProfile(
|
||||
id="profile-1",
|
||||
@@ -171,6 +179,50 @@ class ConnectorProfileStoreUpdateTests(unittest.TestCase):
|
||||
self.assertEqual([self.row.id], [profile.id for profile in profiles])
|
||||
self.assertEqual("original-password", profiles[0].password_value)
|
||||
|
||||
def test_reusable_credential_resolves_and_revocation_keeps_profile_visible(self) -> None:
|
||||
credential = CredentialEnvelope(
|
||||
id="shared-credential",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
name="Shared WebDAV login",
|
||||
credential_kind="username_password",
|
||||
public_data={"username": "ada"},
|
||||
secret_data_encrypted=encrypt_secret(json.dumps({"password": "secret"})),
|
||||
secret_keys=["password"],
|
||||
allowed_modules=["files"],
|
||||
allowed_server_refs=[],
|
||||
inherit_to_lower_scopes=True,
|
||||
is_active=True,
|
||||
revision="revision-1",
|
||||
)
|
||||
self.row.credential_profile_id = "credential-envelope:shared-credential"
|
||||
self.row.username = None
|
||||
self.row.password_encrypted = None
|
||||
self.row.token_encrypted = None
|
||||
self.session.add(credential)
|
||||
self.session.commit()
|
||||
|
||||
profile = list_database_connector_profiles(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
)[0]
|
||||
|
||||
self.assertEqual("ada", profile.username)
|
||||
self.assertEqual("secret", profile.password_value)
|
||||
self.assertTrue(profile.credentials_configured)
|
||||
|
||||
credential.is_active = False
|
||||
self.session.commit()
|
||||
profile = list_database_connector_profiles(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
)[0]
|
||||
|
||||
self.assertIsNone(profile.password_value)
|
||||
self.assertFalse(profile.credentials_configured)
|
||||
self.assertTrue(profile.metadata["credential_unavailable"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user