diff --git a/docs/adr/0153-postgres-position-history-read.md b/docs/adr/0153-postgres-position-history-read.md new file mode 100644 index 000000000..1931866b8 --- /dev/null +++ b/docs/adr/0153-postgres-position-history-read.md @@ -0,0 +1,64 @@ +# ADR 0153: Read Position history from canonical PostgreSQL truth + +- **Status:** Proposed on active stacked PR #153; not protected `develop` truth until integrated +- **Date:** 2026-08-30 +- **Owners:** Orgmetra People API / HRIS persistence +- **Extends:** ADR 0003 (bitemporal HRIS data), ADR 0008 (purpose-bound PII authorization), ADR 0152 (Position-history read contract) + +## Context + +PR #152 defines the buyer-facing, purpose-bound Position-history read but intentionally injects its persistence port. An integrated application still needs a canonical adapter for normalized `position_record` and `position_record_version` truth; otherwise each deployment would supply bespoke persistence code and could silently widen the read. + +The adapter is not a second authorization engine or source of truth. The parent People service authorizes before calling it and revalidates its typed output before disclosure. The existing schema already owns Position/Job lineage, bitemporal version facts, tenant RLS, and immutable-history guards. + +A PostgreSQL statement that says `SET TRANSACTION ... READ ONLY` is not sufficient proof by itself. The adapter also uses transaction-local tenant context (`set_config(..., true)`), so both controls require an actual non-autocommit transaction. Likewise, validating a connection factory and then retaining it in a writable instance slot would create a checked-versus-used capability gap, and retaining caller UUID objects after validation would allow alias mutation between validation and SQL execution. + +## Decision + +Add `PostgresPositionHistoryReadPort` as the canonical PostgreSQL implementation of the `PositionHistoryReadPort` protocol introduced by PR #152. + +The adapter: + +1. validates exact operational tenant/Position UUIDs and an exact built-in UTC `known_at` before acquiring a connection; +2. immediately reduces tenant and Position UUIDs to exact built-in integer scalar authority, then reconstructs fresh UUID views only for DB-API parameter contracts; +3. stores the validated `connection_factory` in immutable tuple payload rather than a writable dataclass slot and uses that exact capability for the read; +4. requires the acquired connection to prove `autocommit is False` before cursor creation; absent, truthy, numeric-zero, string, or otherwise unproven modes fail closed; +5. only after that proof executes `SET TRANSACTION ISOLATION LEVEL READ COMMITTED, READ ONLY`, sets transaction-local tenant context, and runs the protected SELECT inside the same connection context; +6. joins only `public.position_record_version` to its Orgmetra-owned `public.position_record` anchor, preserving Job and organization lineage without Person, Employment, Assignment, compensation, candidate, performance, credential, or decision joins; +7. applies explicit tenant, Position, parent-recorded, and version-recorded half-open predicates; +8. projects recorded timestamps with `AT TIME ZONE 'UTC'`, accepts only exact naive UTC DB projections, and attaches built-in UTC after validation; +9. treats DB-API output as untrusted by checking the default list collection, exact tuple row shape, parent-record integrity, requested target identity, and knowledge-cutoff visibility before returning an immutable tuple; +10. compares returned tenant/Position scalar authority to the detached request scalars, so post-validation mutation of caller UUID aliases cannot change the authorized query target. + +Purpose-bound field authorization remains in the parent service. This adapter performs no mutation, audit/outbox write, foreign-service call, decision, or disclosure. + +## Workflow ownership + +Repository workflow consolidation on protected `develop@eb9757f8649aaad026a9865508d9aad50c1a7a4f` makes Foundation CI the repository acceptance owner. The pre-consolidation `.github/workflows/position-history-postgres-read-quality.yml` is not carried forward by the semantic restack. Stacked feature heads are not treated as GREEN merely because their parent or predecessor once passed a feature-local workflow. + +## Consequences + +### Positive + +- The Position-history application contract can use canonical normalized PostgreSQL truth without host-specific persistence code. +- The read-only and transaction-local tenant controls are only attempted after the connection proves a non-autocommit transaction mode. +- The connection capability used by the adapter cannot be swapped through ordinary or `object.__setattr__` instance mutation after construction. +- Caller-retained UUID aliases cannot change tenant/Position SQL parameters after validation. +- Explicit predicates and forced-RLS tenant context provide layered database scope controls. +- Position, Job, and Assignment remain separate concepts, and business-effective history remains distinct from system-recorded visibility. +- Exact DB timestamp validation prevents driver/session timezone behavior from silently changing evidence meaning. + +### Trade-offs + +- The adapter is PostgreSQL/DB-API specific and intentionally requires the default tuple-row contract. +- Compatible connection objects must expose `autocommit` and prove it with the exact built-in value `False`; implicit or driver-specific lookalikes are rejected at this high-trust boundary. +- Database RLS and bitemporal constraints still require independent PostgreSQL tests; this adapter does not claim that SQL predicates replace authorization or schema constraints. +- The parent service must be integrated first and must continue to revalidate rows before serialization. + +## Verification + +The original contract-first child head `bf93924e` established the adapter boundary while the module was absent. Historical local and isolated PostgreSQL checks from the pre-consolidation branch remain development evidence only; they are not current exact-head merge evidence. + +After PR #152 moved to its protected-workflow-consolidated and retained-authority-hardened head, ordinary two-parent reconciliation `a8edc7d2fa69842d4515def5c8a3e710ed4b4e2b` restacked the adapter without force push and deliberately omitted the stale feature-local quality workflow. Test-only head `800b783a74594a59f4c0d5dd819a6fce4fced4b8` added four contracts: immutable connection capability, explicit rejection of `autocommit=True`, rejection of unproven transaction modes, and request UUID alias detachment before connection acquisition. Production repair `00d8f58d0863ef0d3270c7197d41856de3131e34` implements those boundaries, and `d0a8b9b9b5e43bbc522b97c921a1d67c51b5fb6c` updates the ordinary test fixture to state its non-autocommit contract explicitly. + +Because #153 remains stacked on a feature branch, no hosted RED or GREEN is inferred from the absence of PR-triggered Foundation runs. The final descendant must first inherit an integrated/protected parent, retarget to `develop`, and then obtain fresh exact-current-head Foundation/Security/SAST/CodeQL evidence plus qualifying independent review. Parent #152 and predecessor evidence do not transfer. diff --git a/docs/doctoring/postgres-position-history-read-references.md b/docs/doctoring/postgres-position-history-read-references.md new file mode 100644 index 000000000..cd8de0cdb --- /dev/null +++ b/docs/doctoring/postgres-position-history-read-references.md @@ -0,0 +1,25 @@ +# PostgreSQL Position-history read references + +**Scope:** Standards and research basis for active PR #153. This file does not claim certification or protected-main integration. + +## APA 7 references + +Joint Task Force. (2020). *Security and privacy controls for information systems and organizations* (NIST Special Publication 800-53, Revision 5). National Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-53r5 + +Klyne, G., & Newman, C. (2002). *Date and time on the Internet: Timestamps* (RFC 3339). RFC Editor. https://doi.org/10.17487/RFC3339 + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: SET TRANSACTION*. https://www.postgresql.org/docs/18/sql-set-transaction.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Row security policies*. https://www.postgresql.org/docs/18/ddl-rowsecurity.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Range types*. https://www.postgresql.org/docs/18/rangetypes.html + +## Decision relevance + +PostgreSQL's transaction access mode and isolation level support the adapter's explicit `READ COMMITTED, READ ONLY` boundary. Row security remains database defense in depth, while the application still binds tenant context and checks exact returned identity. PostgreSQL range/exclusion semantics remain the schema-level basis for bitemporal non-overlap; this read adapter does not replace those constraints. + +RFC 3339 and explicit UTC projection support one interoperable representation for system-recorded evidence. NIST SP 800-53 Rev. 5 informs least privilege, access control, and information-integrity evidence readiness; no compliance or certification claim follows from this PR. + +## Research classification + +These references constrain the accepted adapter architecture for PR #153. They do not authorize scope expansion into worker data, Assignment joins, compensation, candidate, performance, or employment-decision automation. diff --git a/docs/traceability/postgres-position-history-read.md b/docs/traceability/postgres-position-history-read.md new file mode 100644 index 000000000..71ad91c40 --- /dev/null +++ b/docs/traceability/postgres-position-history-read.md @@ -0,0 +1,54 @@ +# PostgreSQL Position-history read traceability + +**Lifecycle status:** Proposed on active stacked PR #153 only. This document does not claim protected-`develop` integration. + +## Buyer problem + +PR #152 defines an authorized Position-history read but leaves persistence injected. Without a canonical adapter, an Orgmetra deployment cannot obtain that bounded history from normalized `position_record` and `position_record_version` truth without bespoke host code. + +## Requirement-to-evidence matrix + +| Requirement | Production boundary | Regression | +| --- | --- | --- | +| No DB access on invalid input | exact tenant/Position UUID and built-in UTC `known_at` validation before `connection_factory()` | invalid UUID/time cases assert zero connection calls | +| Stable request identity | tenant/Position UUIDs are reduced to built-in integer scalar authority before connection acquisition and fresh UUID views are used for DB parameters | a connection factory that mutates retained caller UUID aliases cannot change tenant GUC or SELECT target | +| Stable DB capability | `connection_factory` is held in tuple payload, not a writable instance slot | `object.__setattr__` cannot replace the accepted capability | +| Real transaction context | acquired connection must expose exact `autocommit is False` before `cursor()` | `True`, missing/`None`, numeric `0`, and string `"false"` modes fail before cursor access | +| Database cannot mutate HR truth | only after non-autocommit proof, execute `SET TRANSACTION ISOLATION LEVEL READ COMMITTED, READ ONLY` | execution-order and autocommit regressions | +| Tenant defense in depth | transaction-local `pg_catalog.set_config('orgmetra.tenant_record_id', ..., true)` before SELECT in the same proven transaction context | exact SQL and detached tenant parameter assertion | +| Explicit Position scope | fully qualified join between `public.position_record_version` and `public.position_record` with tenant/Position predicates | SQL contract assertions | +| Preserve system knowledge | half-open parent/version `recorded_from`/`recorded_to` predicates at `known_at` | future and closed-at-cutoff rows fail closed | +| Preserve business history | no effective-date filter; deterministic effective start/version ordering | returned typed dates and SQL ordering assertion | +| Canonical UTC | `AT TIME ZONE 'UTC'` projection and exact naive DB timestamp validation | string/aware/non-datetime timestamp regressions | +| Untrusted DB-API boundary | exact list result, exact tuple row shape, parent record reconstruction | malformed collection/row/value regressions | +| Immutable typed result | tuple of scalar-backed `PositionHistoryRecord` values | empty/non-empty result and parent retained-authority regressions | +| Parent authority remains single owner | adapter accepts no purpose or authorization input | PR #152 performs authorization and service revalidation | +| Repository acceptance owner | consolidated Foundation CI after descendant reaches a protected-parent/`develop` integration lane | feature-local pre-consolidation workflow is absent from the current stack | + +## Test-first and restack chain + +1. **Original contract-only child head:** `bf93924e` added adapter regressions while `orgmetra_people_api.postgres_position_history` was absent. +2. **Original implementation lineage:** the pre-consolidation branch added the adapter, tests, ADR/doctoring/traceability, and a feature-local quality workflow. Historical local/full-suite and isolated PostgreSQL results remain development evidence only and are not transferred to the current exact head. +3. **Parent reconciliation:** after #152 advanced to `616ed8c8a410e4dcdf4c31d293dfb082f7ce8297`, ordinary two-parent commit `a8edc7d2fa69842d4515def5c8a3e710ed4b4e2b` restacked #153 on that owner head without force push. The stale `.github/workflows/position-history-postgres-read-quality.yml` was deliberately not overlaid because protected #161 consolidated repository acceptance under Foundation CI. +4. **Retained-authority/transaction test-only head:** `800b783a74594a59f4c0d5dd819a6fce4fced4b8` adds regressions proving immutable connection capability, exact non-autocommit mode, rejection of unproven transaction modes, and detached request UUID authority before connection acquisition. +5. **Production repair:** `00d8f58d0863ef0d3270c7197d41856de3131e34` changes the adapter to a tuple-backed capability holder, snapshots request UUIDs to integer scalars, requires `autocommit is False` before cursor creation, uses fresh detached UUID views for tenant/query parameters, and compares returned scalar identity to the request snapshot. +6. **Fixture contract:** `d0a8b9b9b5e43bbc522b97c921a1d67c51b5fb6c` makes the normal DB-API fixture explicitly non-autocommit rather than relying on an unspecified connection mode. +7. **ADR currentization:** `6d0e185fdbce6d3d5b6a49c70353bcb0809830c2` records transaction proof, immutable capability ownership, UUID detachment, and consolidated workflow ownership. + +No hosted RED or GREEN is claimed for the new stacked heads merely because PR-triggered Foundation runs are absent. Once its parent lineage is protected and #153 can target `develop`, the exact final child head must reacquire Foundation/Security/SAST/CodeQL and qualifying independent review. Parent, predecessor, manual, or historical isolated-PostgreSQL evidence does not transfer. + +## Security and data boundary + +The adapter reads only Position anchor lineage and Position-version fields. It does not join Person, Employment, Assignment, compensation, candidate, performance, credential, prompt, or model-output data. Purpose-bound authorization-before-retrieval remains in the parent service; the adapter performs no mutation, audit/outbox write, or high-impact employment decision. + +The database transaction is intentionally narrow: connection acquisition → explicit non-autocommit proof → read-only/tenant-context setup → bounded SELECT/fetch → context exit. Authorization and other potentially long-running work do not execute while this adapter transaction is open. + +## Out of scope + +- Position-history HTTP/presentation integration (#154). +- Position mutation or correction workflows. +- Assignment/Employment history joins. +- Database migrations; the protected schema already owns these relations and RLS policies. +- Release, tag, publication, or protected-default-branch authority. + +Any later descendant must preserve these boundaries and consume this adapter through its owner stack rather than copying its source. diff --git a/services/people-api/src/orgmetra_people_api/__init__.py b/services/people-api/src/orgmetra_people_api/__init__.py index 3313e7f4f..20c9908e6 100644 --- a/services/people-api/src/orgmetra_people_api/__init__.py +++ b/services/people-api/src/orgmetra_people_api/__init__.py @@ -51,6 +51,7 @@ from orgmetra_people_api.postgres import PostgresPeopleReadPort from orgmetra_people_api.postgres_hire import PostgresHireAcceptancePort from orgmetra_people_api.postgres_mutations import PostgresPeopleMutationPort +from orgmetra_people_api.postgres_position_history import PostgresPositionHistoryReadPort __all__ = [ "AuthenticatedPrincipal", @@ -80,6 +81,7 @@ "PostgresHireAcceptancePort", "PostgresPeopleMutationPort", "PostgresPeopleReadPort", + "PostgresPositionHistoryReadPort", "AssignmentMutationCommand", "AssignmentMutationResult", "EmploymentMutationCommand", diff --git a/services/people-api/src/orgmetra_people_api/postgres_position_history.py b/services/people-api/src/orgmetra_people_api/postgres_position_history.py new file mode 100644 index 000000000..4225bade8 --- /dev/null +++ b/services/people-api/src/orgmetra_people_api/postgres_position_history.py @@ -0,0 +1,207 @@ +"""PostgreSQL adapter for purpose-bound Position-history reads. + +The parent People service owns purpose-bound authorization. This adapter owns +only a read-only, tenant-scoped projection of canonical Orgmetra Position and +Position-version facts, returning typed rows for the parent service to +revalidate before disclosure. +""" + +from __future__ import annotations + +from contextlib import AbstractContextManager +from datetime import datetime, timedelta, timezone +from typing import Any, Callable +from uuid import UUID + +from orgmetra_people_api.position_history import ( + PositionHistoryIntegrityError, + PositionHistoryRecord, +) + +PostgresConnectionFactory = Callable[[], AbstractContextManager[Any]] + +_READ_ONLY_SQL = "SET TRANSACTION ISOLATION LEVEL READ COMMITTED, READ ONLY" +_TENANT_CONTEXT_SQL = "SELECT pg_catalog.set_config('orgmetra.tenant_record_id', %s, true)" +_POSITION_HISTORY_SQL = """ +SELECT + position_version.tenant_record_id, + position_version.position_record_id, + position_version.position_record_version_id, + position_anchor.organization_unit_id, + position_anchor.job_profile_id, + position_version.position_status_code, + position_version.effective_from, + position_version.effective_to, + position_version.recorded_from AT TIME ZONE 'UTC' AS recorded_from_utc, + position_version.recorded_to AT TIME ZONE 'UTC' AS recorded_to_utc +FROM public.position_record_version AS position_version +JOIN public.position_record AS position_anchor + ON position_anchor.tenant_record_id = position_version.tenant_record_id + AND position_anchor.position_record_id = position_version.position_record_id +WHERE position_version.tenant_record_id = %s + AND position_version.position_record_id = %s + AND position_anchor.recorded_from <= %s + AND (position_anchor.recorded_to IS NULL OR %s < position_anchor.recorded_to) + AND position_version.recorded_from <= %s + AND (position_version.recorded_to IS NULL OR %s < position_version.recorded_to) +ORDER BY position_version.effective_from, position_version.position_record_version_id +""".strip() +_MAX_UUID_INT = (1 << 128) - 1 + + +def _operational_uuid_scalar(field_name: str, value: object) -> int: + """Detach exact UUID input into immutable scalar authority before DB access.""" + if type(value) is not UUID: + raise ValueError(f"{field_name} must be an operational UUID.") + scalar = value.int + if type(scalar) is not int or not 0 < scalar < _MAX_UUID_INT: + raise ValueError(f"{field_name} must be an operational UUID.") + return scalar + + +def _require_utc_instant(field_name: str, value: object) -> None: + """Require exact built-in UTC time before using it as a history cutoff.""" + if type(value) is not datetime: + raise ValueError(f"{field_name} must be a timezone-aware UTC datetime.") + if type(value.tzinfo) is not timezone: + raise ValueError(f"{field_name} must be a timezone-aware UTC datetime.") + if value.utcoffset() != timedelta(0): + raise ValueError(f"{field_name} must be a timezone-aware UTC datetime.") + + +def _db_utc_instant(value: object) -> datetime: + """Attach built-in UTC only to PostgreSQL's explicit naive UTC projection.""" + if type(value) is not datetime or value.tzinfo is not None: + raise PositionHistoryIntegrityError( + "database recorded time must be a naive UTC projection" + ) + return value.replace(tzinfo=timezone.utc) + + +def _record_from_row(row: object) -> PositionHistoryRecord: + """Convert one untrusted DB-API row into the parent governed record type.""" + if type(row) is not tuple or len(row) != 10: + raise PositionHistoryIntegrityError("database Position-history row has an invalid shape") + ( + tenant_record_id, + position_record_id, + position_record_version_id, + organization_unit_id, + job_profile_id, + position_status_code, + effective_from, + effective_to, + recorded_from, + recorded_to, + ) = row + try: + return PositionHistoryRecord( + tenant_record_id=tenant_record_id, + position_record_id=position_record_id, + position_record_version_id=position_record_version_id, + organization_unit_id=organization_unit_id, + job_profile_id=job_profile_id, + position_status_code=position_status_code, + effective_from=effective_from, + effective_to=effective_to, + recorded_from=_db_utc_instant(recorded_from), + recorded_to=None if recorded_to is None else _db_utc_instant(recorded_to), + ) + except ValueError as exc: + raise PositionHistoryIntegrityError( + "database Position-history row failed integrity" + ) from exc + + +class PostgresPositionHistoryReadPort(tuple): + """Read canonical Position history through one proven non-autocommit transaction. + + The validated connection capability is stored in tuple payload rather than a + writable instance slot. Request UUIDs are detached before connection + acquisition, and transaction-local tenant context is established only after + the connection proves ``autocommit is False``. + """ + + __slots__ = () + + def __new__( + cls, + connection_factory: PostgresConnectionFactory, + ) -> PostgresPositionHistoryReadPort: + if not callable(connection_factory): + raise TypeError("connection_factory must be callable") + return tuple.__new__(cls, (connection_factory,)) + + @property + def connection_factory(self) -> PostgresConnectionFactory: + """Return the structurally bound connection capability.""" + return tuple.__getitem__(self, 0) + + def read_position_history( + self, + *, + tenant_record_id: UUID, + position_record_id: UUID, + known_at: datetime, + ) -> tuple[PositionHistoryRecord, ...]: + """Return Position versions visible at ``known_at`` without authorizing disclosure.""" + tenant_record_id_scalar = _operational_uuid_scalar( + "tenant_record_id", + tenant_record_id, + ) + position_record_id_scalar = _operational_uuid_scalar( + "position_record_id", + position_record_id, + ) + _require_utc_instant("known_at", known_at) + + trusted_tenant_record_id = UUID(int=tenant_record_id_scalar) + trusted_position_record_id = UUID(int=position_record_id_scalar) + connection_factory = tuple.__getitem__(self, 0) + with connection_factory() as connection: + if getattr(connection, "autocommit", None) is not False: + raise ValueError( + "connection autocommit must be explicitly False for Position-history reads" + ) + with connection.cursor() as cursor: + cursor.execute(_READ_ONLY_SQL) + cursor.execute( + _TENANT_CONTEXT_SQL, + (str(trusted_tenant_record_id),), + ) + cursor.execute( + _POSITION_HISTORY_SQL, + ( + trusted_tenant_record_id, + trusted_position_record_id, + known_at, + known_at, + known_at, + known_at, + ), + ) + rows = cursor.fetchall() + + if type(rows) is not list: + raise PositionHistoryIntegrityError( + "database Position-history read must return the default list row collection" + ) + + records: list[PositionHistoryRecord] = [] + for row in rows: + record = _record_from_row(row) + if ( + record.tenant_record_id_scalar != tenant_record_id_scalar + or record.position_record_id_scalar != position_record_id_scalar + ): + raise PositionHistoryIntegrityError( + "database Position-history row does not match the requested target" + ) + if record.recorded_from > known_at or ( + record.recorded_to is not None and known_at >= record.recorded_to + ): + raise PositionHistoryIntegrityError( + "database Position-history row is not visible at the requested knowledge cutoff" + ) + records.append(record) + return tuple(records) diff --git a/services/people-api/tests/test_postgres_position_history.py b/services/people-api/tests/test_postgres_position_history.py new file mode 100644 index 000000000..8452075fc --- /dev/null +++ b/services/people-api/tests/test_postgres_position_history.py @@ -0,0 +1,328 @@ +"""Regression contract for the PostgreSQL Position-history read adapter.""" + +from __future__ import annotations + +from contextlib import AbstractContextManager +from datetime import date, datetime, timedelta, timezone, tzinfo +from uuid import UUID + +import pytest + +from orgmetra_people_api.position_history import PositionHistoryIntegrityError +from orgmetra_people_api.postgres_position_history import PostgresPositionHistoryReadPort + + +TENANT_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c1") +POSITION_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c2") +VERSION_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c3") +ORGANIZATION_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c4") +JOB_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c5") +KNOWN_AT = datetime(2026, 8, 30, 0, 0, tzinfo=timezone.utc) + + +class ForgedUUID(UUID): + """Prove caller-controlled UUID subclasses cannot cross the adapter boundary.""" + + +class ZeroOffsetProvider(tzinfo): + """Prove a caller-controlled zero-offset timezone is not canonical UTC.""" + + def utcoffset(self, dt: datetime | None) -> timedelta: + return timedelta(0) + + def dst(self, dt: datetime | None) -> timedelta: + return timedelta(0) + + +def position_row( + *, + tenant_record_id: object = TENANT_ID, + position_record_id: object = POSITION_ID, + position_record_version_id: object = VERSION_ID, + organization_unit_id: object = ORGANIZATION_ID, + job_profile_id: object = JOB_ID, + position_status_code: object = "active", + effective_from: object = date(2026, 1, 1), + effective_to: object = date(2026, 7, 1), + recorded_from: object = datetime(2026, 8, 1), + recorded_to: object = None, +) -> tuple[object, ...]: + """Return one default DB row projected by the governed SQL query.""" + return ( + tenant_record_id, + position_record_id, + position_record_version_id, + organization_unit_id, + job_profile_id, + position_status_code, + effective_from, + effective_to, + recorded_from, + recorded_to, + ) + + +class FakeCursor(AbstractContextManager["FakeCursor"]): + """Minimal DB-API cursor that records SQL and returns configured rows.""" + + def __init__(self, rows: object) -> None: + self.rows = rows + self.executions: list[tuple[str, object | None]] = [] + + def __enter__(self) -> "FakeCursor": + return self + + def __exit__(self, exc_type: object, exc: object, traceback: object) -> None: + return None + + def execute(self, statement: str, parameters: object | None = None) -> None: + self.executions.append((statement, parameters)) + + def fetchall(self) -> object: + return self.rows + + +class FakeConnection(AbstractContextManager["FakeConnection"]): + """Minimal proven non-autocommit connection exposing one stable cursor.""" + + def __init__(self, cursor: FakeCursor) -> None: + self._cursor = cursor + self.autocommit = False + + def __enter__(self) -> "FakeConnection": + return self + + def __exit__(self, exc_type: object, exc: object, traceback: object) -> None: + return None + + def cursor(self) -> FakeCursor: + return self._cursor + + +class ConnectionFactory: + """Count connection acquisition so invalid inputs prove zero database access.""" + + def __init__(self, rows: object) -> None: + self.calls = 0 + self.cursor = FakeCursor(rows) + + def __call__(self) -> FakeConnection: + self.calls += 1 + return FakeConnection(self.cursor) + + +@pytest.mark.parametrize("invalid_factory", [None, 7, "connection"]) +def test_rejects_non_callable_connection_factory(invalid_factory: object) -> None: + with pytest.raises(TypeError, match="connection_factory must be callable"): + PostgresPositionHistoryReadPort(invalid_factory) # type: ignore[arg-type] + + +def test_read_is_tenant_scoped_read_only_bitemporal_and_typed() -> None: + factory = ConnectionFactory( + [position_row(recorded_to=datetime(2026, 9, 1))] + ) + port = PostgresPositionHistoryReadPort(factory) + + records = port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + assert len(records) == 1 + record = records[0] + assert record.tenant_record_id == TENANT_ID + assert record.position_record_id == POSITION_ID + assert record.position_record_version_id == VERSION_ID + assert record.organization_unit_id == ORGANIZATION_ID + assert record.job_profile_id == JOB_ID + assert record.position_status_code == "active" + assert record.effective_from == date(2026, 1, 1) + assert record.effective_to == date(2026, 7, 1) + assert record.recorded_from == datetime(2026, 8, 1, tzinfo=timezone.utc) + assert record.recorded_to == datetime(2026, 9, 1, tzinfo=timezone.utc) + assert factory.calls == 1 + + assert len(factory.cursor.executions) == 3 + transaction_sql, transaction_parameters = factory.cursor.executions[0] + tenant_sql, tenant_parameters = factory.cursor.executions[1] + history_sql, history_parameters = factory.cursor.executions[2] + assert transaction_sql == "SET TRANSACTION ISOLATION LEVEL READ COMMITTED, READ ONLY" + assert transaction_parameters is None + assert "pg_catalog.set_config('orgmetra.tenant_record_id', %s, true)" in tenant_sql + assert tenant_parameters == (str(TENANT_ID),) + assert "FROM public.position_record_version AS position_version" in history_sql + assert "JOIN public.position_record AS position_anchor" in history_sql + assert "position_version.tenant_record_id = %s" in history_sql + assert "position_version.position_record_id = %s" in history_sql + assert "position_anchor.recorded_from <= %s" in history_sql + assert "%s < position_anchor.recorded_to" in history_sql + assert "position_version.recorded_from <= %s" in history_sql + assert "%s < position_version.recorded_to" in history_sql + assert "AT TIME ZONE 'UTC'" in history_sql + assert "ORDER BY position_version.effective_from, position_version.position_record_version_id" in history_sql + assert "SELECT *" not in history_sql.upper() + assert history_parameters == ( + TENANT_ID, + POSITION_ID, + KNOWN_AT, + KNOWN_AT, + KNOWN_AT, + KNOWN_AT, + ) + + +def test_empty_database_result_returns_immutable_empty_tuple() -> None: + factory = ConnectionFactory([]) + port = PostgresPositionHistoryReadPort(factory) + + assert port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) == () + + +@pytest.mark.parametrize( + ("tenant_record_id", "position_record_id", "known_at"), + [ + ("not-a-uuid", POSITION_ID, KNOWN_AT), + (ForgedUUID(str(TENANT_ID)), POSITION_ID, KNOWN_AT), + (UUID(int=0), POSITION_ID, KNOWN_AT), + (TENANT_ID, ForgedUUID(str(POSITION_ID)), KNOWN_AT), + (TENANT_ID, UUID(int=(1 << 128) - 1), KNOWN_AT), + (TENANT_ID, POSITION_ID, "2026-08-30"), + (TENANT_ID, POSITION_ID, datetime(2026, 8, 30)), + (TENANT_ID, POSITION_ID, datetime(2026, 8, 30, tzinfo=timezone(timedelta(hours=9)))), + (TENANT_ID, POSITION_ID, datetime(2026, 8, 30, tzinfo=ZeroOffsetProvider())), + ], +) +def test_invalid_request_identity_or_time_fails_before_database_access( + tenant_record_id: object, + position_record_id: object, + known_at: object, +) -> None: + factory = ConnectionFactory([]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(ValueError): + port.read_position_history( # type: ignore[arg-type] + tenant_record_id=tenant_record_id, + position_record_id=position_record_id, + known_at=known_at, + ) + + assert factory.calls == 0 + + +def test_rejects_non_default_fetchall_collection() -> None: + factory = ConnectionFactory((position_row(),)) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="default list row collection"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +@pytest.mark.parametrize("row", [[1] * 10, (1, 2)]) +def test_rejects_unsupported_row_container_or_shape(row: object) -> None: + factory = ConnectionFactory([row]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="row has an invalid shape"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +@pytest.mark.parametrize( + ("recorded_from", "recorded_to"), + [ + ("2026-08-01", None), + (datetime(2026, 8, 1, tzinfo=timezone.utc), None), + (datetime(2026, 8, 1), "2026-09-01"), + (datetime(2026, 8, 1), datetime(2026, 9, 1, tzinfo=timezone.utc)), + ], +) +def test_rejects_noncanonical_database_timestamp_projection( + recorded_from: object, + recorded_to: object, +) -> None: + factory = ConnectionFactory([position_row(recorded_from=recorded_from, recorded_to=recorded_to)]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="database recorded time must be a naive UTC projection"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +def test_rejects_database_row_that_fails_position_record_integrity() -> None: + factory = ConnectionFactory([position_row(position_status_code="NOT_CANONICAL")]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="database Position-history row failed integrity"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +@pytest.mark.parametrize( + "row", + [ + position_row(tenant_record_id=UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9d1")), + position_row(position_record_id=UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9d2")), + ], +) +def test_rejects_row_outside_requested_tenant_or_position(row: tuple[object, ...]) -> None: + factory = ConnectionFactory([row]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="does not match the requested target"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +@pytest.mark.parametrize( + "row", + [ + position_row(recorded_from=datetime(2026, 8, 31)), + position_row(recorded_to=datetime(2026, 8, 30)), + ], +) +def test_rejects_row_outside_requested_system_knowledge_cutoff(row: tuple[object, ...]) -> None: + factory = ConnectionFactory([row]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="not visible at the requested knowledge cutoff"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +def test_open_recorded_interval_is_visible_at_known_at() -> None: + factory = ConnectionFactory([position_row()]) + port = PostgresPositionHistoryReadPort(factory) + + records = port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + assert records[0].recorded_to is None + assert isinstance(records, tuple) diff --git a/services/people-api/tests/test_postgres_position_history_integrity_hardening.py b/services/people-api/tests/test_postgres_position_history_integrity_hardening.py new file mode 100644 index 000000000..2da4f2b11 --- /dev/null +++ b/services/people-api/tests/test_postgres_position_history_integrity_hardening.py @@ -0,0 +1,120 @@ +"""Regression contracts for Position-history PostgreSQL trust boundaries.""" + +from __future__ import annotations + +from contextlib import AbstractContextManager +from datetime import datetime, timezone +from uuid import UUID + +import pytest + +from orgmetra_people_api.postgres_position_history import PostgresPositionHistoryReadPort + +TENANT = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c1") +OTHER_TENANT = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9d1") +POSITION = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c2") +OTHER_POSITION = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9d2") +KNOWN_AT = datetime(2026, 8, 30, 0, 0, tzinfo=timezone.utc) + + +class Cursor(AbstractContextManager["Cursor"]): + def __init__(self) -> None: + self.executions: list[tuple[str, object | None]] = [] + + def __enter__(self) -> "Cursor": + return self + + def __exit__(self, exc_type: object, exc: object, traceback: object) -> None: + return None + + def execute(self, statement: str, parameters: object | None = None) -> None: + self.executions.append((statement, parameters)) + + def fetchall(self) -> list[object]: + return [] + + +class Connection(AbstractContextManager["Connection"]): + def __init__(self, *, autocommit: object = False) -> None: + self.autocommit = autocommit + self.cursor_calls = 0 + self.cursor_instance = Cursor() + + def __enter__(self) -> "Connection": + return self + + def __exit__(self, exc_type: object, exc: object, traceback: object) -> None: + return None + + def cursor(self) -> Cursor: + self.cursor_calls += 1 + return self.cursor_instance + + +class MutatingFactory: + def __init__(self, tenant_alias: UUID, position_alias: UUID) -> None: + self.tenant_alias = tenant_alias + self.position_alias = position_alias + self.connection = Connection() + + def __call__(self) -> Connection: + object.__setattr__(self.tenant_alias, "int", OTHER_TENANT.int) + object.__setattr__(self.position_alias, "int", OTHER_POSITION.int) + return self.connection + + +def test_connection_factory_capability_cannot_be_replaced_after_validation() -> None: + accepted = lambda: Connection() + replacement = lambda: Connection() + port = PostgresPositionHistoryReadPort(accepted) + + with pytest.raises(AttributeError): + object.__setattr__(port, "connection_factory", replacement) + + assert port.connection_factory is accepted + + +def test_autocommit_connection_fails_before_cursor_access() -> None: + connection = Connection(autocommit=True) + port = PostgresPositionHistoryReadPort(lambda: connection) + + with pytest.raises(ValueError, match="autocommit"): + port.read_position_history( + tenant_record_id=TENANT, + position_record_id=POSITION, + known_at=KNOWN_AT, + ) + + assert connection.cursor_calls == 0 + + +@pytest.mark.parametrize("unproven_mode", [None, 0, "false"]) +def test_unproven_transaction_mode_fails_before_cursor_access(unproven_mode: object) -> None: + connection = Connection(autocommit=unproven_mode) + port = PostgresPositionHistoryReadPort(lambda: connection) + + with pytest.raises(ValueError, match="autocommit"): + port.read_position_history( + tenant_record_id=TENANT, + position_record_id=POSITION, + known_at=KNOWN_AT, + ) + + assert connection.cursor_calls == 0 + + +def test_request_uuid_aliases_are_detached_before_connection_acquisition() -> None: + tenant_alias = UUID(str(TENANT)) + position_alias = UUID(str(POSITION)) + factory = MutatingFactory(tenant_alias, position_alias) + port = PostgresPositionHistoryReadPort(factory) + + assert port.read_position_history( + tenant_record_id=tenant_alias, + position_record_id=position_alias, + known_at=KNOWN_AT, + ) == () + + executions = factory.connection.cursor_instance.executions + assert executions[1][1] == (str(TENANT),) + assert executions[2][1][:2] == (TENANT, POSITION)