292 lines
12 KiB
Python
292 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from packages.hub_core.domain import (
|
|
Approval,
|
|
AuditEvent,
|
|
Connector,
|
|
Environment,
|
|
Operation,
|
|
Project,
|
|
Scope,
|
|
Workspace,
|
|
)
|
|
from packages.hub_core.errors import (
|
|
AuditMutationError,
|
|
ConcurrentUpdateError,
|
|
InvalidTransition,
|
|
MissingProjectScope,
|
|
)
|
|
from services.control_api.persistence.db import create_engine_from_url, session_factory, utc_now
|
|
from services.control_api.persistence.models import RiskClassRow, WorkspaceRow
|
|
from services.control_api.persistence.repositories import (
|
|
ApprovalRepository,
|
|
AuditEventRepository,
|
|
ConnectorRepository,
|
|
EnvironmentRepository,
|
|
OperationRepository,
|
|
ProjectRepository,
|
|
WorkspaceRepository,
|
|
)
|
|
from services.control_api.persistence.seed import seed_database
|
|
|
|
|
|
class PersistenceTestCase(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.db_path = Path(self.temp_dir.name) / "hub.db"
|
|
self.engine = create_engine_from_url(f"sqlite:///{self.db_path}")
|
|
from services.control_api.persistence.models import Base
|
|
|
|
Base.metadata.create_all(self.engine)
|
|
self.Session = session_factory(self.engine)
|
|
|
|
def tearDown(self) -> None:
|
|
self.engine.dispose()
|
|
self.temp_dir.cleanup()
|
|
|
|
@staticmethod
|
|
def workspace(workspace_id: object, slug: str) -> Workspace:
|
|
now = utc_now()
|
|
return Workspace(
|
|
id=workspace_id, slug=slug, name=slug.title(), created_at=now, updated_at=now
|
|
)
|
|
|
|
@staticmethod
|
|
def project(workspace_id: object, slug: str) -> Project:
|
|
now = utc_now()
|
|
return Project(
|
|
id=uuid4(),
|
|
workspace_id=workspace_id,
|
|
slug=slug,
|
|
name=slug.title(),
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
def test_seed_is_idempotent_and_contains_no_projects_or_credentials(self) -> None:
|
|
with self.Session.begin() as session:
|
|
first = seed_database(session)
|
|
with self.Session.begin() as session:
|
|
second = seed_database(session)
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(session.scalar(select(func.count()).select_from(WorkspaceRow)), 1)
|
|
self.assertEqual(session.scalar(select(func.count()).select_from(RiskClassRow)), 3)
|
|
self.assertFalse(
|
|
any("password" in code or "secret" in code for code in second.risk_class_codes)
|
|
)
|
|
|
|
def test_scoped_repositories_cannot_cross_workspace_or_project(self) -> None:
|
|
workspace_a, workspace_b = uuid4(), uuid4()
|
|
with self.Session.begin() as session:
|
|
WorkspaceRepository(session, Scope.workspace(workspace_a)).create(
|
|
self.workspace(workspace_a, "a")
|
|
)
|
|
WorkspaceRepository(session, Scope.workspace(workspace_b)).create(
|
|
self.workspace(workspace_b, "b")
|
|
)
|
|
project_a = ProjectRepository(session, Scope.workspace(workspace_a)).create(
|
|
self.project(workspace_a, "same")
|
|
)
|
|
ProjectRepository(session, Scope.workspace(workspace_b)).create(
|
|
self.project(workspace_b, "same")
|
|
)
|
|
self.assertEqual(
|
|
len(ProjectRepository(session, Scope.workspace(workspace_a)).list()), 1
|
|
)
|
|
self.assertEqual(
|
|
len(ProjectRepository(session, Scope.workspace(workspace_b)).list()), 1
|
|
)
|
|
self.assertEqual(
|
|
len(ProjectRepository(session, Scope.project(workspace_a, project_a.id)).list()), 1
|
|
)
|
|
with self.assertRaises(MissingProjectScope):
|
|
EnvironmentRepository(session, Scope.workspace(workspace_a))
|
|
|
|
def test_duplicate_slug_and_orphan_binding_are_rejected(self) -> None:
|
|
workspace_id = uuid4()
|
|
with self.Session.begin() as session:
|
|
WorkspaceRepository(session, Scope.workspace(workspace_id)).create(
|
|
self.workspace(workspace_id, "itworx")
|
|
)
|
|
project_repo = ProjectRepository(session, Scope.workspace(workspace_id))
|
|
project_repo.create(self.project(workspace_id, "duplicate"))
|
|
with self.assertRaises(IntegrityError):
|
|
project_repo.create(self.project(workspace_id, "duplicate"))
|
|
session.flush()
|
|
with self.Session() as session:
|
|
orphan = Environment(
|
|
id=uuid4(),
|
|
workspace_id=workspace_id,
|
|
project_id=uuid4(),
|
|
name="dev",
|
|
classification="development",
|
|
created_at=utc_now(),
|
|
updated_at=utc_now(),
|
|
)
|
|
with self.assertRaises(IntegrityError):
|
|
EnvironmentRepository(
|
|
session, Scope.project(workspace_id, orphan.project_id)
|
|
).create(orphan)
|
|
session.commit()
|
|
session.rollback()
|
|
|
|
def test_restart_persists_project_and_connector_health_transitions(self) -> None:
|
|
workspace_id, project_id = uuid4(), uuid4()
|
|
with self.Session.begin() as session:
|
|
WorkspaceRepository(session, Scope.workspace(workspace_id)).create(
|
|
self.workspace(workspace_id, "restart")
|
|
)
|
|
project_repo = ProjectRepository(session, Scope.workspace(workspace_id))
|
|
project = self.project(workspace_id, "persisted")
|
|
project.id = project_id
|
|
project_repo.create(project)
|
|
scope = Scope.project(workspace_id, project_id)
|
|
environment = Environment(
|
|
id=uuid4(),
|
|
workspace_id=workspace_id,
|
|
project_id=project_id,
|
|
name="dev",
|
|
classification="development",
|
|
created_at=utc_now(),
|
|
updated_at=utc_now(),
|
|
)
|
|
EnvironmentRepository(session, scope).create(environment)
|
|
connector = Connector(
|
|
id=uuid4(),
|
|
workspace_id=workspace_id,
|
|
project_id=project_id,
|
|
type="test",
|
|
slug="connector",
|
|
created_at=utc_now(),
|
|
updated_at=utc_now(),
|
|
)
|
|
created = ConnectorRepository(session, scope).create(connector)
|
|
transitioned = ConnectorRepository(session, scope).transition_health(
|
|
created.id, "checking", expected_version=1
|
|
)
|
|
self.assertEqual(transitioned.health_state, "checking")
|
|
with self.assertRaises(ConcurrentUpdateError):
|
|
ConnectorRepository(session, scope).transition_health(
|
|
created.id, "healthy", expected_version=1
|
|
)
|
|
self.engine.dispose()
|
|
self.engine = create_engine_from_url(f"sqlite:///{self.db_path}")
|
|
self.Session = session_factory(self.engine)
|
|
with self.Session() as session:
|
|
self.assertEqual(
|
|
len(ProjectRepository(session, Scope.project(workspace_id, project_id)).list()), 1
|
|
)
|
|
|
|
def test_operations_and_approvals_reject_invalid_transitions(self) -> None:
|
|
workspace_id, project_id, environment_id = uuid4(), uuid4(), uuid4()
|
|
now = utc_now()
|
|
with self.Session.begin() as session:
|
|
WorkspaceRepository(session, Scope.workspace(workspace_id)).create(
|
|
self.workspace(workspace_id, "ops")
|
|
)
|
|
project = self.project(workspace_id, "ops-project")
|
|
project.id = project_id
|
|
ProjectRepository(session, Scope.workspace(workspace_id)).create(project)
|
|
scope = Scope.project(workspace_id, project_id)
|
|
EnvironmentRepository(session, scope).create(
|
|
Environment(
|
|
id=environment_id,
|
|
workspace_id=workspace_id,
|
|
project_id=project_id,
|
|
name="dev",
|
|
classification="development",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
operation_repo = OperationRepository(session, scope)
|
|
operation = operation_repo.create(
|
|
Operation(
|
|
id=uuid4(),
|
|
workspace_id=workspace_id,
|
|
project_id=project_id,
|
|
environment_id=environment_id,
|
|
operation_type="future-write",
|
|
target_ref="target",
|
|
plan_hash="sha256:plan",
|
|
requested_by="subject",
|
|
created_at=now,
|
|
updated_at=now,
|
|
expires_at=now + timedelta(minutes=5),
|
|
)
|
|
)
|
|
planned = operation_repo.transition_state(operation.id, "planned", expected_version=1)
|
|
self.assertEqual(planned.state, "planned")
|
|
with self.assertRaises(InvalidTransition):
|
|
operation_repo.transition_state(operation.id, "succeeded", expected_version=2)
|
|
approval_required = operation_repo.transition_state(
|
|
operation.id, "approval_required", expected_version=2
|
|
)
|
|
self.assertEqual(approval_required.state, "approval_required")
|
|
approval = ApprovalRepository(session, scope).create(
|
|
Approval(
|
|
id=uuid4(),
|
|
workspace_id=workspace_id,
|
|
project_id=project_id,
|
|
operation_id=operation.id,
|
|
plan_hash=operation.plan_hash,
|
|
actor="approver",
|
|
expires_at=now + timedelta(minutes=5),
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
approved = ApprovalRepository(session, scope).transition_state(
|
|
approval.id, "approved", expected_version=1
|
|
)
|
|
self.assertEqual(approved.state, "approved")
|
|
with self.assertRaises(InvalidTransition):
|
|
ApprovalRepository(session, scope).transition_state(
|
|
approval.id, "rejected", expected_version=2
|
|
)
|
|
|
|
def test_audit_is_append_only_and_sanitized(self) -> None:
|
|
workspace_id = uuid4()
|
|
now = datetime.now(UTC)
|
|
with self.Session.begin() as session:
|
|
WorkspaceRepository(session, Scope.workspace(workspace_id)).create(
|
|
self.workspace(workspace_id, "audit")
|
|
)
|
|
repository = AuditEventRepository(session, Scope.workspace(workspace_id))
|
|
event = repository.create(
|
|
AuditEvent(
|
|
occurred_at=now,
|
|
trace_id="trace-1",
|
|
workspace_id=workspace_id,
|
|
event_type="test",
|
|
sanitized_arguments={"containerRef": "safe"},
|
|
outcome="success",
|
|
)
|
|
)
|
|
self.assertEqual(repository.list()[0].id, event.id)
|
|
self.assertFalse(hasattr(repository, "update"))
|
|
with self.assertRaises(AuditMutationError):
|
|
repository.create(
|
|
AuditEvent(
|
|
occurred_at=now,
|
|
trace_id="trace-2",
|
|
workspace_id=workspace_id,
|
|
event_type="test",
|
|
sanitized_arguments={"password": "redacted"},
|
|
outcome="denied",
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|