142 lines
5.6 KiB
Python
142 lines
5.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from uuid import uuid4
|
|
|
|
from services.control_api.persistence.db import Base, create_engine_from_url, session_factory
|
|
from services.control_api.persistence.demo_client import (
|
|
DEMO_CLIENT_ID,
|
|
DEMO_CLIENT_ISSUER,
|
|
DEMO_CLIENT_SUBJECT,
|
|
provision_demo_client,
|
|
)
|
|
from services.control_api.persistence.models import (
|
|
ClientBindingRow,
|
|
ClientRow,
|
|
EnvironmentRow,
|
|
ProjectRow,
|
|
VirtualServerRow,
|
|
WorkspaceRow,
|
|
)
|
|
|
|
|
|
class DemoClientProvisioningTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine_from_url("sqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine)
|
|
self.sessions = session_factory(self.engine)
|
|
self.workspace_id = uuid4()
|
|
self.project_id = uuid4()
|
|
self.environment_id = uuid4()
|
|
self.server_id = uuid4()
|
|
with self.sessions() as session, session.begin():
|
|
session.add(WorkspaceRow(id=self.workspace_id, slug="itworx", name="ITWorx"))
|
|
session.flush()
|
|
session.add(
|
|
ProjectRow(
|
|
id=self.project_id,
|
|
workspace_id=self.workspace_id,
|
|
slug="mobilityops",
|
|
name="MobilityOps",
|
|
)
|
|
)
|
|
session.flush()
|
|
session.add(
|
|
EnvironmentRow(
|
|
id=self.environment_id,
|
|
workspace_id=self.workspace_id,
|
|
project_id=self.project_id,
|
|
name="prod",
|
|
classification="production",
|
|
)
|
|
)
|
|
session.flush()
|
|
session.add(
|
|
VirtualServerRow(
|
|
id=self.server_id,
|
|
workspace_id=self.workspace_id,
|
|
project_id=self.project_id,
|
|
environment_id=self.environment_id,
|
|
slug="mobilityops-prod-readonly",
|
|
desired_hash="a" * 64,
|
|
enabled=True,
|
|
# Must satisfy the [0-9a-f]{32} route-identifier shape that
|
|
# SqlAlchemyAliasSource validates -- provisioning now
|
|
# requires a real external gateway route.
|
|
gateway_external_id="a" * 32,
|
|
)
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.engine.dispose()
|
|
|
|
def test_provisions_client_and_binding(self) -> None:
|
|
with self.sessions() as session, session.begin():
|
|
result = provision_demo_client(session)
|
|
self.assertEqual(result.client_id, DEMO_CLIENT_ID)
|
|
with self.sessions() as session:
|
|
client = session.get(ClientRow, DEMO_CLIENT_ID)
|
|
assert client is not None
|
|
self.assertEqual(client.issuer, DEMO_CLIENT_ISSUER)
|
|
self.assertEqual(client.subject, DEMO_CLIENT_SUBJECT)
|
|
self.assertEqual(client.status, "active")
|
|
binding = session.get(ClientBindingRow, result.binding_id)
|
|
assert binding is not None
|
|
self.assertEqual(binding.client_id, DEMO_CLIENT_ID)
|
|
self.assertEqual(binding.virtual_server_id, self.server_id)
|
|
self.assertTrue(binding.enabled)
|
|
|
|
def test_idempotent_on_repeat_run(self) -> None:
|
|
with self.sessions() as session, session.begin():
|
|
first = provision_demo_client(session)
|
|
with self.sessions() as session, session.begin():
|
|
second = provision_demo_client(session)
|
|
self.assertEqual(first, second)
|
|
with self.sessions() as session:
|
|
self.assertEqual(
|
|
session.query(ClientRow).filter(ClientRow.id == DEMO_CLIENT_ID).count(), 1
|
|
)
|
|
self.assertEqual(
|
|
session.query(ClientBindingRow)
|
|
.filter(ClientBindingRow.client_id == DEMO_CLIENT_ID)
|
|
.count(),
|
|
1,
|
|
)
|
|
|
|
def test_raises_when_virtual_server_missing(self) -> None:
|
|
with self.sessions() as session, session.begin():
|
|
with self.assertRaises(ValueError):
|
|
provision_demo_client(session, project_slug="does-not-exist")
|
|
|
|
def test_raises_when_virtual_server_is_not_uniquely_determined(self) -> None:
|
|
"""Two enabled, externally-routed servers for the same project and
|
|
environment must fail closed rather than bind the demo identity to an
|
|
arbitrary one of them."""
|
|
with self.sessions() as session, session.begin():
|
|
session.add(
|
|
VirtualServerRow(
|
|
id=uuid4(),
|
|
workspace_id=self.workspace_id,
|
|
project_id=self.project_id,
|
|
environment_id=self.environment_id,
|
|
slug="mobilityops-prod-readonly-second",
|
|
desired_hash="b" * 64,
|
|
enabled=True,
|
|
gateway_external_id="b" * 32,
|
|
)
|
|
)
|
|
with self.sessions() as session, session.begin():
|
|
with self.assertRaises(ValueError):
|
|
provision_demo_client(session)
|
|
|
|
def test_raises_when_virtual_server_has_no_external_route(self) -> None:
|
|
"""A server that exists but was never routed externally is not a valid
|
|
demo binding target."""
|
|
with self.sessions() as session, session.begin():
|
|
stored = session.get(VirtualServerRow, self.server_id)
|
|
assert stored is not None
|
|
stored.gateway_external_id = None
|
|
with self.sessions() as session, session.begin():
|
|
with self.assertRaises(ValueError):
|
|
provision_demo_client(session)
|