"""Pipeline state, receipt and pending-work failures use actual PostgreSQL transactions.""" from dataclasses import replace import os from typing import cast import psycopg from psycopg import Connection from psycopg.types.json import Jsonb import pytest from metallo.adaptors import codec from metallo.adaptors.postgres import Database, Row from metallo.contexts.pipeline_coordination.adaptors.postgres import PipelineQueries, PostgresPipelineCommands from metallo.contexts.pipeline_coordination.application.commands import ReadPipelineQuery from metallo.contexts.pipeline_coordination.domain.pipeline import Pipeline from metallo.foundation.domain import CorruptState, record from metallo.foundation.identity import new_id from metallo.foundation.pagination import PageRequest from metallo.tests.infrastructure.pipeline_fixtures import ( Authority, Workloads, cancellation, command, counts, database as database, forget_damaged, operations, save, work, ) from metallo.tests.pipeline_coordination.application_fixtures import TEST_LIFETIME, accepted @pytest.mark.parametrize("operation", ["start", "cancel"]) def test_pipeline_root_receipt_and_work_roll_back_after_all_three_writes( database: Database, monkeypatch: pytest.MonkeyPatch, operation: str) -> None: if operation == "start": request = command() handler = operations(database, Authority(), Workloads(request)) initial: Pipeline | None = None marker: Row | None = None else: saved = save(database) request, handler = saved.request, saved.handler initial = accepted(request, TEST_LIFETIME) marker = work(database, request.pipeline_id) previous = counts(database, request.pipeline_id) writer = cast(PostgresPipelineCommands, handler.commands) original = writer._save_work def broken(connection: Connection[Row], pipeline: Pipeline, *, creating: bool) -> None: original(connection, pipeline, creating=creating) rows = connection.execute("""SELECT (SELECT version FROM pharos.aggregates WHERE kind='pipeline' AND id=%s) AS version, (SELECT count(*) FROM pharos.command_receipts WHERE kind='pipeline' AND aggregate_id=%s) AS receipts, (SELECT pipeline_version FROM pharos.pipeline_work WHERE pipeline_id=%s) AS work_version """, (request.pipeline_id,) * 3).fetchone() assert rows == {"version": 1 if creating else 2, "receipts": 1 if creating else 2, "work_version": 1 if creating else 2} raise RuntimeError("Injected failure after Pipeline root, receipt and work") cancel = cancellation(request) with monkeypatch.context() as fault: fault.setattr(writer, "_save_work", broken) with pytest.raises(RuntimeError, match="Injected"): if operation == "start": handler.start(request) else: handler.cancel(cancel) assert counts(database, request.pipeline_id) == previous if initial is None: assert PipelineQueries(database).get(request.pipeline_id) is None assert handler.start(request)["status"] == "waiting" else: assert handler.read(ReadPipelineQuery(request.pipeline_id, new_id())) == initial assert work(database, request.pipeline_id) == marker assert handler.cancel(cancel)["status"] == "cancellation_requested" def test_pipeline_missing_projection_registration_rolls_back_without_implicit_frontend_delivery( database: Database, monkeypatch: pytest.MonkeyPatch) -> None: request = command() handler = operations(database, Authority(), Workloads(request)) with monkeypatch.context() as fault: fault.delitem(codec.PROJECTIONS, "pipeline_coordination/pipeline") with pytest.raises(KeyError): handler.start(request) assert all(value == 0 for value in counts(database, request.pipeline_id).values()) assert handler.start(request)["status"] == "waiting" assert counts(database, request.pipeline_id)["realtime"] == 0 @pytest.mark.parametrize("damage", ["identity", "member", "project", "step", "environment", "status", "header-version"]) def test_pipeline_corrupt_snapshot_or_header_cannot_read_replay_or_cancel(database: Database, damage: str) -> None: saved = save(database) request = saved.request snapshot = request.to_pipeline().to_mapping() snapshot["version"] = 2 if damage == "identity": snapshot["id"] = new_id() elif damage in ("member", "project"): snapshot[damage + "Id"] = "legacy-prefixed-id" elif damage in ("step", "environment"): steps = snapshot["steps"] assert isinstance(steps, list) first = record(steps[0]) first["id" if damage == "step" else "environment"] = "invalid" steps[0] = first elif damage == "status": snapshot["status"] = "running" else: snapshot["version"] = 1 with psycopg.connect(os.environ["PIPELINE_COORDINATION_ADMIN_URL"]) as connection: connection.execute("SELECT set_config('pharos.command_target',%s,true)", ("pipeline:" + request.pipeline_id,)) connection.execute("UPDATE pharos.aggregates SET version=version+1,state=%s WHERE kind='pipeline' AND id=%s", (Jsonb(snapshot), request.pipeline_id)) connection.execute("UPDATE pharos.pipeline_work SET pipeline_version=2 WHERE pipeline_id=%s", (request.pipeline_id,)) try: before, marker = counts(database, request.pipeline_id), work(database, request.pipeline_id) queries = PipelineQueries(database) for read in (lambda: saved.handler.read(ReadPipelineQuery(request.pipeline_id, new_id())), lambda: queries.get(request.pipeline_id)): with pytest.raises(CorruptState): read() with pytest.raises(CorruptState): saved.handler.start(request) with pytest.raises(CorruptState): saved.handler.cancel(cancellation(request, expected_version=2)) assert counts(database, request.pipeline_id) == before and work(database, request.pipeline_id) == marker finally: forget_damaged(request.pipeline_id) @pytest.mark.parametrize("damage", ["missing", "version"]) def test_pipeline_commands_reject_missing_or_outdated_pending_work_even_on_replay( database: Database, damage: str) -> None: saved = save(database) request = saved.request with psycopg.connect(os.environ["PIPELINE_COORDINATION_ADMIN_URL"]) as connection: if damage == "missing": connection.execute("DELETE FROM pharos.pipeline_work WHERE pipeline_id=%s", (request.pipeline_id,)) else: connection.execute("UPDATE pharos.pipeline_work SET pipeline_version=2 WHERE pipeline_id=%s", (request.pipeline_id,)) before = counts(database, request.pipeline_id) with pytest.raises(CorruptState): saved.handler.start(request) with pytest.raises(CorruptState): saved.handler.cancel(cancellation(request)) assert counts(database, request.pipeline_id) == before def test_pipeline_runtime_role_cannot_change_schema_or_delete_its_work(database: Database) -> None: saved = save(database) for query, parameters in ( ("CREATE TABLE pharos.forbidden_pipeline_ddl(id uuid)", ()), ("DELETE FROM pharos.pipeline_work WHERE pipeline_id=%s", (saved.request.pipeline_id,)), ): with database.pool.connection() as connection, pytest.raises(psycopg.errors.InsufficientPrivilege): connection.execute(query, parameters) assert work(database, saved.request.pipeline_id)["pipeline_version"] == 1 def test_pipeline_owner_lookups_happen_outside_a_database_transaction(database: Database) -> None: request = command() def check_idle() -> None: with database.pool.connection() as connection: row = connection.execute("""SELECT count(*) AS count FROM pg_stat_activity WHERE datname=current_database() AND usename=current_user AND pid<>pg_backend_pid() AND state='idle in transaction'""").fetchone() assert row == {"count": 0} workloads, authority = Workloads(request), Authority() workloads.before_get = check_idle authority.before_check = check_idle assert operations(database, authority, workloads).start(request)["status"] == "waiting" def test_pipeline_queries_validate_header_version_in_targeted_paged_reads(database: Database) -> None: template = command() prefix = template.pipeline_id[:-12] request = replace(template, pipeline_id=prefix + "000000000002") assert operations(database, Authority(), Workloads(request)).start(request)["status"] == "waiting" with psycopg.connect(os.environ["PIPELINE_COORDINATION_ADMIN_URL"]) as connection: connection.execute("SELECT set_config('pharos.command_target',%s,true)", ("pipeline:" + request.pipeline_id,)) connection.execute("UPDATE pharos.aggregates SET version=version+1 WHERE kind='pipeline' AND id=%s", (request.pipeline_id,)) queries = PipelineQueries(database) try: with pytest.raises(CorruptState): queries.page(PageRequest(limit=1, after=prefix + "000000000001")) finally: forget_damaged(request.pipeline_id)