diff --git a/TESTING_STRATEGY.md b/TESTING_STRATEGY.md index f841d94..6e7c317 100644 --- a/TESTING_STRATEGY.md +++ b/TESTING_STRATEGY.md @@ -473,6 +473,12 @@ is unavailable, reads source-derived facility rows from the hosted source-derive table group when a database-backed context is supplied, keeps request queue and reviewer detail on source-derived plus reviewer-created route seams, and keeps fixture-demo behavior isolated behind explicit local/demo configuration. +Hosted PostgreSQL connection-lifecycle tests must also prove that health polls +and hosted route requests reuse one process-owned engine, release every +request-owned connection after success and failure, roll back failed or +unfinished transactions before release, and remain bounded under repeated +calls. A POST validation or early-return path that performs a read must not +leave a connection idle in transaction. Hosted CCLD import/reload tests must prove local validated artifacts are validated before load, source URL/raw SHA-256/raw path/connector traceability is preserved, existing source-derived keys are refreshed without duplicates, diff --git a/src/ccld_complaints/hosted_app/app.py b/src/ccld_complaints/hosted_app/app.py index 93503d6..136464e 100644 --- a/src/ccld_complaints/hosted_app/app.py +++ b/src/ccld_complaints/hosted_app/app.py @@ -7,14 +7,18 @@ import html import json import os +from collections.abc import Iterator +from contextlib import ExitStack, contextmanager from dataclasses import dataclass from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path +from threading import Lock from typing import Any from urllib.parse import parse_qs, parse_qsl, urlencode, urlparse from sqlalchemy import create_engine, text +from sqlalchemy.engine import Engine from sqlalchemy.exc import SQLAlchemyError from ccld_complaints.hosted_app.audit_coverage_plan import ( @@ -170,7 +174,8 @@ ) FAVICON_FILE = Path(__file__).resolve().parent / "static" / "favicon.ico" SOURCE_COVERAGE_FACILITY_FIXTURE = "chhs_facility_master_tiny.csv" -_DEFAULT_POSTGRES_REVIEWER_UI_CONTEXT: ReviewerUiContext | None = None +_DEFAULT_POSTGRES_ENGINE: Engine | None = None +_DEFAULT_POSTGRES_ENGINE_LOCK = Lock() @dataclass(frozen=True) @@ -483,10 +488,9 @@ def format_host(host: object) -> str: def _check_source_data_loaded() -> bool: """Return True if at least one row exists in hosted_source_derived_records.""" try: - config = load_hosted_database_config() - if config.database_url is None: + engine = _default_postgres_engine() + if engine is None: return False - engine = create_engine(config.database_url) with engine.connect() as conn: result = conn.execute( text("SELECT COUNT(*) FROM hosted_source_derived_records") @@ -1102,9 +1106,80 @@ def route_response( reviewer_ui_context: ReviewerUiContext | None = None, ccld_record_request_ui_context: CcldRecordRequestUiContext | None = None, operator_coverage_context: OperatorCoverageDashboardContext | None = None, + _postgres_context_resolved: bool = False, ) -> tuple[int, str, bytes]: parsed_url = urlparse(path) parsed_path = parsed_url.path + active_page_data_mode = _active_page_data_mode(page_data_mode) + needs_postgres_context = ( + not _postgres_context_resolved + and active_page_data_mode == POSTGRES_PAGE_DATA_MODE + and ( + parsed_path == "/" + or parsed_path.startswith(CCLD_UI_PREFIX) + or parsed_path.startswith(REVIEWER_UI_PREFIX) + ) + ) + if needs_postgres_context: + active_auth_config = _active_auth_runtime_config(auth_runtime_config) + with ExitStack() as request_resources: + active_reviewer_context = reviewer_ui_context + if ( + active_reviewer_context is None + and ccld_record_request_ui_context is not None + ): + active_reviewer_context = ( + ccld_record_request_ui_context.reviewer_ui_context + ) + if active_reviewer_context is None: + active_reviewer_context = request_resources.enter_context( + _default_postgres_reviewer_context() + ) + + active_ccld_context = ccld_record_request_ui_context + if ( + active_ccld_context is None + and active_reviewer_context is not None + and (parsed_path == "/" or parsed_path.startswith(CCLD_UI_PREFIX)) + ): + active_ccld_context = ccld_record_request_context_for_reviewer_context( + active_reviewer_context, + retrieval_context=_default_retrieval_context_for_reviewer_context( + active_reviewer_context, + active_auth_config, + ), + ) + + return route_response( + path, + method=method, + request_body=request_body, + request_headers=request_headers, + audit_coverage_plan_context=audit_coverage_plan_context, + auth_provider_integration_plan_context=( + auth_provider_integration_plan_context + ), + source_derived_api_context=source_derived_api_context, + audit_events_api_context=audit_events_api_context, + reviewer_workflow_shell_context=reviewer_workflow_shell_context, + reviewer_created_state_api_context=reviewer_created_state_api_context, + reset_reload_dry_run_context=reset_reload_dry_run_context, + reset_reload_execution_plan_context=reset_reload_execution_plan_context, + reset_reload_planning_metadata_api_context=( + reset_reload_planning_metadata_api_context + ), + auth_runtime_config=active_auth_config, + page_data_mode=active_page_data_mode, + feedback_context=feedback_context, + github_feedback_config=github_feedback_config, + github_feedback_client=github_feedback_client, + cloudflare_jwks_fetcher=cloudflare_jwks_fetcher, + cloudflare_auth_now=cloudflare_auth_now, + reviewer_ui_context=active_reviewer_context, + ccld_record_request_ui_context=active_ccld_context, + operator_coverage_context=operator_coverage_context, + _postgres_context_resolved=True, + ) if method == "GET": redirect_location = _legacy_compare_facilities_redirect_location(path) if redirect_location is not None: @@ -1114,7 +1189,6 @@ def route_response( return 404, "text/plain; charset=utf-8", b"Not found" return 200, "image/x-icon", FAVICON_FILE.read_bytes() active_auth_config = _active_auth_runtime_config(auth_runtime_config) - active_page_data_mode = _active_page_data_mode(page_data_mode) if parsed_path == OPERATOR_COVERAGE_PREFIX or parsed_path.startswith( f"{OPERATOR_COVERAGE_PREFIX}/" ): @@ -1542,7 +1616,7 @@ def _default_reviewer_context_for_mode( # Postgres mode: outer access control (e.g. Cloudflare Access) handles # authentication. If a request reaches here, the caller is considered # authorized. Full OIDC/session auth is a future milestone. - return _default_postgres_reviewer_context() + return None def _default_ccld_context_for_mode( @@ -1635,25 +1709,49 @@ def _default_feedback_context( ) -def _default_postgres_reviewer_context() -> ReviewerUiContext | None: - global _DEFAULT_POSTGRES_REVIEWER_UI_CONTEXT - if _DEFAULT_POSTGRES_REVIEWER_UI_CONTEXT is not None: - return _DEFAULT_POSTGRES_REVIEWER_UI_CONTEXT +def _default_postgres_engine() -> Engine | None: + global _DEFAULT_POSTGRES_ENGINE + if _DEFAULT_POSTGRES_ENGINE is not None: + return _DEFAULT_POSTGRES_ENGINE try: database_config = load_hosted_database_config() except HostedDatabaseConfigError: return None if database_config.database_url is None: return None - engine = create_engine(database_config.database_url) - connection = engine.connect() - _DEFAULT_POSTGRES_REVIEWER_UI_CONTEXT = reviewer_ui_context_for_connection( - connection, - actor=local_test_reviewer_actor(scopes=(POSTGRES_REVIEWER_UI_SCOPE,)), - scope=POSTGRES_REVIEWER_UI_SCOPE, - manage_read_transactions=True, - ) - return _DEFAULT_POSTGRES_REVIEWER_UI_CONTEXT + with _DEFAULT_POSTGRES_ENGINE_LOCK: + if _DEFAULT_POSTGRES_ENGINE is None: + _DEFAULT_POSTGRES_ENGINE = create_engine(database_config.database_url) + return _DEFAULT_POSTGRES_ENGINE + + +@contextmanager +def _default_postgres_reviewer_context() -> Iterator[ReviewerUiContext | None]: + engine = _default_postgres_engine() + if engine is None: + yield None + return + try: + connection = engine.connect() + except SQLAlchemyError: + yield None + return + with connection: + yield reviewer_ui_context_for_connection( + connection, + actor=local_test_reviewer_actor(scopes=(POSTGRES_REVIEWER_UI_SCOPE,)), + scope=POSTGRES_REVIEWER_UI_SCOPE, + manage_read_transactions=True, + ) + + +def _dispose_default_postgres_engine() -> None: + global _DEFAULT_POSTGRES_ENGINE + with _DEFAULT_POSTGRES_ENGINE_LOCK: + engine = _DEFAULT_POSTGRES_ENGINE + _DEFAULT_POSTGRES_ENGINE = None + if engine is not None: + engine.dispose() def _facility_reference_from_context( @@ -1973,12 +2071,14 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--port", type=int, default=8000) args = parser.parse_args(argv) - with create_server(args.host, args.port) as server: - host, port = server.server_address[:2] - print(f"{APP_NAME} running locally at http://{format_host(host)}:{port}/") - print(SCAFFOLD_NOTICE) - try: + try: + with create_server(args.host, args.port) as server: + host, port = server.server_address[:2] + print(f"{APP_NAME} running locally at http://{format_host(host)}:{port}/") + print(SCAFFOLD_NOTICE) server.serve_forever() - except KeyboardInterrupt: - print("Stopping local hosted tester MVP scaffold.") + except KeyboardInterrupt: + print("Stopping local hosted tester MVP scaffold.") + finally: + _dispose_default_postgres_engine() return 0 diff --git a/tests/unit/test_hosted_postgres_connection_lifecycle.py b/tests/unit/test_hosted_postgres_connection_lifecycle.py new file mode 100644 index 0000000..0598346 --- /dev/null +++ b/tests/unit/test_hosted_postgres_connection_lifecycle.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from urllib.parse import urlencode + +import pytest +from sqlalchemy import Column, Integer, MetaData, Table, create_engine, event, func, select +from sqlalchemy.engine import Connection, Engine +from sqlalchemy.pool import QueuePool + +from ccld_complaints.hosted_app import app as hosted_app +from ccld_complaints.hosted_app.reviewer_created_state import ( + hosted_reviewer_created_state, +) +from ccld_complaints.hosted_app.reviewer_ui import ( + REVIEWER_UI_PREFIX, + REVIEWER_UI_UPDATE_PATH, +) +from ccld_complaints.hosted_app.seeded_import import ( + hosted_seeded_import_metadata, + import_seeded_corpus_artifact, + load_seeded_corpus_artifact, +) + +FIXTURE = Path("tests/fixtures/hosted_seeded_corpus/validated_seeded_corpus.json") +SOURCE_RECORD_KEY = "complaint:ccld:complaint:32-CR-20220407124448" + + +def test_default_postgres_engine_is_created_once_under_concurrent_access( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + engine = _seeded_engine(tmp_path) + create_count = 0 + count_lock = threading.Lock() + + def create_once(_database_url: str) -> Engine: + nonlocal create_count + with count_lock: + create_count += 1 + time.sleep(0.02) + return engine + + monkeypatch.setattr(hosted_app, "_DEFAULT_POSTGRES_ENGINE", None) + monkeypatch.setattr( + hosted_app, + "load_hosted_database_config", + lambda: SimpleNamespace(database_url="postgresql://example.invalid/records"), + ) + monkeypatch.setattr(hosted_app, "create_engine", create_once) + + try: + with ThreadPoolExecutor(max_workers=12) as executor: + resolved = tuple( + executor.map( + lambda _index: hosted_app._default_postgres_engine(), + range(24), + ) + ) + + assert create_count == 1 + assert all(item is engine for item in resolved) + finally: + hosted_app._dispose_default_postgres_engine() + + +def test_repeated_health_checks_reuse_engine_and_release_connections( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + engine = _seeded_engine(tmp_path) + checkouts, checkins = _pool_event_counts(engine) + monkeypatch.setattr(hosted_app, "_DEFAULT_POSTGRES_ENGINE", engine) + monkeypatch.setattr( + hosted_app, + "create_engine", + lambda _database_url: pytest.fail("health checks must reuse the process engine"), + ) + + for _ in range(25): + assert hosted_app.health_response()["source_data_loaded"] is True + assert engine.pool.checkedout() == 0 + + assert checkouts[0] == 25 + assert checkins[0] == 25 + engine.dispose() + + +def test_repeated_successful_hosted_routes_release_request_connections( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + engine = _seeded_engine(tmp_path) + checkouts, checkins = _pool_event_counts(engine) + monkeypatch.setattr(hosted_app, "_DEFAULT_POSTGRES_ENGINE", engine) + + for _ in range(25): + status, _content_type, _body = hosted_app.route_response( + REVIEWER_UI_PREFIX, + page_data_mode="postgres", + ) + assert status == 200 + assert engine.pool.checkedout() == 0 + + assert checkouts[0] == 25 + assert checkins[0] == 25 + engine.dispose() + + +def test_failed_hosted_route_rolls_back_before_connection_release( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + engine = _seeded_engine(tmp_path) + marker_metadata = MetaData() + markers = Table( + "connection_lifecycle_markers", + marker_metadata, + Column("marker_id", Integer, primary_key=True), + ) + marker_metadata.create_all(engine) + captured_connections: list[Connection] = [] + monkeypatch.setattr(hosted_app, "_DEFAULT_POSTGRES_ENGINE", engine) + + def failing_route( + _path: str, + context: Any, + **_kwargs: Any, + ) -> tuple[int, str, bytes]: + connection = context.workflow_shell_context.source_derived_api_context.connection + captured_connections.append(connection) + connection.execute(markers.insert().values(marker_id=1)) + raise RuntimeError("forced route failure") + + monkeypatch.setattr(hosted_app, "route_reviewer_ui_response", failing_route) + + with pytest.raises(RuntimeError, match="forced route failure"): + hosted_app.route_response(REVIEWER_UI_PREFIX, page_data_mode="postgres") + + assert len(captured_connections) == 1 + assert captured_connections[0].closed is True + assert engine.pool.checkedout() == 0 + with engine.connect() as connection: + assert connection.scalar(select(func.count()).select_from(markers)) == 0 + engine.dispose() + + +def test_post_authorization_early_return_does_not_leave_idle_transaction( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + engine = _seeded_engine(tmp_path) + rollback_count = [0] + event.listen( + engine, + "rollback", + lambda _connection: rollback_count.__setitem__(0, rollback_count[0] + 1), + ) + monkeypatch.setattr(hosted_app, "_DEFAULT_POSTGRES_ENGINE", engine) + + status, _content_type, _body = hosted_app.route_response( + REVIEWER_UI_UPDATE_PATH, + method="POST", + request_body=urlencode({"source_record_key": SOURCE_RECORD_KEY}).encode("utf-8"), + page_data_mode="postgres", + ) + + assert status == 403 + assert rollback_count[0] == 1 + assert engine.pool.checkedout() == 0 + engine.dispose() + + +def _seeded_engine(tmp_path: Path) -> Engine: + engine = create_engine( + f"sqlite+pysqlite:///{tmp_path / 'hosted-connection-lifecycle.sqlite'}", + poolclass=QueuePool, + pool_size=2, + max_overflow=0, + ) + hosted_seeded_import_metadata.create_all(engine) + hosted_reviewer_created_state.metadata.create_all(engine) + with engine.begin() as connection: + import_seeded_corpus_artifact( + connection, + load_seeded_corpus_artifact(FIXTURE), + ) + return engine + + +def _pool_event_counts(engine: Engine) -> tuple[list[int], list[int]]: + checkouts = [0] + checkins = [0] + event.listen( + engine, + "checkout", + lambda *_args: checkouts.__setitem__(0, checkouts[0] + 1), + ) + event.listen( + engine, + "checkin", + lambda *_args: checkins.__setitem__(0, checkins[0] + 1), + ) + return checkouts, checkins diff --git a/tests/unit/test_hosted_reviewer_ui.py b/tests/unit/test_hosted_reviewer_ui.py index c5c726a..c482a15 100644 --- a/tests/unit/test_hosted_reviewer_ui.py +++ b/tests/unit/test_hosted_reviewer_ui.py @@ -5404,7 +5404,7 @@ def test_default_postgres_reviewer_context_uses_loaded_ccld_corpus_scope( ) -> None: engine = create_engine("sqlite+pysqlite:///:memory:") hosted_seeded_import_metadata.create_all(engine) - monkeypatch.setattr(hosted_app, "_DEFAULT_POSTGRES_REVIEWER_UI_CONTEXT", None) + monkeypatch.setattr(hosted_app, "_DEFAULT_POSTGRES_ENGINE", None) monkeypatch.setattr( hosted_app, "load_hosted_database_config", @@ -5413,18 +5413,17 @@ def test_default_postgres_reviewer_context_uses_loaded_ccld_corpus_scope( monkeypatch.setattr(hosted_app, "create_engine", lambda _url: engine) try: - context = hosted_app._default_postgres_reviewer_context() - assert context is not None - source_context = context.workflow_shell_context.source_derived_api_context - - assert source_context.scope == POSTGRES_REVIEWER_UI_SCOPE - assert source_context.scope.scope_id != "seeded-ccld-fixture-2026-06-13" - assert source_context.actor is not None - assert source_context.actor.scopes == (POSTGRES_REVIEWER_UI_SCOPE,) - assert context.manage_read_transactions is True + with hosted_app._default_postgres_reviewer_context() as context: + assert context is not None + source_context = context.workflow_shell_context.source_derived_api_context + + assert source_context.scope == POSTGRES_REVIEWER_UI_SCOPE + assert source_context.scope.scope_id != "seeded-ccld-fixture-2026-06-13" + assert source_context.actor is not None + assert source_context.actor.scopes == (POSTGRES_REVIEWER_UI_SCOPE,) + assert context.manage_read_transactions is True finally: - monkeypatch.setattr(hosted_app, "_DEFAULT_POSTGRES_REVIEWER_UI_CONTEXT", None) - engine.dispose() + hosted_app._dispose_default_postgres_engine() def _local_dev_auth_config() -> Any: return load_hosted_auth_runtime_config(