0179355fe7
The cleanup I added dropped every tenant schema in a single transaction, and a hundred runs' worth of them exhausts the lock table: the session teardown failed with 'out of shared memory ... max_locks_per_transaction', which reads like a PostgreSQL defect and is not one. It is the same constraint the 0005 rollback documents and I wrote there before hitting it here. Committing after each drop releases the locks, and the teardown now leaves zero schemas behind, so a run does not make the next one slower.
1731 lines
74 KiB
Python
1731 lines
74 KiB
Python
"""Tests for the PostgreSQL adapter.
|
|
|
|
Two layers, because the two have very different verification status.
|
|
|
|
**Mapping and contract tests run everywhere.** The row mappers, the
|
|
role-assignment rules, the unique-violation mapping, and the row-level-security
|
|
identity contract are pure logic, so they are asserted directly. These are the
|
|
places where a silent misreading of the control-plane schema would produce a
|
|
wrong tenant scope rather than an obvious crash, which is why they are covered
|
|
in detail.
|
|
|
|
**Integration tests require a real database and skip without one.** They are
|
|
selected by ``UMBRATRANS_TEST_DSN`` and must point at a *scratch* database, never
|
|
at ``umbracontrol``. The fixture refuses a DSN whose database name contains
|
|
``umbracontrol``.
|
|
|
|
The integration layer has already earned its keep: running the forward migration
|
|
against the real control-plane schema surfaced an ``ADD CONSTRAINT`` that failed
|
|
because the constraint already existed, and the adapter was setting a
|
|
row-level-security GUC whose name did not match the one the policy helpers read.
|
|
Neither is visible without a real database.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import inspect
|
|
import os
|
|
import re
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from psycopg.rows import dict_row
|
|
from psycopg_pool import ConnectionPool
|
|
|
|
from umbratrans_web.config import Environment, Settings
|
|
from umbratrans_web.dev_repository import InMemoryAccountRepository, JsonDevelopmentRepository
|
|
from umbratrans_web.errors import (
|
|
ConflictError,
|
|
EmailNotVerifiedError,
|
|
ForbiddenError,
|
|
InvalidCredentialsError,
|
|
RepositoryError,
|
|
)
|
|
from umbratrans_web.mailer import CapturingMailer
|
|
from umbratrans_web.models import (
|
|
AccountStatus,
|
|
OrganizationStatus,
|
|
Principal,
|
|
ProjectStatus,
|
|
RoleName,
|
|
User,
|
|
)
|
|
from umbratrans_web.postgres_repository import (
|
|
RLS_USER_FUNCTION,
|
|
RLS_USER_SETTING,
|
|
STATEMENT_TIMEOUT_MS,
|
|
PostgresAccountRepository,
|
|
_assignment_shape,
|
|
_audit_from_row,
|
|
_organization_from_row,
|
|
_session_from_row,
|
|
_unique_conflict,
|
|
_user_from_row,
|
|
)
|
|
from umbratrans_web.repository import AccountRepository
|
|
from umbratrans_web.service import AccountService, OperatorContext
|
|
|
|
from .conftest import provision_invite, provision_local, random_password, verification_token
|
|
|
|
TEST_DSN_ENV = "UMBRATRANS_TEST_DSN"
|
|
|
|
|
|
def _row(**overrides: Any) -> dict[str, Any]:
|
|
"""One ``users`` row as psycopg would return it with ``dict_row``."""
|
|
|
|
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
base: dict[str, Any] = {
|
|
"id": uuid.uuid4(),
|
|
"username": "Umbraadmin",
|
|
"display_name": "Umbra Administrator",
|
|
"email": "umbraadmin@example.test",
|
|
"status": "active",
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Protocol conformance
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def _unimplemented_protocol_methods(implementation: type) -> list[str]:
|
|
"""Return the protocol methods ``implementation`` does not actually provide.
|
|
|
|
Walking ``vars()`` rather than using ``hasattr`` is the whole point. Every
|
|
implementation inherits from the protocol, so ``hasattr`` is true for all of
|
|
them no matter what is written: it passed while five methods were missing and
|
|
email verification and login were both non-functional against a real
|
|
database. A method counts as provided only if some class below the protocol
|
|
defines it.
|
|
"""
|
|
|
|
provided: set[str] = set()
|
|
for klass in implementation.__mro__:
|
|
if klass is AccountRepository:
|
|
break
|
|
provided.update(name for name, value in vars(klass).items() if callable(value))
|
|
declared = {name for name, value in vars(AccountRepository).items() if callable(value)}
|
|
return sorted(declared - provided)
|
|
|
|
|
|
def test_the_adapter_satisfies_the_repository_protocol() -> None:
|
|
"""Every declared method must be implemented, not merely inherited.
|
|
|
|
The protocol raises :class:`NotImplementedError` for anything it does not
|
|
implement, so an inherited stub is a loud failure rather than a silent
|
|
``None``. This test is what keeps the two consistent: a method added to the
|
|
protocol without an implementation fails here rather than in production.
|
|
"""
|
|
|
|
missing = _unimplemented_protocol_methods(PostgresAccountRepository)
|
|
assert missing == [], f"PostgresAccountRepository does not implement: {missing}"
|
|
|
|
|
|
def test_every_repository_implements_every_protocol_method() -> None:
|
|
"""The same check for the development repositories.
|
|
|
|
The JSON repository inherits from the in-memory one, so this also proves the
|
|
inheritance actually carries the methods rather than the protocol's stubs.
|
|
"""
|
|
|
|
for implementation in (InMemoryAccountRepository, JsonDevelopmentRepository):
|
|
missing = _unimplemented_protocol_methods(implementation)
|
|
assert missing == [], f"{implementation.__name__} does not implement: {missing}"
|
|
|
|
|
|
def test_every_repository_declares_the_same_signature_as_the_protocol() -> None:
|
|
"""Presence is not enough; the call has to be the same call.
|
|
|
|
The previous check asked only whether a method existed, which is how the
|
|
PostgreSQL adapter shipped a ``get_live_verification_token`` that took a
|
|
``purpose`` but not the ``now`` its in-memory counterpart required. Every test
|
|
that did not open a real database passed, and the first call against a real
|
|
one raised ``TypeError`` and returned a 500.
|
|
|
|
A protocol that names a parameter every implementation must accept is the only
|
|
thing that catches that before a deployment does.
|
|
"""
|
|
|
|
from umbratrans_web.dev_repository import InMemoryAccountRepository, JsonDevelopmentRepository
|
|
|
|
for implementation in (InMemoryAccountRepository, JsonDevelopmentRepository, PostgresAccountRepository):
|
|
for name, declared in vars(AccountRepository).items():
|
|
if not callable(declared) or name.startswith("_"):
|
|
continue
|
|
actual = getattr(implementation, name, None)
|
|
assert actual is not None, f"{implementation.__name__}.{name} is missing"
|
|
expected = inspect.signature(declared)
|
|
found = inspect.signature(actual)
|
|
assert list(found.parameters) == list(expected.parameters), (
|
|
f"{implementation.__name__}.{name}{found} does not match the protocol's "
|
|
f"{expected}"
|
|
)
|
|
|
|
|
|
def test_an_inherited_stub_raises_rather_than_returning_none() -> None:
|
|
"""A repository missing a method must fail loudly.
|
|
|
|
The failure this guards against is subtle: a stub returning ``None`` is
|
|
indistinguishable from "no such row", so a missing method presents as an
|
|
empty database instead of an error.
|
|
"""
|
|
|
|
class Incomplete(AccountRepository): # type: ignore[misc]
|
|
pass
|
|
|
|
with pytest.raises(NotImplementedError):
|
|
Incomplete().get_user_by_email("someone@example.test")
|
|
|
|
|
|
def test_the_adapter_imports_without_opening_a_connection() -> None:
|
|
"""Importing the module must not touch the network.
|
|
|
|
``psycopg_pool`` is imported at module scope, so a missing database is a
|
|
runtime failure at construction, never an import-time surprise.
|
|
"""
|
|
|
|
assert PostgresAccountRepository.__doc__ is not None
|
|
assert STATEMENT_TIMEOUT_MS > 0
|
|
|
|
|
|
def test_the_identity_setting_matches_what_the_policies_read() -> None:
|
|
"""The adapter and the policy helpers must agree, by construction.
|
|
|
|
``app.current_user_id()`` is defined in the control plane as
|
|
``current_setting('app.user_id', true)``. A similarly named GUC would leave
|
|
every helper resolving to NULL, so every policy would deny and every query
|
|
would return zero rows - silently. This was wrong once and the integration
|
|
run is what found it, so it is pinned here where it is cheap to check.
|
|
"""
|
|
|
|
assert RLS_USER_SETTING == "app.user_id"
|
|
assert RLS_USER_FUNCTION == "app.current_user_id"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Role and organization resolution
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_a_platform_admin_is_assigned_at_platform_scope() -> None:
|
|
"""A platform assignment must carry no organization.
|
|
|
|
This is what the ``user_roles`` check constraint requires, and getting it
|
|
wrong is a database error on insert rather than a silent bug.
|
|
"""
|
|
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
username="Umbraadmin",
|
|
username_key="umbraadmin",
|
|
display_name="Umbra Administrator",
|
|
email="umbraadmin@example.test",
|
|
role=RoleName.PLATFORM_ADMIN,
|
|
status=AccountStatus.ACTIVE,
|
|
)
|
|
assert _assignment_shape(user) == ("platform", None)
|
|
|
|
|
|
def test_an_organization_scoped_role_requires_an_organization() -> None:
|
|
"""An organization assignment without one is refused before the database.
|
|
|
|
The same combination is rejected by the ``user_roles`` check constraint;
|
|
refusing here names the real problem instead of surfacing a constraint
|
|
violation.
|
|
"""
|
|
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
username="Orphan.Member",
|
|
username_key="orphan.member",
|
|
display_name="Orphan",
|
|
email="orphan@example.test",
|
|
role=RoleName.MEMBER,
|
|
status=AccountStatus.ACTIVE,
|
|
)
|
|
with pytest.raises(RepositoryError) as error:
|
|
_assignment_shape(user)
|
|
assert error.value.code == "organization_required"
|
|
|
|
|
|
def test_a_member_carries_its_organization() -> None:
|
|
organization_id = uuid.uuid4()
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
username="Member.One",
|
|
username_key="member.one",
|
|
display_name="Member One",
|
|
email="member.one@example.test",
|
|
role=RoleName.MEMBER,
|
|
status=AccountStatus.ACTIVE,
|
|
organization_id=organization_id,
|
|
)
|
|
assert _assignment_shape(user) == ("organization", organization_id)
|
|
|
|
|
|
def test_an_organization_admin_is_also_organization_scoped() -> None:
|
|
organization_id = uuid.uuid4()
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
username="Org.Admin",
|
|
username_key="org.admin",
|
|
display_name="Org Admin",
|
|
email="org.admin@example.test",
|
|
role=RoleName.ORGANIZATION_ADMIN,
|
|
status=AccountStatus.ACTIVE,
|
|
organization_id=organization_id,
|
|
)
|
|
assert _assignment_shape(user) == ("organization", organization_id)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# User row mapping
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_a_user_row_maps_with_a_resolved_assignment() -> None:
|
|
row = _row()
|
|
organization_id = uuid.uuid4()
|
|
user = _user_from_row(row, role=RoleName.MEMBER, organization_id=organization_id)
|
|
|
|
assert user.id == row["id"]
|
|
assert user.username == "Umbraadmin"
|
|
assert user.username_key == "umbraadmin"
|
|
assert user.role is RoleName.MEMBER
|
|
assert user.status is AccountStatus.ACTIVE
|
|
assert user.organization_id == organization_id
|
|
assert user.email == "umbraadmin@example.test"
|
|
|
|
|
|
def test_a_user_row_falls_back_to_the_username_when_the_name_is_null() -> None:
|
|
"""users.display_name is nullable in the control plane."""
|
|
|
|
user = _user_from_row(_row(display_name=None))
|
|
assert user.display_name == "Umbraadmin"
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["active", "suspended", "deactivated"])
|
|
def test_modelled_account_statuses_are_accepted(status: str) -> None:
|
|
assert _user_from_row(_row(status=status)).status is AccountStatus(status)
|
|
|
|
|
|
def test_an_invited_account_maps_and_is_not_yet_verified() -> None:
|
|
"""``invited`` is the control plane's own state for an unconfirmed address.
|
|
|
|
It is modelled rather than refused, because an account that exists but has
|
|
not proved it can receive mail must be distinguishable from one that does
|
|
not exist yet.
|
|
"""
|
|
|
|
user = _user_from_row(_row(status="invited"))
|
|
assert user.status is AccountStatus.INVITED
|
|
assert user.is_email_verified is False
|
|
assert _user_from_row(_row(status="active")).is_email_verified is True
|
|
|
|
|
|
def test_an_account_requiring_mfa_is_refused() -> None:
|
|
"""The control plane can require a second factor this phase cannot do.
|
|
|
|
Authenticating such an account anyway would grant access to someone who was
|
|
supposed to present a second factor, so the read fails closed.
|
|
"""
|
|
|
|
with pytest.raises(RepositoryError) as error:
|
|
_user_from_row(_row(mfa_required=True))
|
|
assert error.value.code == "mfa_required_unsupported"
|
|
|
|
|
|
def test_a_profile_row_without_the_mfa_column_is_accepted() -> None:
|
|
"""Tolerates a row shape that predates the column being selected."""
|
|
|
|
assert _user_from_row(_row()).status is AccountStatus.ACTIVE
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["locked", "paused", "ACTIVE"])
|
|
def test_an_unmodelled_status_is_refused_rather_than_coerced(status: str) -> None:
|
|
"""``invited`` and ``locked`` exist in the control plane but not here.
|
|
|
|
Mapping one of them onto ``active`` would grant access to an account that
|
|
the control plane considers not yet usable, so the read fails closed.
|
|
"""
|
|
|
|
with pytest.raises(RepositoryError) as error:
|
|
_user_from_row(_row(status=status))
|
|
assert error.value.code == "unmapped_account_status"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Organization row mapping
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_an_organization_row_maps() -> None:
|
|
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
organization = _organization_from_row(
|
|
{
|
|
"id": uuid.uuid4(),
|
|
"name": "Northstar Media",
|
|
"slug": "northstar-media",
|
|
"status": "active",
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
assert organization.name == "Northstar Media"
|
|
assert organization.slug == "northstar-media"
|
|
assert organization.status is OrganizationStatus.ACTIVE
|
|
|
|
|
|
def test_an_unmodelled_organization_status_is_refused() -> None:
|
|
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
with pytest.raises(ValueError):
|
|
_organization_from_row(
|
|
{
|
|
"id": uuid.uuid4(),
|
|
"name": "Odd",
|
|
"slug": "odd",
|
|
"status": "deleted",
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Session row mapping
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_a_session_row_maps_expires_at_to_the_absolute_deadline() -> None:
|
|
"""``sessions.expires_at`` predates this application.
|
|
|
|
The adapter keeps the two equal on write, so anything else reading
|
|
``expires_at`` sees the same deadline.
|
|
"""
|
|
|
|
created = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
session = _session_from_row(
|
|
{
|
|
"id": uuid.uuid4(),
|
|
"user_id": uuid.uuid4(),
|
|
"token_hash": "a" * 64,
|
|
"created_at": created,
|
|
"last_seen_at": created + timedelta(minutes=1),
|
|
"idle_deadline_at": created + timedelta(minutes=30),
|
|
"expires_at": created + timedelta(hours=12),
|
|
"revoked_at": None,
|
|
"mfa_verified_at": None,
|
|
}
|
|
)
|
|
assert session.absolute_deadline_at == created + timedelta(hours=12)
|
|
assert session.idle_deadline_at == created + timedelta(minutes=30)
|
|
assert session.is_revoked() is False
|
|
# A hash must not surface through repr.
|
|
assert "a" * 64 not in repr(session)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Audit row mapping
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_an_audit_row_reports_the_store_identifier() -> None:
|
|
"""``audit_events.id`` is a bigint identity owned by the database.
|
|
|
|
The application mints a UUID for correlation, which is preserved in the
|
|
detail, but the API must report the identifier the store will actually find.
|
|
"""
|
|
|
|
event = _audit_from_row(
|
|
{
|
|
"id": 4211,
|
|
"actor_user_id": uuid.uuid4(),
|
|
"organization_id": None,
|
|
"action": "account.created",
|
|
"target_type": "user",
|
|
"target_id": str(uuid.uuid4()),
|
|
"request_id": "req-1",
|
|
"metadata": {"role": "member", "application_event_id": str(uuid.uuid4())},
|
|
"created_at": datetime(2026, 1, 1, tzinfo=timezone.utc),
|
|
},
|
|
original=None,
|
|
)
|
|
assert event.id == "4211"
|
|
assert event.action == "account.created"
|
|
assert event.target_user_id is not None
|
|
# The correlation UUID is retained for the caller but is not the row key.
|
|
assert "application_event_id" not in event.detail
|
|
assert event.detail["role"] == "member"
|
|
|
|
|
|
def test_a_non_user_target_maps_to_no_user() -> None:
|
|
event = _audit_from_row(
|
|
{
|
|
"id": 7,
|
|
"actor_user_id": None,
|
|
"organization_id": uuid.uuid4(),
|
|
"action": "organization.created",
|
|
"target_type": "organization",
|
|
"target_id": str(uuid.uuid4()),
|
|
"request_id": None,
|
|
"metadata": {},
|
|
"created_at": datetime(2026, 1, 1, tzinfo=timezone.utc),
|
|
},
|
|
original=None,
|
|
)
|
|
assert event.target_user_id is None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Conflict mapping
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
class _FakeDiagnostic:
|
|
def __init__(self, constraint_name: str) -> None:
|
|
self.constraint_name = constraint_name
|
|
|
|
|
|
class _FakeUniqueViolation(Exception):
|
|
def __init__(self, constraint_name: str) -> None:
|
|
super().__init__("duplicate key value violates unique constraint")
|
|
self.diag = _FakeDiagnostic(constraint_name)
|
|
|
|
|
|
def test_a_username_violation_maps_to_username_taken() -> None:
|
|
error = _unique_conflict(_FakeUniqueViolation("users_username_key"), "Umbraadmin")
|
|
assert isinstance(error, ConflictError)
|
|
assert error.code == "username_taken"
|
|
|
|
|
|
def test_an_email_violation_maps_to_email_taken() -> None:
|
|
error = _unique_conflict(_FakeUniqueViolation("users_email_key"), "Umbraadmin")
|
|
assert error.code == "email_taken"
|
|
|
|
|
|
def test_an_unknown_constraint_falls_back_to_the_username_rule() -> None:
|
|
error = _unique_conflict(_FakeUniqueViolation("something_else"), "Umbraadmin")
|
|
assert error.code == "username_taken"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Adapter source hygiene
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_the_adapter_never_interpolates_a_value_into_sql() -> None:
|
|
"""Every value reaches PostgreSQL as a bound parameter.
|
|
|
|
The statements are assembled from fixed fragments only; a value is never
|
|
formatted into the string. This asserts the absence of the common mistake
|
|
rather than trying to prove the presence of the correct pattern.
|
|
"""
|
|
|
|
import inspect
|
|
|
|
source = inspect.getsource(PostgresAccountRepository)
|
|
for pattern in (
|
|
f"{{{{{'user'}.username}}}}",
|
|
f"{{{{{'user'}.email}}}}",
|
|
f"{{{{{'event'}.action}}}}",
|
|
f"{{{{{'organization'}.slug}}}}",
|
|
):
|
|
assert pattern not in source, pattern
|
|
|
|
|
|
def test_every_user_roles_query_distinguishes_tenancy_from_project_membership() -> None:
|
|
"""No query may read ``user_roles`` without saying which kind of row it wants.
|
|
|
|
``user_roles`` holds two unrelated statements: a tenant row saying which
|
|
organization an account belongs to, and a project row saying who is working
|
|
on what. A project row also carries an organization, so nothing but the scope
|
|
tells them apart.
|
|
|
|
This exists because the omission was a live lockout. Role resolution counted
|
|
every active row, so the moment an account was assigned to a project it held
|
|
two rows and became unresolvable - unable to sign in, and locked out of the
|
|
page the assignment was meant to appear on. Four further queries were
|
|
scope-blind in the same way, and three of them happened to be harmless only
|
|
because the assignment role is named ``viewer`` and so never matched a
|
|
tenant role name. That is luck, not a property.
|
|
|
|
The check is over the adapter's source rather than its behaviour because the
|
|
behaviour is correct today by coincidence in three places, and a behavioural
|
|
test cannot tell a coincidence from a rule. Naming the permitted
|
|
project-scoped statements keeps the ban narrow enough to be worth having:
|
|
the queries that *should* filter on ``scope = 'project'`` are exactly the ones
|
|
implementing assignment, and they are listed here by intent.
|
|
"""
|
|
|
|
source = inspect.getsource(PostgresAccountRepository) + inspect.getsource(
|
|
sys.modules[PostgresAccountRepository.__module__]
|
|
)
|
|
# The adapter builds long SQL from adjacent f-strings, one clause per line, so
|
|
# the source has to be joined up before a statement can be read. Without this
|
|
# a query is cut in half at its first line break and its predicate - which is
|
|
# always in the last clause - lands in a different "statement".
|
|
source = re.sub(r'"\s*\n\s*(?:f|r|rf|fr)?"', "", source)
|
|
source = re.sub(r'""', "", source)
|
|
|
|
# Only real SQL, keyed on the statement verb. Matching on the table name alone
|
|
# would pick up the docstrings, several of which explain this very rule.
|
|
statements = re.findall(r'(?is)(FROM|INTO|UPDATE)\s+public\.user_roles(?:\s+ur\b)?(.*?)(?=")', source)
|
|
assert statements, "no user_roles queries found; this test is not looking at what it thinks"
|
|
|
|
# The queries that are *supposed* to be project-scoped: the ones that read or
|
|
# write an assignment.
|
|
project_scoped = (
|
|
"list_project_member_ids",
|
|
"assign_user_to_project",
|
|
"unassign_user_from_project",
|
|
)
|
|
for name in project_scoped:
|
|
body = inspect.getsource(getattr(PostgresAccountRepository, name))
|
|
assert "scope = 'project'" in body, f"{name} must select project rows explicitly"
|
|
|
|
for verb, rest in statements:
|
|
# The verb is part of the statement; the capture group starts after the
|
|
# table name, so it has to be put back or an INSERT is not recognisable
|
|
# as one.
|
|
body = f"{verb} public.user_roles{rest}".replace("{TENANT_ROW_PREDICATE}", "scope <> 'project'")
|
|
if re.search(r"scope\s*=\s*'project'", body):
|
|
continue # deliberately reading or writing project rows
|
|
if re.search(r"scope\s*<>\s*'project'", body):
|
|
continue # deliberately excluding them
|
|
if re.search(r"(?i)\bINSERT\b", body) or verb.upper() == "INTO":
|
|
if not re.search(r"(?i)\bVALUES\b", body):
|
|
raise AssertionError(f"user_roles insert without VALUES: {body[:200]!r}")
|
|
# A row this statement creates. The scope was computed a line earlier
|
|
# and named in the column list, so there is nothing to disambiguate -
|
|
# reading it back is what would need the predicate.
|
|
continue
|
|
if verb.upper() == "UPDATE" and re.search(r"(?is)WHERE\s+u?r?\.?id\s*=\s*%s\s*$", body.strip()):
|
|
# The one narrow exemption: a write keyed on the row's own primary
|
|
# key. It names exactly one row, and the row it names was chosen
|
|
# deliberately by a preceding SELECT - in _sync_assignment, the tenant
|
|
# row it just read. Such a statement cannot reach a project row, so
|
|
# there is nothing for a scope predicate to exclude.
|
|
#
|
|
# Matched as the *whole* WHERE clause rather than a substring, so
|
|
# `WHERE id = %s OR scope = 'project'` is not quietly admitted.
|
|
continue
|
|
if "scope" in body:
|
|
# Scope appears but filters nothing: the column is in a projection or
|
|
# a comment, which is the shape of the bug this test exists to catch.
|
|
raise AssertionError(f"user_roles query mentions scope without filtering on it: {body[:200]!r}")
|
|
raise AssertionError(
|
|
"this user_roles query does not mention scope, so it cannot tell a tenant row from a "
|
|
f"project row: {body[:200]!r}"
|
|
)
|
|
|
|
|
|
def test_the_adapter_issues_no_destructive_statement() -> None:
|
|
"""Insert and update are the adapter's job; nothing that destroys live data is.
|
|
|
|
``INSERT INTO`` is expected here, unlike in the migration file. What must
|
|
never appear is anything that drops, truncates, deletes, or alters schema:
|
|
those belong in a reviewed migration, not in request handling.
|
|
|
|
Two deletes are permitted, each by name and each pinned separately.
|
|
Purging verification tokens that are already spent or expired, which the
|
|
storage contract requires and which cannot reach a live row. Removing one
|
|
account's assignment to one project, which is the operation's purpose and is
|
|
bounded by both a user and a project. Both are excluded here and both have a
|
|
test asserting their predicate, so widening either cannot quietly become data
|
|
loss.
|
|
"""
|
|
|
|
import inspect
|
|
|
|
source = inspect.getsource(PostgresAccountRepository)
|
|
for permitted in (
|
|
inspect.getsource(PostgresAccountRepository.delete_expired_verification_tokens),
|
|
inspect.getsource(PostgresAccountRepository.unassign_user_from_project),
|
|
):
|
|
source = source.replace(permitted, "")
|
|
for statement in (
|
|
"DROP ",
|
|
"TRUNCATE",
|
|
"DELETE FROM",
|
|
"ALTER TABLE",
|
|
"CREATE TABLE",
|
|
"GRANT ",
|
|
"REVOKE ",
|
|
):
|
|
assert statement not in source, statement
|
|
|
|
|
|
def test_the_token_purge_cannot_reach_a_live_token() -> None:
|
|
"""The one permitted delete must stay bounded to rows that are already dead.
|
|
|
|
``bootstrap_tokens`` holds a live row for exactly one reason: an unconsumed,
|
|
unexpired link that someone may still click. The purge is allowed to remove
|
|
spent and expired rows and nothing else, so both conditions are pinned here
|
|
rather than trusted to review.
|
|
"""
|
|
|
|
import inspect
|
|
|
|
purge = inspect.getsource(PostgresAccountRepository.delete_expired_verification_tokens)
|
|
assert "DELETE FROM public.bootstrap_tokens" in purge
|
|
assert "consumed_at IS NOT NULL" in purge
|
|
assert "expires_at <= %s" in purge
|
|
# An unconsumed, unexpired row must not match: that is the whole point.
|
|
assert "consumed_at IS NULL" not in purge
|
|
assert "expires_at > " not in purge
|
|
# And it must not be able to reach any other table.
|
|
assert purge.count("DELETE FROM") == 1
|
|
|
|
|
|
def test_the_adapter_never_selects_a_hash_into_a_user_row() -> None:
|
|
"""A ``users`` read must not drag a credential hash along with the profile.
|
|
|
|
Checked against the actual column list of every statement that reads
|
|
``public.users``, rather than a blanket scan of the module, because the
|
|
credential reads legitimately do select ``password_hash``.
|
|
"""
|
|
|
|
import inspect
|
|
import re
|
|
|
|
source = inspect.getsource(PostgresAccountRepository)
|
|
statements = re.findall(r"SELECT\s+(.*?)\s+FROM\s+public\.users", source, re.DOTALL)
|
|
assert statements, "expected at least one statement reading public.users"
|
|
for columns in statements:
|
|
for protected in ("password_hash", "token_hash"):
|
|
assert protected not in columns, (protected, columns)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Integration: requires a real scratch database
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def _integration_dsn() -> str | None:
|
|
return os.environ.get(TEST_DSN_ENV) or None
|
|
|
|
|
|
def _destructive_run_acknowledged() -> bool:
|
|
return os.environ.get(TEST_ACK_ENV, "").strip() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
#: Must accompany the DSN, and says so in the variable name.
|
|
#:
|
|
#: Setting a DSN alone is not deliberate enough. This suite empties the tables it
|
|
#: touches, it cannot do so selectively, and an empty database gives no way to
|
|
#: tell a scratch one from a live deployment that has not been set up yet. The
|
|
#: emptiness guard below catches a populated database; this catches the empty one,
|
|
#: by requiring the operator to say in as many words that the target is disposable.
|
|
TEST_ACK_ENV = "UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE"
|
|
|
|
integration = pytest.mark.skipif(
|
|
_integration_dsn() is None or not _destructive_run_acknowledged(),
|
|
reason=(
|
|
f"{TEST_DSN_ENV} is unset, or {TEST_ACK_ENV} is not 1. This suite empties the tables it "
|
|
"touches and cannot do so selectively, because audit_events is append-only. Set both, and "
|
|
f"mean it: {TEST_DSN_ENV}=postgresql:///scratch?... {TEST_ACK_ENV}=1. Never point this at a "
|
|
"database that holds anything you want to keep, and never at umbracontrol."
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def require_empty_database():
|
|
"""Refuse to run against a database that holds anything.
|
|
|
|
This suite empties ``users``, ``organizations`` and ``audit_events`` when it
|
|
finishes, and that is unavoidable: ``audit_events`` is append-only, with a
|
|
trigger that raises on delete, so there is no way to remove only the rows
|
|
this run created. Truncating is the only option, and truncating is
|
|
indiscriminate by definition.
|
|
|
|
An earlier version tried to be careful instead, deleting only rows absent
|
|
from a pre-run snapshot. It could not work: the audit trigger refuses the
|
|
delete. And the version before that, which truncated without checking,
|
|
destroyed a live deployment's accounts and organization. The check that had
|
|
been passing only asserted the suite's own rows were gone, which is also true
|
|
of a full database, and it had only ever been run against one I had emptied
|
|
by hand.
|
|
|
|
So the safety is not in cleaning up carefully. It is in refusing to start.
|
|
A database with accounts in it is not a test database, whatever its name, and
|
|
the only safe thing to do with one is decline.
|
|
|
|
Since migration 0005 the check also looks inside the tenant schemas, and the
|
|
teardown drops them. That is not tidiness: a project's rows now live in a
|
|
schema per organization, so a database that holds nothing in ``public`` can
|
|
still hold somebody's books. Checking only the platform tables would have made
|
|
this guard pass on a database it would then have emptied.
|
|
"""
|
|
|
|
dsn = _integration_dsn()
|
|
if dsn is None or not _destructive_run_acknowledged() or "umbracontrol" in dsn:
|
|
return
|
|
counts = _occupancy(dsn)
|
|
populated = {name: count for name, count in counts.items() if count}
|
|
if populated:
|
|
detail = ", ".join(f"{name}={count}" for name, count in sorted(populated.items()))
|
|
pytest.skip(
|
|
"refusing to run the adapter suite against a database that already holds data "
|
|
f"({detail}). This suite empties those tables when it finishes and cannot do so "
|
|
"selectively, because audit_events is append-only. Use a scratch database."
|
|
)
|
|
yield
|
|
_drop_tenant_schemas(dsn)
|
|
|
|
|
|
#: The tables the suite empties, and the per-tenant tables it has to look inside.
|
|
PLATFORM_TABLES = ("users", "organizations", "audit_events", "user_roles", "sessions")
|
|
TENANT_TABLES = ("projects", "project_authors", "project_publishers", "invitations")
|
|
|
|
|
|
def _occupancy(dsn: str) -> dict[str, int]:
|
|
"""Row counts for every table this suite could destroy, platform and tenant."""
|
|
|
|
counts: dict[str, int] = {}
|
|
pool = ConnectionPool(dsn, min_size=1, max_size=2, open=True, kwargs={"row_factory": dict_row})
|
|
try:
|
|
with pool.connection() as connection:
|
|
for table in PLATFORM_TABLES:
|
|
row = connection.execute(f"SELECT count(*) AS total FROM public.{table}").fetchone()
|
|
counts[table] = int(row["total"])
|
|
schemas = connection.execute(
|
|
"SELECT nspname FROM pg_namespace WHERE nspname LIKE 'tenant\\_%' ORDER BY nspname"
|
|
).fetchall()
|
|
for schema in schemas:
|
|
for table in TENANT_TABLES:
|
|
row = connection.execute(
|
|
f'SELECT count(*) AS total FROM "{schema["nspname"]}"."{table}"'
|
|
).fetchone()
|
|
counts[f"{schema['nspname']}.{table}"] = int(row["total"])
|
|
finally:
|
|
pool.close()
|
|
return counts
|
|
|
|
|
|
def _drop_tenant_schemas(dsn: str) -> None:
|
|
"""Remove every tenant schema, so a run leaves nothing behind for the next one.
|
|
|
|
`TRUNCATE ... CASCADE` empties the tenant *tables* - each of the four has a
|
|
foreign key to `public.users`, so truncating users takes them with it - but the
|
|
schemas themselves would survive, and a hundred runs would leave a hundred
|
|
schemas for organizations that no longer exist. Only reached when the guard
|
|
above established the database held nothing to begin with.
|
|
|
|
One schema per transaction, and that is load-bearing rather than tidy. Dropping
|
|
several hundred schemas in a single transaction exhausts the lock table and
|
|
fails with "out of shared memory ... max_locks_per_transaction" - a message
|
|
that reads like a PostgreSQL defect and is not one. It is the same constraint
|
|
the 0005 rollback documents, learned here first.
|
|
"""
|
|
|
|
pool = ConnectionPool(dsn, min_size=1, max_size=2, open=True, kwargs={"row_factory": dict_row})
|
|
try:
|
|
with pool.connection() as connection:
|
|
schemas = connection.execute(
|
|
"SELECT nspname FROM pg_namespace WHERE nspname LIKE 'tenant\\_%' ORDER BY nspname"
|
|
).fetchall()
|
|
for schema in schemas:
|
|
connection.execute(f'DROP SCHEMA IF EXISTS "{schema["nspname"]}" CASCADE')
|
|
connection.commit()
|
|
finally:
|
|
pool.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def postgres(require_empty_database: None) -> Any:
|
|
dsn = _integration_dsn()
|
|
if dsn is None: # pragma: no cover - the skip above already guards this
|
|
pytest.skip("no test DSN")
|
|
if "umbracontrol" in dsn: # pragma: no cover - a safety net
|
|
pytest.skip("refusing to run the adapter suite against umbracontrol")
|
|
repository = PostgresAccountRepository.from_dsn(
|
|
dsn, min_size=1, max_size=4, application_name="umbratrans-adapter-test"
|
|
)
|
|
try:
|
|
yield repository
|
|
# Safe precisely because of the guard above: this database held nothing
|
|
# before the run, so emptying these tables removes only what this run
|
|
# created. TRUNCATE rather than DELETE because audit_events has a trigger
|
|
# that raises on delete; truncating does not fire it, and takes the
|
|
# tables that reference these with it.
|
|
#
|
|
# `CASCADE` reaches the tenant schemas as well, which is the point of
|
|
# mentioning it: each of the four tenant tables has a foreign key to
|
|
# `public.users`, so a project's rows go when users go. The schemas
|
|
# themselves are dropped once per session, in `require_empty_database`.
|
|
with repository._pool.connection() as connection: # noqa: SLF001 - test teardown
|
|
connection.execute("TRUNCATE public.users, public.organizations, public.audit_events CASCADE")
|
|
finally:
|
|
repository.close()
|
|
|
|
|
|
# -- fixtures over the real database --------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def pg_settings() -> Settings:
|
|
return Settings(
|
|
environment=Environment.TEST,
|
|
repository_kind="postgres",
|
|
database_dsn=_integration_dsn(),
|
|
argon2_memory_cost_kib=8 * 1024,
|
|
argon2_time_cost=1,
|
|
argon2_parallelism=1,
|
|
password_min_length=14,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def pg_service(postgres: PostgresAccountRepository, pg_settings: Settings) -> AccountService:
|
|
return AccountService(postgres, pg_settings, mailer=CapturingMailer())
|
|
|
|
|
|
@pytest.fixture
|
|
def platform(pg_service: AccountService) -> Principal:
|
|
"""A real platform administrator, bootstrapped through the operator path.
|
|
|
|
A platform admin needs no organization, which is what makes it the one role
|
|
that can be created before anything else exists.
|
|
"""
|
|
|
|
password = random_password(30)
|
|
view = pg_service.create_account(
|
|
enrollment_mode="temporary_password",
|
|
username=f"Platform.Admin.{uuid4().hex[:8]}",
|
|
display_name="Platform Admin",
|
|
email=f"platform-{uuid4().hex[:8]}@example.test",
|
|
role="platform_admin",
|
|
password=password,
|
|
operator=OperatorContext(source="operator-cli"),
|
|
)
|
|
return Principal(
|
|
user_id=view.id,
|
|
username=view.username or "",
|
|
role=RoleName.PLATFORM_ADMIN,
|
|
session_id=uuid4(),
|
|
)
|
|
|
|
|
|
# -- the contract that only a real database can prove ----------------------
|
|
|
|
|
|
@integration
|
|
def test_the_rls_identity_contract_holds_against_the_real_schema(
|
|
postgres: PostgresAccountRepository,
|
|
) -> None:
|
|
"""The startup self-check must pass against the control plane's own function.
|
|
|
|
This is the test that would have caught the wrong GUC name. It is cheap, it
|
|
runs before anything else, and a failure here means every policy would deny
|
|
every request.
|
|
"""
|
|
|
|
postgres.verify_rls_contract()
|
|
|
|
|
|
@integration
|
|
def test_the_tenant_schema_contract_holds_against_the_real_schema(
|
|
postgres: PostgresAccountRepository,
|
|
) -> None:
|
|
"""The same self-check for the tenant schema name.
|
|
|
|
The name is derived in two languages - Python here, ``app.tenant_schema_name``
|
|
in the migration that created the schemas - and nothing joins them. A
|
|
disagreement is silent and total: every tenant query would resolve into a
|
|
schema that does not exist, so every project request for every organization
|
|
would raise "relation does not exist", and the error would name the statement
|
|
rather than the cause.
|
|
|
|
Asserted against the real function rather than against a constant, because a
|
|
constant in the test would agree with itself whatever the database said.
|
|
"""
|
|
|
|
postgres.verify_tenant_schema_contract()
|
|
|
|
|
|
@integration
|
|
def test_a_tenant_schema_mismatch_is_refused_rather_than_served(
|
|
postgres: PostgresAccountRepository,
|
|
) -> None:
|
|
"""The check has to be able to fail, or passing it means nothing.
|
|
|
|
The adapter's name derivation is pointed at a prefix the database does not
|
|
use, and the self-check has to refuse. A check that cannot fail is the same
|
|
failure mode as a policy nobody wrote: it reports success while enforcing
|
|
nothing.
|
|
"""
|
|
|
|
with pytest.MonkeyPatch.context() as monkeypatch:
|
|
monkeypatch.setattr("umbratrans_web.postgres_repository.tenant_schema_name", lambda _: "tenant_not_this_one")
|
|
with pytest.raises(RepositoryError) as error:
|
|
postgres.verify_tenant_schema_contract()
|
|
assert error.value.code == "tenant_schema_mismatch"
|
|
|
|
|
|
@integration
|
|
def test_organization_rows_round_trip(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
created = pg_service.create_organization(platform, name="Round Trip Org", slug=f"round-trip-{uuid4().hex[:8]}")
|
|
stored = postgres.get_organization(created.id)
|
|
assert stored is not None
|
|
assert stored.name == "Round Trip Org"
|
|
|
|
by_slug = postgres.get_organization_by_slug(stored.slug.upper())
|
|
assert by_slug is not None
|
|
assert by_slug.id == created.id
|
|
|
|
organizations, total = postgres.find_organizations(limit=10)
|
|
assert total >= 1
|
|
assert any(item.id == created.id for item in organizations)
|
|
|
|
|
|
@integration
|
|
def test_a_role_resolves_through_user_roles(
|
|
pg_service: AccountService, postgres: PostgresAccountRepository, platform: Principal
|
|
) -> None:
|
|
"""The adapter has no role column to read, so this exercises the join.
|
|
|
|
Creating the account also has to write the user_roles row, and reading it
|
|
back has to resolve the role and organization from that table alone.
|
|
"""
|
|
|
|
organization = pg_service.create_organization(platform, name="Scoped Org", slug=f"scoped-{uuid4().hex[:8]}")
|
|
email = f"scoped-{uuid4().hex[:8]}@example.test"
|
|
created = pg_service.create_oidc_account(
|
|
username=f"Scoped.Member.{uuid4().hex[:6]}",
|
|
display_name="Scoped Member",
|
|
email=email,
|
|
organization_id=organization.id,
|
|
actor=platform,
|
|
)
|
|
assert created.organization_id == organization.id
|
|
|
|
fetched = postgres.get_user_by_email(email)
|
|
assert fetched is not None
|
|
assert fetched.role is RoleName.MEMBER
|
|
assert fetched.organization_id == organization.id
|
|
assert fetched.status is AccountStatus.INVITED
|
|
assert fetched.is_email_verified is False
|
|
|
|
|
|
@integration
|
|
def test_the_database_stores_only_an_argon2id_hash(
|
|
pg_service: AccountService, postgres: PostgresAccountRepository
|
|
) -> None:
|
|
"""A stored credential holds an Argon2id hash, a change flag, and nothing else.
|
|
|
|
Scoped to the account this test creates. An earlier version selected every
|
|
credential in the table, which made the result depend on the history of
|
|
whatever shared database the suite pointed at: it passed only while no
|
|
account had replaced its one-time password, and failed as soon as one had.
|
|
A test that asserts about another run's rows is not a test of this code.
|
|
"""
|
|
|
|
# Platform scope, because the operator path still assigns a role and every
|
|
# other role needs an organization. The tenant rules have their own tests;
|
|
# this one is only about what lands in user_credentials.
|
|
account = pg_service.create_account(
|
|
enrollment_mode="temporary_password",
|
|
username=f"Hash.Check.{uuid4().hex[:6]}",
|
|
display_name="Hash Check",
|
|
email=f"hash-{uuid4().hex[:8]}@example.test",
|
|
role="platform_admin",
|
|
password=random_password(30),
|
|
operator=OperatorContext(source="operator-cli"),
|
|
)
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - test introspection
|
|
row = connection.execute(
|
|
"SELECT password_hash, must_change_password, credential_type "
|
|
"FROM public.user_credentials WHERE user_id = %s",
|
|
(account.id,),
|
|
).fetchone()
|
|
|
|
assert row is not None, "no credential row was written"
|
|
assert row["password_hash"].startswith("$argon2id$")
|
|
assert row["must_change_password"] is True
|
|
assert row["credential_type"] == "local_password"
|
|
|
|
|
|
@integration
|
|
def test_a_replaced_password_is_not_still_flagged_for_change(
|
|
pg_service: AccountService, postgres: PostgresAccountRepository
|
|
) -> None:
|
|
"""The flag clears when the password is replaced, in the store.
|
|
|
|
The other half of the property above, and the reason it has to be scoped:
|
|
a credential whose one-time password has been replaced must stop being marked
|
|
as one, or every later sign-in would be pushed back through the change flow.
|
|
"""
|
|
|
|
account = provision_local(pg_service, username=f"Replaced.{uuid4().hex[:6]}")
|
|
service = AccountService(pg_service.repository, pg_service.settings)
|
|
# Signed in first, because the change requires a principal resolved from a
|
|
# session. That is the real order: the one-time password produces a session
|
|
# flagged for change, and the change is made on that session.
|
|
result = service.authenticate(email=account.email, password=account.password)
|
|
assert result.session.must_change_password is True
|
|
principal = service.resolve_session(result.session.token)
|
|
replacement = random_password(30)
|
|
service.change_first_password(
|
|
principal,
|
|
current_password=account.password,
|
|
new_password=replacement,
|
|
)
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - test introspection
|
|
row = connection.execute(
|
|
"SELECT must_change_password, password_hash FROM public.user_credentials WHERE user_id = %s",
|
|
(account.user_id,),
|
|
).fetchone()
|
|
assert row is not None
|
|
assert row["must_change_password"] is False
|
|
assert row["password_hash"].startswith("$argon2id$")
|
|
|
|
|
|
@integration
|
|
def test_email_verification_survives_a_real_round_trip(pg_service: AccountService, platform: Principal) -> None:
|
|
"""The full flow, through real tables, using the real policy helpers."""
|
|
|
|
mailer = CapturingMailer()
|
|
service = AccountService(pg_service.repository, pg_service.settings, mailer=mailer)
|
|
organization = service.create_organization(platform, name="Verify Org", slug=f"verify-{uuid4().hex[:8]}")
|
|
created = provision_invite(
|
|
service,
|
|
actor=platform,
|
|
username=f"Verify.Live.{uuid4().hex[:6]}",
|
|
display_name="Verify Live",
|
|
email=f"verify-{uuid4().hex[:8]}@example.test",
|
|
organization_id=organization.id,
|
|
)
|
|
assert created.account.status is AccountStatus.INVITED
|
|
with pytest.raises(EmailNotVerifiedError):
|
|
service.authenticate(email=created.account.email, password=created.temporary_password)
|
|
|
|
verified = service.verify_email(verification_token(mailer))
|
|
assert verified.status is AccountStatus.ACTIVE
|
|
# And the password now works.
|
|
assert service.authenticate(email=verified.email, password=created.temporary_password)
|
|
|
|
|
|
@integration
|
|
def test_suspending_an_organization_blocks_its_member(pg_service: AccountService, platform: Principal) -> None:
|
|
organization = pg_service.create_organization(platform, name="Lockout Org", slug=f"lockout-{uuid4().hex[:8]}")
|
|
password = random_password(30)
|
|
email = f"lockout-{uuid4().hex[:8]}@example.test"
|
|
pg_service.create_account(
|
|
enrollment_mode="temporary_password",
|
|
username=f"Lockout.Member.{uuid4().hex[:6]}",
|
|
display_name="Lockout Member",
|
|
email=email,
|
|
organization_id=organization.id,
|
|
password=password,
|
|
operator=OperatorContext(source="operator-cli"),
|
|
)
|
|
assert pg_service.authenticate(email=email, password=password) is not None
|
|
|
|
pg_service.set_organization_status(platform, organization.id, "suspended")
|
|
with pytest.raises(InvalidCredentialsError):
|
|
pg_service.authenticate(email=email, password=password)
|
|
|
|
|
|
@integration
|
|
def test_audit_events_round_trip_with_the_store_identifier(pg_service: AccountService, platform: Principal) -> None:
|
|
pg_service.create_organization(platform, name="Audited Org", slug=f"audited-{uuid4().hex[:8]}")
|
|
events, total = pg_service.repository.list_audit_events(limit=50)
|
|
assert total >= 1
|
|
assert events
|
|
# audit_events.id is a bigint identity, so the domain id is its text form.
|
|
assert all(event.id.isdigit() for event in events)
|
|
assert any(event.action == "organization.created" for event in events)
|
|
|
|
|
|
@integration
|
|
def test_the_health_probe_reports_reachability(postgres: PostgresAccountRepository) -> None:
|
|
assert postgres.check() is True
|
|
|
|
|
|
@integration
|
|
def test_a_populated_database_stops_the_suite_rather_than_being_emptied(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService
|
|
) -> None:
|
|
"""The guard that replaces trying to clean up carefully.
|
|
|
|
Creating a row and then asserting the suite declines to run is awkward to
|
|
express as a test of the suite, because the guard runs before any test does.
|
|
So this asserts the property the guard depends on: that a populated database
|
|
is detected, and that the detection names the tables rather than failing
|
|
obscurely.
|
|
"""
|
|
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - test setup
|
|
row = connection.execute(
|
|
"INSERT INTO public.organizations (id, name, slug, status, created_at, updated_at) "
|
|
"VALUES (%s, 'Occupancy Check', %s, 'active', now(), now()) RETURNING id",
|
|
(uuid4(), f"occupancy-{uuid4().hex[:12]}"),
|
|
).fetchone()
|
|
identifier = row["id"]
|
|
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - test assertion
|
|
counts = {
|
|
table: int(connection.execute(f"SELECT count(*) AS total FROM public.{table}").fetchone()["total"])
|
|
for table in ("users", "organizations", "audit_events")
|
|
}
|
|
assert counts["organizations"] >= 1
|
|
assert any(count for count in counts.values()), "the guard would not notice this database"
|
|
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - test cleanup
|
|
connection.execute("DELETE FROM public.organizations WHERE id = %s", (identifier,))
|
|
|
|
|
|
def test_the_guard_covers_every_table_the_teardown_empties() -> None:
|
|
"""The guard and the teardown must name the same tables, or one is a lie.
|
|
|
|
A table emptied by the teardown but not checked by the guard would be
|
|
destroyed without the suite ever objecting.
|
|
|
|
The names are read out of the teardown's own TRUNCATE statement and then looked
|
|
up in the guard's list, rather than the other way round. An earlier version
|
|
searched the guard's source text for the string `"users"`, which stopped being
|
|
true the moment the guard began delegating to a constant - and a text search
|
|
that fails when a refactor moves a string is a test of the layout, not of the
|
|
property. Deriving both sides from what they actually do keeps the property and
|
|
drops the coincidence.
|
|
"""
|
|
|
|
import inspect
|
|
import re
|
|
|
|
source = inspect.getsource(postgres)
|
|
# Comments first. The teardown's own comment explains why it truncates rather
|
|
# than deletes, and the word TRUNCATE appears in that sentence too - which is
|
|
# how the first version of this parsed a paragraph as a table list.
|
|
statements = re.sub(r"#.*", "", source)
|
|
truncate = re.search(r"TRUNCATE\s+(public\.[^;)]+?)(?:CASCADE)?[\"']", statements)
|
|
assert truncate is not None, "the teardown no longer truncates anything; this test cannot check it"
|
|
emptied = {name.strip().split(".")[-1] for name in truncate.group(1).split(",")}
|
|
unguarded = emptied - set(PLATFORM_TABLES)
|
|
assert not unguarded, f"the teardown empties {sorted(unguarded)} and the guard does not look at them"
|
|
|
|
|
|
@integration
|
|
def test_unassigning_can_only_reach_one_accounts_assignment_to_one_project(
|
|
postgres: PostgresAccountRepository,
|
|
) -> None:
|
|
"""The one permitted project delete, bounded by both a user and a project.
|
|
|
|
Removing an assignment is the operation's whole purpose, so unlike the token
|
|
purge it cannot be avoided. What must not be possible is for it to reach
|
|
anything else: another account's assignment, another project, or the
|
|
account's organization-scoped assignment, which grants tenant membership and
|
|
is not the same thing at all.
|
|
"""
|
|
|
|
import inspect
|
|
|
|
source = inspect.getsource(PostgresAccountRepository.unassign_user_from_project)
|
|
assert "DELETE FROM public.user_roles" in source
|
|
# Both bounds, and the scope, are in the predicate.
|
|
assert "user_id = %s" in source
|
|
assert "project_id = %s" in source
|
|
assert "scope = 'project'" in source
|
|
# It must not reach the accounts, the projects, or anything else.
|
|
for other in ("public.users", "public.projects", "public.organizations", "public.user_credentials"):
|
|
assert other not in source, other
|
|
assert "TRUNCATE" not in source
|
|
|
|
|
|
def _active_tenant_rows(connection: Any, user_id: UUID) -> list[Any]:
|
|
"""Active assignments that state which *tenant* the account belongs to.
|
|
|
|
A project-scoped row carries an organization too, but it says who is working
|
|
on what rather than where the account lives, so it is not one of these.
|
|
"""
|
|
|
|
return connection.execute(
|
|
"SELECT id FROM public.user_roles WHERE user_id = %s AND status = 'active' AND scope <> 'project'",
|
|
(user_id,),
|
|
).fetchall()
|
|
|
|
|
|
def _active_project_rows(connection: Any, user_id: UUID) -> list[Any]:
|
|
return connection.execute(
|
|
"SELECT id FROM public.user_roles WHERE user_id = %s AND status = 'active' AND scope = 'project'",
|
|
(user_id,),
|
|
).fetchall()
|
|
|
|
|
|
def _member(pg_service: AccountService, platform: Principal, organization_id: UUID, name: str) -> UUID:
|
|
"""Create an ordinary member of one organization and return their user id.
|
|
|
|
Goes through the real service so the account lands the way a real one does -
|
|
an active organization-scoped assignment, and a ``user_roles`` row the
|
|
project-assignment tests then have to coexist with.
|
|
"""
|
|
|
|
view = pg_service.create_account(
|
|
enrollment_mode="temporary_password",
|
|
username=f"{name}.{uuid4().hex[:6]}",
|
|
display_name=name,
|
|
email=f"{name.lower()}-{uuid4().hex[:8]}@example.test",
|
|
role=str(RoleName.MEMBER),
|
|
organization_id=organization_id,
|
|
password=random_password(28),
|
|
# The operator path, because temporary-password enrollment is
|
|
# operator-only by design. A platform administrator provisioning through
|
|
# the API generates the password in process instead, and these tests want
|
|
# the rows, not the enrollment mechanics.
|
|
operator=OperatorContext(source="operator-cli"),
|
|
)
|
|
return view.id
|
|
|
|
|
|
# -- projects, against the real schema ---------------------------------------
|
|
#
|
|
# The ``projects`` table and the project-scoped arm of ``user_roles`` were already
|
|
# in the control plane and unused by this application. Everything below therefore
|
|
# checks statements against a real database rather than a fake one: the statuses
|
|
# come from a check constraint, the slug uniqueness is a unique index over
|
|
# ``(organization_id, slug)``, and membership is expressed as rows in the same
|
|
# table the account roles live in, which is where a wrong column or a wrong scope
|
|
# would hide.
|
|
|
|
|
|
@integration
|
|
def test_project_rows_round_trip_through_the_real_table(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
organization = pg_service.create_organization(platform, name="Project Org", slug=f"proj-{uuid4().hex[:8]}")
|
|
created = pg_service.create_project(platform, name="Book One", organization_id=organization.id)
|
|
assert created.status is ProjectStatus.ACTIVE
|
|
assert created.slug == "book-one", "the slug must be derived from the name by the service"
|
|
|
|
stored = postgres.get_project(created.id, organization_id=organization.id)
|
|
assert stored is not None
|
|
assert stored.id == created.id
|
|
assert stored.name == "Book One"
|
|
assert stored.organization_id == organization.id
|
|
|
|
# And it is findable by slug, case-insensitively, as citext promises.
|
|
by_slug = postgres.get_project_by_slug("BOOK-ONE", organization_id=organization.id)
|
|
assert by_slug is not None and by_slug.id == created.id
|
|
|
|
|
|
@integration
|
|
def test_a_project_lookup_is_scoped_in_the_query_not_after_it(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""Naming another organization must not return a row at all.
|
|
|
|
The distinction is the point of the ``organization_id`` parameter: filtering
|
|
afterwards would have read a row belonging to another tenant and then decided
|
|
it was not the caller's, which is a read that should never have happened and
|
|
which a later change to the check could turn into a leak.
|
|
"""
|
|
|
|
first = pg_service.create_organization(platform, name="Scope One", slug=f"scope-a-{uuid4().hex[:8]}")
|
|
second = pg_service.create_organization(platform, name="Scope Two", slug=f"scope-b-{uuid4().hex[:8]}")
|
|
project = pg_service.create_project(platform, name="Private", organization_id=first.id)
|
|
|
|
assert postgres.get_project(project.id, organization_id=first.id) is not None
|
|
assert postgres.get_project(project.id, organization_id=second.id) is None
|
|
assert postgres.get_project_by_slug(project.slug, organization_id=second.id) is None
|
|
|
|
|
|
@integration
|
|
def test_project_status_moves_through_the_columns_own_check_constraint(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""The four modelled values are the constraint's, read from the table itself.
|
|
|
|
A status the database refuses would raise here rather than surfacing as a
|
|
500 on a request an administrator believed had worked.
|
|
"""
|
|
|
|
organization = pg_service.create_organization(platform, name="Status Org", slug=f"st-{uuid4().hex[:8]}")
|
|
project = pg_service.create_project(platform, name="Lifecycle", organization_id=organization.id)
|
|
for status in (ProjectStatus.ON_HOLD, ProjectStatus.ACTIVE, ProjectStatus.RELEASED, ProjectStatus.ARCHIVED):
|
|
pg_service.set_project_status(platform, project.id, status)
|
|
assert postgres.get_project(project.id).status is status
|
|
|
|
|
|
@integration
|
|
def test_a_slug_repeats_across_organizations_but_not_within_one(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""The unique index is over ``(organization_id, slug)``, and that is the rule.
|
|
|
|
The same handle in two tenants is not a collision; in one tenant it is. If
|
|
this ever fails, the index has been narrowed to ``slug`` alone and a tenant
|
|
can no longer use a name another one already took.
|
|
"""
|
|
|
|
first = pg_service.create_organization(platform, name="Slug One", slug=f"slug-a-{uuid4().hex[:8]}")
|
|
second = pg_service.create_organization(platform, name="Slug Two", slug=f"slug-b-{uuid4().hex[:8]}")
|
|
pg_service.create_project(platform, name="Shared Name", slug="shared-name", organization_id=first.id)
|
|
pg_service.create_project(platform, name="Shared Name", slug="shared-name", organization_id=second.id)
|
|
with pytest.raises(ConflictError):
|
|
pg_service.create_project(platform, name="Other", slug="shared-name", organization_id=first.id)
|
|
|
|
|
|
@integration
|
|
def test_assignment_is_a_project_scoped_row_in_user_roles(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""Membership lives in the same table as roles, distinguished by scope.
|
|
|
|
Getting the scope or the organization wrong here is the quiet failure worth
|
|
catching: an assignment written with an organization scope would grant tenant
|
|
membership, and one written without a project would be invisible to the
|
|
``project`` scope every policy and every query filters on.
|
|
"""
|
|
|
|
organization = pg_service.create_organization(platform, name="Assign Org", slug=f"as-{uuid4().hex[:8]}")
|
|
project = pg_service.create_project(platform, name="Assigned", organization_id=organization.id)
|
|
account = _member(pg_service, platform, organization.id, "Assignee")
|
|
|
|
assert postgres.assign_user_to_project(project.id, account, organization_id=organization.id) is True
|
|
assert (
|
|
postgres.assign_user_to_project(project.id, account, organization_id=organization.id) is False
|
|
), "assigning twice is a no-op"
|
|
assert postgres.list_project_member_ids(project.id) == [account]
|
|
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - asserting the stored row's shape
|
|
row = connection.execute(
|
|
"SELECT scope, organization_id, project_id, status FROM public.user_roles "
|
|
"WHERE user_id = %s AND project_id = %s",
|
|
(account, project.id),
|
|
).fetchone()
|
|
assert row is not None
|
|
assert row["scope"] == "project"
|
|
assert row["status"] == "active"
|
|
# The organization is carried on the project-scoped row too, so the row cannot
|
|
# name a project in one tenant while claiming membership of another.
|
|
assert row["organization_id"] == organization.id
|
|
|
|
assert postgres.unassign_user_from_project(project.id, account) is True
|
|
assert postgres.unassign_user_from_project(project.id, account) is False
|
|
assert postgres.list_project_member_ids(project.id) == []
|
|
|
|
|
|
@integration
|
|
def test_unassigning_leaves_the_accounts_tenant_membership_alone(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""Two rows in one table, and only one of them may be removed.
|
|
|
|
This is the specific damage a wrong predicate would cause: an account removed
|
|
from a project and, in the same statement, from its organization - which ends
|
|
their access to every project in the tenant. So the test asserts the account
|
|
is still a member afterwards, not merely that the project has no members.
|
|
"""
|
|
|
|
organization = pg_service.create_organization(platform, name="Keep Org", slug=f"kp-{uuid4().hex[:8]}")
|
|
project = pg_service.create_project(platform, name="Keep", organization_id=organization.id)
|
|
account = _member(pg_service, platform, organization.id, "Keeper")
|
|
postgres.assign_user_to_project(project.id, account, organization_id=organization.id)
|
|
postgres.unassign_user_from_project(project.id, account)
|
|
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - asserting nothing else was touched
|
|
membership = connection.execute(
|
|
"SELECT count(*) AS total FROM public.user_roles "
|
|
"WHERE user_id = %s AND organization_id = %s AND project_id IS NULL",
|
|
(account, organization.id),
|
|
).fetchone()
|
|
assert membership["total"] == 1, "the account lost its organization membership"
|
|
|
|
|
|
@integration
|
|
def test_assigning_an_account_to_a_project_in_another_organization_is_refused_by_the_service(
|
|
pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""The check is in the service and holds against the real rows.
|
|
|
|
The database has no policy that prevents it - row-level security is not
|
|
enforced, because the application owns its tables - so this is enforced
|
|
exactly once, here, and the test exists to keep that honest.
|
|
"""
|
|
|
|
mine = pg_service.create_organization(platform, name="Mine Real", slug=f"mr-{uuid4().hex[:8]}")
|
|
theirs = pg_service.create_organization(platform, name="Theirs Real", slug=f"tr-{uuid4().hex[:8]}")
|
|
project = pg_service.create_project(platform, name="Boundary", organization_id=mine.id)
|
|
outsider = _member(pg_service, platform, theirs.id, "Outsider")
|
|
with pytest.raises(ForbiddenError):
|
|
pg_service.assign_account_to_project(platform, project.id, outsider)
|
|
assert pg_service.get_project(platform, project.id).assigned_count == 0
|
|
|
|
|
|
@integration
|
|
def test_projects_are_listed_with_their_organization_and_paged(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
organization = pg_service.create_organization(platform, name="Paging Org", slug=f"pg-{uuid4().hex[:8]}")
|
|
for index in range(3):
|
|
pg_service.create_project(platform, name=f"Book {index}", organization_id=organization.id)
|
|
|
|
rows, total = postgres.find_projects(organization_id=organization.id, limit=2, offset=0)
|
|
assert total == 3
|
|
assert len(rows) == 2
|
|
assert {row.organization_id for row in rows} == {organization.id}
|
|
|
|
page_two, _ = postgres.find_projects(organization_id=organization.id, limit=2, offset=2)
|
|
assert len(page_two) == 1
|
|
# Paging must not repeat a row across the boundary.
|
|
assert not ({row.id for row in rows} & {row.id for row in page_two})
|
|
assert postgres.count_projects(organization_id=organization.id) == 3
|
|
|
|
|
|
@integration
|
|
def test_the_organization_wide_listing_reaches_a_members_projects(
|
|
pg_service: AccountService, postgres: PostgresAccountRepository, platform: Principal
|
|
) -> None:
|
|
"""Bob and Sally, end to end, over the real adapter.
|
|
|
|
The service tests prove the rule with the in-memory store, which cannot
|
|
disagree with itself. This proves the SQL it compiles to returns the same
|
|
thing: one member's assignment does not narrow the other's list.
|
|
"""
|
|
|
|
organization = pg_service.create_organization(platform, name="Shared Real", slug=f"sr-{uuid4().hex[:8]}")
|
|
book_one = pg_service.create_project(platform, name="Real Book One", organization_id=organization.id)
|
|
book_two = pg_service.create_project(platform, name="Real Book Two", organization_id=organization.id)
|
|
bob = _member(pg_service, platform, organization.id, "Bob")
|
|
sally = _member(pg_service, platform, organization.id, "Sally")
|
|
pg_service.assign_account_to_project(platform, book_one.id, bob)
|
|
pg_service.assign_account_to_project(platform, book_two.id, sally)
|
|
|
|
def member(user_id: UUID) -> Principal:
|
|
return Principal(
|
|
user_id=user_id, username="m", role=RoleName.MEMBER, organization_id=organization.id, session_id=uuid4()
|
|
)
|
|
|
|
for viewer in (bob, sally):
|
|
seen, total = pg_service.list_projects(member(viewer))
|
|
assert total == 2, "a member must see the whole organization"
|
|
assert {project.name for project in seen} == {"Real Book One", "Real Book Two"}
|
|
# And sees the other person's work described accurately: Sally is told Bob
|
|
# is on book one, and Bob is told Sally is on book two.
|
|
assert pg_service.get_project(member(viewer), book_one.id).assigned_account_ids == (bob,)
|
|
assert pg_service.get_project(member(viewer), book_two.id).assigned_account_ids == (sally,)
|
|
|
|
assert postgres.list_project_member_ids(book_one.id) == [bob]
|
|
|
|
|
|
@integration
|
|
def test_an_assigned_member_can_still_sign_in(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""The regression this feature would otherwise have shipped.
|
|
|
|
A project assignment writes a *second* active ``user_roles`` row, and the
|
|
adapter's role resolution refuses an account holding more than one. So the
|
|
moment anybody was assigned to anything, they could no longer authenticate -
|
|
locked out of the very page the assignment was meant to show them, and only
|
|
discoverable by assigning somebody and then trying to sign in as them.
|
|
|
|
Both halves are asserted here rather than the resolution in isolation,
|
|
because the resolution is only the thing that breaks: the password, the
|
|
credential, and the session are all fine, and a test that stopped at the
|
|
adapter would say the account is resolvable while the sign-in still fails.
|
|
"""
|
|
|
|
organization = pg_service.create_organization(platform, name="Sign In Org", slug=f"si-{uuid4().hex[:8]}")
|
|
project = pg_service.create_project(platform, name="Assigned Book", organization_id=organization.id)
|
|
member = _member(pg_service, platform, organization.id, "Signin")
|
|
|
|
# Before any assignment the account resolves, and the check the adapter
|
|
# performs is visible rather than assumed.
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - asserting the stored shape
|
|
assert len(_active_tenant_rows(connection, member)) == 1
|
|
|
|
assert postgres.assign_user_to_project(project.id, member, organization_id=organization.id) is True
|
|
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - and now there are two rows
|
|
tenant_rows = _active_tenant_rows(connection, member)
|
|
project_rows = _active_project_rows(connection, member)
|
|
# The account really does hold two active rows now. That is intended: a
|
|
# project row is a statement about work, not about tenancy.
|
|
assert len(tenant_rows) == 1
|
|
assert len(project_rows) == 1
|
|
assert len(tenant_rows) + len(project_rows) == 2
|
|
|
|
# And the account still signs in, with the organization intact.
|
|
account = pg_service.self_view(member)
|
|
assert account.role == RoleName.MEMBER
|
|
assert account.organization_id == organization.id
|
|
assert pg_service.repository.get_user(member).organization_id == organization.id
|
|
|
|
|
|
@integration
|
|
def test_assigning_to_several_projects_does_not_accumulate_tenancy_rows(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""Working on three books is three project rows and still one tenant row.
|
|
|
|
Without this, the failure grows quietly: one assignment might be fixed by a
|
|
narrow predicate while the second and third are what actually lock somebody
|
|
out, and a member assigned to a busy project is locked out sooner.
|
|
"""
|
|
|
|
organization = pg_service.create_organization(platform, name="Busy Org", slug=f"bo-{uuid4().hex[:8]}")
|
|
member = _member(pg_service, platform, organization.id, "Busy")
|
|
for index in range(3):
|
|
project = pg_service.create_project(platform, name=f"Busy {index}", organization_id=organization.id)
|
|
assert postgres.assign_user_to_project(project.id, member, organization_id=organization.id) is True
|
|
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - asserting the stored shape
|
|
assert len(_active_tenant_rows(connection, member)) == 1
|
|
assert len(_active_project_rows(connection, member)) == 3
|
|
assert pg_service.self_view(member).organization_id == organization.id
|
|
|
|
|
|
@integration
|
|
def test_two_tenants_for_one_account_is_still_refused(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""The guard must survive the fix that lets project rows through.
|
|
|
|
``_resolve_assignment`` refuses an account holding more than one *tenant*
|
|
assignment, because the organization it returns scopes every administrative
|
|
decision. Narrowing the query to tenant rows must narrow it to the ambiguity
|
|
that matters, not remove the check: two organizations is a genuine conflict,
|
|
and resolving it to whichever row sorted first would hand out the wrong
|
|
tenant's authority.
|
|
"""
|
|
|
|
first = pg_service.create_organization(platform, name="Cross One", slug=f"x1-{uuid4().hex[:8]}")
|
|
second = pg_service.create_organization(platform, name="Cross Two", slug=f"x2-{uuid4().hex[:8]}")
|
|
member = _member(pg_service, platform, first.id, "Crosser")
|
|
|
|
# Written directly, because the service has no path that would do this - which
|
|
# is the point. The guard is the last line if one ever appears.
|
|
with postgres._pool.connection() as connection: # noqa: SLF001 - constructing the impossible state
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO public.user_roles (user_id, role_id, scope, organization_id, status)
|
|
SELECT %s, r.id, 'organization', %s, 'active'
|
|
FROM public.roles r WHERE r.name = 'member' AND r.scope = 'organization'
|
|
""",
|
|
(member, second.id),
|
|
)
|
|
|
|
with pytest.raises(RepositoryError) as error:
|
|
postgres.get_user(member)
|
|
assert error.value.code == "ambiguous_role_assignment"
|
|
|
|
|
|
@integration
|
|
def test_the_health_probe_queries_run_against_the_real_schema(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""``/api/health`` is unauthenticated and polled, and it is not safe.
|
|
|
|
It calls ``setup_is_required``, which asks whether any platform administrator
|
|
exists. That makes ``count_users_with_role`` the one query that runs on every
|
|
probe, and a syntax error in it is a 500 on the endpoint used to decide whether
|
|
the deployment is up at all.
|
|
|
|
This reached production: the tenant-row predicate was added to that query as a
|
|
literal ``{TENANT_ROW_PREDICATE}`` inside a plain string rather than an
|
|
f-string, so the braces were sent to PostgreSQL verbatim. Every unit test
|
|
passed, because none of them execute this query; the deployment's health check
|
|
was the first thing to run it.
|
|
"""
|
|
|
|
# Every query the probe makes, run against the real schema so a statement that
|
|
# cannot parse fails here rather than in production. The values are secondary;
|
|
# the point is that they come back at all.
|
|
assert pg_service.setup_is_required() is False
|
|
assert isinstance(pg_service.setup_is_required(), bool)
|
|
assert postgres.count_users_with_role(RoleName.PLATFORM_ADMIN) >= 1
|
|
assert postgres.count_users_with_role(RoleName.MEMBER) >= 0
|
|
assert postgres.count_users_with_role(RoleName.ORGANIZATION_ADMIN) >= 0
|
|
|
|
|
|
@integration
|
|
def test_setup_is_required_ignores_project_assignments(
|
|
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
|
|
) -> None:
|
|
"""A project row must never be mistaken for a platform administrator.
|
|
|
|
The predicate exists for this. Without it, a role whose *name* matched would
|
|
be counted at project scope, and first-run setup would report itself required
|
|
or unnecessary on the strength of who is working on a book.
|
|
"""
|
|
|
|
organization = pg_service.create_organization(platform, name="Setup Org", slug=f"so-{uuid4().hex[:8]}")
|
|
project = pg_service.create_project(platform, name="Setup Book", organization_id=organization.id)
|
|
member = _member(pg_service, platform, organization.id, "Setup")
|
|
postgres.assign_user_to_project(project.id, member, organization_id=organization.id)
|
|
|
|
before = postgres.count_users_with_role(RoleName.PLATFORM_ADMIN)
|
|
assert pg_service.setup_is_required() is False
|
|
assert postgres.count_users_with_role(RoleName.PLATFORM_ADMIN) == before
|
|
# And a member is a member exactly once, not once per project worked on.
|
|
assert postgres.count_users_with_role(RoleName.MEMBER) == 1
|
|
|
|
|
|
def test_no_sql_string_carries_an_uninterpolated_placeholder() -> None:
|
|
"""A ``{...}`` in SQL must be an f-string, or the braces reach PostgreSQL.
|
|
|
|
This shipped once. A predicate was introduced as a named constant and
|
|
interpolated into a query written as a plain triple-quoted string, so
|
|
``{TENANT_ROW_PREDICATE}`` was sent to the database as literal text and the
|
|
statement failed to parse - in ``count_users_with_role``, which every
|
|
unauthenticated health probe calls, so the whole deployment reported 500.
|
|
|
|
Nothing caught it. The tests that exercise this query are integration tests,
|
|
the health endpoint is unauthenticated so it is not covered by an
|
|
authenticated request, and a source-level check for the predicate's *presence*
|
|
is satisfied just as well by the broken version. This one looks at the parse
|
|
tree instead of the text, so a placeholder in a non-f-string is a failure
|
|
whether or not the surrounding code mentions the constant.
|
|
|
|
Placeholders that legitimately reach the database - psycopg's own ``%s`` - are
|
|
not ``{...}``, so they cannot be confused with this.
|
|
"""
|
|
|
|
source = inspect.getsource(PostgresAccountRepository) + inspect.getsource(
|
|
sys.modules[PostgresAccountRepository.__module__]
|
|
)
|
|
tree = ast.parse(source)
|
|
# Snippets, not line numbers. The source here is what ``inspect.getsource``
|
|
# returns, which is a class body lifted out of the file, so its line numbers do
|
|
# not correspond to lines in the file a reader would open. Naming a wrong line
|
|
# is worse than naming none.
|
|
offenders = [
|
|
node.value.strip().splitlines()[0] + " ..."
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str) and re.search(r"\{[A-Za-z_]", node.value)
|
|
]
|
|
assert not offenders, (
|
|
"these plain strings contain a {placeholder} that will be sent to PostgreSQL verbatim; "
|
|
f"prefix each with f: {offenders}"
|
|
)
|
|
|
|
# And the constant the queries rely on is actually interpolated somewhere, so
|
|
# it cannot be renamed out from under them.
|
|
interpolated = [
|
|
True
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.JoinedStr)
|
|
and any(
|
|
# Inside an f-string the variable is a Name wrapped in a
|
|
# FormattedValue, not a direct child of the JoinedStr.
|
|
isinstance(part, ast.FormattedValue)
|
|
and isinstance(part.value, ast.Name)
|
|
and part.value.id == "TENANT_ROW_PREDICATE"
|
|
for part in node.values
|
|
)
|
|
]
|
|
assert interpolated, "no f-string interpolates TENANT_ROW_PREDICATE; the queries would send braces"
|