from __future__ import annotations import inspect import unittest from pathlib import Path from unittest.mock import Mock from app.core.postgres import PostgresClient class PostgresSchemaContractTests(unittest.TestCase): def test_upsert_employee_work_log_session_sql_references_runtime_columns( self, ) -> None: source = inspect.getsource(PostgresClient.upsert_employee_work_log_session) self.assertIn("INSERT INTO employee_work_log_sessions", source) self.assertIn("metadata", source) self.assertNotIn("session_metadata", source) def test_append_employee_work_log_action_sql_references_runtime_columns( self, ) -> None: source = inspect.getsource(PostgresClient.append_employee_work_log_action) self.assertIn("INSERT INTO employee_work_log_actions", source) self.assertIn("shift_date", source) self.assertIn("metadata", source) self.assertNotIn("event_metadata", source) def test_submit_dossier_review_supplement_sql_requires_review_session_id( self, ) -> None: source = inspect.getsource(PostgresClient.submit_dossier_review_supplement) self.assertIn("INSERT INTO dossier_review_submissions", source) self.assertIn("review_session_id", source) def test_create_dossier_review_session_serializes_dossier_scope_as_jsonb( self, ) -> None: create_source = inspect.getsource(PostgresClient.create_dossier_review_session) params_source = inspect.getsource(PostgresClient._build_create_session_params) self.assertIn("%(dossier_scope)s", create_source) self.assertIn("::jsonb", create_source) self.assertIn('json.dumps(payload.get("dossier_scope")', params_source) def test_assign_dossier_review_finding_uses_runtime_assignment_contract( self, ) -> None: source = inspect.getsource(PostgresClient.assign_dossier_review_finding) self.assertIn("INSERT INTO dossier_review_assignments", source) self.assertIn("assigned_department_code", source) self.assertIn("assigned_employee_id", source) self.assertIn("assigned_by_employee_id", source) self.assertNotIn("assignment_type", source) def test_create_dossier_review_finding_uses_runtime_status_columns(self) -> None: source = inspect.getsource(PostgresClient.create_dossier_review_finding) assignment_source = inspect.getsource( PostgresClient.assign_dossier_review_finding ) self.assertIn("INSERT INTO dossier_review_findings", source) self.assertIn("supplement_status", source) self.assertIn("status", source) self.assertIn("responsible_department_code", source) self.assertIn("current_assignment_id", assignment_source) def test_bootstrap_sql_matches_runtime_contracts_for_wave_one_guardrails( self, ) -> None: init_sql = Path("docker/postgres/init/01_init.sql").read_text(encoding="utf-8") self.assertIn("metadata JSONB", init_sql) self.assertNotIn("session_metadata JSONB", init_sql) self.assertNotIn("event_metadata JSONB", init_sql) self.assertIn("dossier_scope JSONB", init_sql) self.assertIn( "review_session_id UUID NOT NULL REFERENCES dossier_review_sessions", init_sql, ) self.assertIn("chk_employee_work_log_sessions_status", init_sql) self.assertIn("'planned', 'in_progress', 'completed', 'blocked'", init_sql) self.assertIn("chk_employee_work_log_actions_type", init_sql) self.assertIn( "'manual_update', 'ai_analysis', 'task_update', 'issue_recorded'", init_sql, ) self.assertIn( "'handover_note', 'progress_update', 'checklist_update', 'system_sync'", init_sql, ) self.assertIn("chk_dossier_review_sessions_status", init_sql) self.assertIn( "'draft', 'in_review', 'awaiting_assignment', 'assigned'", init_sql ) self.assertIn( "'partially_resolved', 'resolved', 'closed', 'cancelled'", init_sql ) self.assertIn("chk_dossier_review_findings_status", init_sql) self.assertIn( "'open', 'assigned', 'in_progress', 'pending_verification'", init_sql ) self.assertIn("'resolved', 'waived', 'rejected'", init_sql) self.assertIn("chk_dossier_review_findings_supplement_status", init_sql) self.assertIn("'not_requested', 'requested', 'collecting'", init_sql) self.assertIn("'submitted', 'insufficient', 'accepted'", init_sql) self.assertIn("chk_dossier_review_assignments_priority", init_sql) self.assertIn("'low', 'medium', 'high', 'critical'", init_sql) self.assertIn("current_assignment_id UUID", init_sql) def test_migration_backbone_exists_for_schema_governance(self) -> None: migration_sql = Path("db/migrations/0001_schema_baseline.sql").read_text( encoding="utf-8" ) self.assertIn("CREATE TABLE IF NOT EXISTS schema_migrations", migration_sql) self.assertIn("version VARCHAR(50) PRIMARY KEY", migration_sql) self.assertIn("applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()", migration_sql) self.assertIn("0001_schema_baseline", migration_sql) def test_transactional_review_writes_refetch_using_active_connection_when_provided( self, ) -> None: create_source = inspect.getsource(PostgresClient.create_dossier_review_session) finding_source = inspect.getsource(PostgresClient.create_dossier_review_finding) assign_source = inspect.getsource(PostgresClient.assign_dossier_review_finding) submit_source = inspect.getsource( PostgresClient.submit_dossier_review_supplement ) verify_source = inspect.getsource(PostgresClient.verify_dossier_review_finding) close_source = inspect.getsource(PostgresClient.close_dossier_review_session) self.assertIn("_fetch_single_dossier_review_session(", create_source) self.assertIn("connection=connection", create_source) self.assertIn("_fetch_single_dossier_review_session(", finding_source) self.assertIn("connection=connection", finding_source) self.assertIn("_fetch_single_dossier_review_session(", assign_source) self.assertIn("connection=connection", assign_source) self.assertIn("_fetch_single_dossier_review_session(", submit_source) self.assertIn("connection=connection", submit_source) self.assertIn("_fetch_single_dossier_review_session(", verify_source) self.assertIn("connection=connection", verify_source) self.assertIn("_fetch_single_dossier_review_session(", close_source) self.assertIn("connection=connection", close_source) def test_fetch_single_dossier_review_session_uses_transaction_scoped_reader( self, ) -> None: helper_source = inspect.getsource( PostgresClient._fetch_single_dossier_review_session ) scoped_reader_source = inspect.getsource( PostgresClient._fetch_dossier_review_sessions_with_connection ) self.assertIn("_fetch_dossier_review_sessions_with_connection(", helper_source) self.assertIn( "self.fetch_dossier_review_sessions(review_id=review_id)", helper_source ) self.assertIn("with connection.cursor() as cursor", scoped_reader_source) def test_run_in_transaction_commits_on_success_and_closes_connection(self) -> None: from unittest.mock import Mock, MagicMock connection = Mock() client = PostgresClient.__new__(PostgresClient) client.get_connection = MagicMock() client.get_connection.return_value.__enter__.return_value = connection observed = [] def operation(passed_connection: object) -> str: observed.append(passed_connection) return "ok" result = PostgresClient.run_in_transaction(client, operation) self.assertEqual(result, "ok") self.assertEqual(observed, [connection]) connection.commit.assert_called_once_with() connection.rollback.assert_not_called() def test_run_in_transaction_rolls_back_on_failure_and_closes_connection( self, ) -> None: from unittest.mock import Mock, MagicMock connection = Mock() client = PostgresClient.__new__(PostgresClient) client.get_connection = MagicMock() client.get_connection.return_value.__enter__.return_value = connection def operation(_: object) -> None: raise RuntimeError("boom") with self.assertRaisesRegex(RuntimeError, "boom"): PostgresClient.run_in_transaction(client, operation) connection.commit.assert_not_called() connection.rollback.assert_called_once_with() if __name__ == "__main__": unittest.main()