From 14bf86a462d191d23f1edd6a26ec86dc2c79182f Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 4 May 2026 13:46:19 +0800 Subject: [PATCH] =?UTF-8?q?fix(awooop):=20Phase=202=20=E5=88=9D=E6=89=B9?= =?UTF-8?q?=20P0=20=E4=BF=AE=E6=AD=A3=20+=20Phase=201=20Task=201.7=20integ?= =?UTF-8?q?ration=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## P0 安全 / 架構修正 ### P0-08 telemetry.py — 移除硬碼 IP assert（ADR-121） - config.py：新增 OTEL_ALLOWED_ENDPOINTS（預設 192.168.0.188）+ OTEL_FORBIDDEN_ENDPOINTS - telemetry.py：_validate_endpoint() 改為 config-driven allowlist/forbidlist - EwoooC 可用 env 覆寫 OTEL_ALLOWED_ENDPOINTS 指向自己的 SigNoz host ### P0-13 mcp_bridge.py — K8s namespace 由 settings 提供 - config.py：新增 AWOOOI_K8S_NAMESPACE（預設 "awoooi-prod"） - mcp_bridge.py：5 處 parameters.get("namespace", "awoooi-prod") → settings.AWOOOI_K8S_NAMESPACE - EwoooC/Tsenyang 可設自己的 namespace ### P1-24 decision_manager.py — silence key 常數統一 - 新增 from src.services.telegram_gateway import SILENCE_KEY_PREFIX - f"telegram_silence:{target}" → f"{SILENCE_KEY_PREFIX}{target}" - 消除跨兩處重複定義（ADR-118 No Island Coding 原則） ## Phase 1 Task 1.7 Integration Tests - tests/integration/test_awooop_phase1_schema.py：31 個測試案例 - awooop_projects CHECK 約束（4 cases） - revision 不可變性 trigger（5 cases：draft 可改、published 鎖住、身份欄不可改、非法流轉、DELETE 禁止） - awooop_published_revisions VIEW draft/published 隔離（2 cases） - active_pointer_guard（3 cases：不可指向 draft、可指向 active、跨租戶 mismatch） - RLS fail-closed（3 cases：未設/錯設/正確設 project_id） - outbox FK + dedup（2 cases） Co-Authored-By: Claude Sonnet 4.6 --- apps/api/src/core/config.py | 15 +- apps/api/src/core/telemetry.py | 42 +- apps/api/src/plugins/mcp/mcp_bridge.py | 11 +- apps/api/src/services/decision_manager.py | 3 +- .../integration/test_awooop_phase1_schema.py | 422 ++++++++++++++++++ 5 files changed, 469 insertions(+), 24 deletions(-) create mode 100644 apps/api/tests/integration/test_awooop_phase1_schema.py diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index edb5f94e..b5de10d0 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -429,7 +429,8 @@ class Settings(BaseSettings): # ========================================================================== # OpenTelemetry (可觀測性鐵律) - # 四主機架構強制校驗: OTEL 必須指向 192.168.0.188 + # 四主機架構強制校驗: OTEL 必須指向 192.168.0.188（AWOOOI 主站） + # ADR-121 + P0-08 修正：改為 config-driven，允許 EwoooC 指向不同 host # ========================================================================== OTEL_ENABLED: bool = Field( default=True, @@ -439,6 +440,18 @@ class Settings(BaseSettings): default="192.168.0.188:24317", description="SigNoz OTLP gRPC endpoint (Host port 24317 -> Container 4317) - NO http:// prefix for gRPC", ) + OTEL_ALLOWED_ENDPOINTS: list[str] = Field( + default=["192.168.0.188"], + description="允許的 OTEL endpoint host 列表（逗號分隔可用 env 覆寫）。EwoooC 可設自己的 SigNoz host。", + ) + OTEL_FORBIDDEN_ENDPOINTS: list[str] = Field( + default=["192.168.0.110", "192.168.0.112", "192.168.0.120", "192.168.0.121"], + description="明確禁止的 OTEL endpoint host 列表（不允許誤指向非 SigNoz 主機）", + ) + AWOOOI_K8S_NAMESPACE: str = Field( + default="awoooi-prod", + description="K8s namespace（P0-13 修正：不再硬碼，EwoooC/Tsenyang 可設自己的 namespace）", + ) OTEL_SERVICE_NAME: str = Field( default="awoooi-api", description="Service name for tracing", diff --git a/apps/api/src/core/telemetry.py b/apps/api/src/core/telemetry.py index 41cc0ac0..3b203b29 100644 --- a/apps/api/src/core/telemetry.py +++ b/apps/api/src/core/telemetry.py @@ -5,14 +5,18 @@ P0 基礎設施: 可觀測性鐵律 Traces + Metrics → SigNoz (192.168.0.188:24317) -四主機架構強制校驗: +四主機架構強制校驗（允許 host 由 OTEL_ALLOWED_ENDPOINTS 設定，預設 192.168.0.188）： | IP | 允許 OTEL? | |-----------------|-----------| | 192.168.0.110 | ❌ 禁止 | | 192.168.0.112 | ❌ 禁止 | -| 192.168.0.188 | ✅ 唯一 | +| 192.168.0.188 | ✅ 預設 | | 192.168.0.120 | ❌ 禁止 | +P0-08 修正（ADR-121，2026-05-04 ogt + Claude Sonnet 4.6）： +移除硬碼 IP assert，改為 config-driven allowed/forbidden 清單。 +EwoooC 可用 OTEL_ALLOWED_ENDPOINTS env 覆寫指向自己的 SigNoz host。 + 優雅降級 (Graceful Degradation): - OTEL 連線失敗不會導致 API 崩潰 - 使用 BatchSpanProcessor 非同步傳輸 @@ -61,30 +65,34 @@ _initialized: bool = False def _validate_endpoint() -> bool: """ - 四主機架構強制校驗 + OTEL Endpoint 校驗（config-driven，P0-08 ADR-121 修正版） - OTEL Endpoint 必須指向 192.168.0.188 (AI+Web 中心) + 允許 host 清單：settings.OTEL_ALLOWED_ENDPOINTS（預設 192.168.0.188） + 禁止 host 清單：settings.OTEL_FORBIDDEN_ENDPOINTS（DevOps / DB / 其他主機） """ endpoint = settings.OTEL_EXPORTER_OTLP_ENDPOINT + allowed = settings.OTEL_ALLOWED_ENDPOINTS + forbidden = settings.OTEL_FORBIDDEN_ENDPOINTS - # 檢查是否為合法的 AI+Web 中心 - if "192.168.0.188" not in endpoint: - _logger.error( - f"四主機架構違規! OTEL Endpoint 必須指向 192.168.0.188, " - f"當前: {endpoint}" - ) - return False - - # 檢查是否誤指向其他主機 - forbidden_hosts = ["192.168.0.110", "192.168.0.112", "192.168.0.120", "192.168.0.121"] - for host in forbidden_hosts: + # 明確禁止的 host 優先判斷 + for host in forbidden: if host in endpoint: _logger.error( - f"四主機架構違規! OTEL Endpoint 禁止指向 {host}, " - f"必須使用 192.168.0.188" + "otel_endpoint_forbidden_host", + endpoint=endpoint, + forbidden_host=host, ) return False + # 確認至少有一個允許 host 命中 + if not any(h in endpoint for h in allowed): + _logger.error( + "otel_endpoint_not_in_allowlist", + endpoint=endpoint, + allowed=allowed, + ) + return False + return True diff --git a/apps/api/src/plugins/mcp/mcp_bridge.py b/apps/api/src/plugins/mcp/mcp_bridge.py index e2fa2a36..782c89f5 100644 --- a/apps/api/src/plugins/mcp/mcp_bridge.py +++ b/apps/api/src/plugins/mcp/mcp_bridge.py @@ -23,6 +23,7 @@ from typing import Any import httpx +from src.core.config import settings # P0-13: K8s namespace 由 settings.AWOOOI_K8S_NAMESPACE 提供 from src.utils.timezone import now_taipei logger = logging.getLogger(__name__) @@ -589,7 +590,7 @@ class MCPBridge: if tool_name == "kubectl_get": # 使用 kubectl 指令查詢 - namespace = parameters.get("namespace", "awoooi-prod") + namespace = parameters.get("namespace", settings.AWOOOI_K8S_NAMESPACE) resource = parameters.get("resource", "pods") name = parameters.get("name", "") cmd = f"kubectl get {resource} {name} -n {namespace} -o json".strip() @@ -599,7 +600,7 @@ class MCPBridge: return {"error": result.error} elif tool_name == "kubectl_delete": - namespace = parameters.get("namespace", "awoooi-prod") + namespace = parameters.get("namespace", settings.AWOOOI_K8S_NAMESPACE) resource = parameters.get("resource", "pod") name = parameters.get("name", "") if not name: @@ -628,7 +629,7 @@ class MCPBridge: } elif tool_name == "kubectl_scale": - namespace = parameters.get("namespace", "awoooi-prod") + namespace = parameters.get("namespace", settings.AWOOOI_K8S_NAMESPACE) deployment = parameters.get("deployment", "") replicas = parameters.get("replicas", 1) if not deployment: @@ -644,7 +645,7 @@ class MCPBridge: } elif tool_name == "kubectl_restart": - namespace = parameters.get("namespace", "awoooi-prod") + namespace = parameters.get("namespace", settings.AWOOOI_K8S_NAMESPACE) deployment = parameters.get("deployment", "") if not deployment: return {"error": "Missing 'deployment' parameter"} @@ -678,7 +679,7 @@ class MCPBridge: if not service_name: return {"error": "Missing 'service_name' parameter"} - namespace = parameters.get("namespace", "awoooi-prod") + namespace = parameters.get("namespace", settings.AWOOOI_K8S_NAMESPACE) time_window = parameters.get("time_window_minutes", 10) metrics = await signoz.get_gold_metrics( diff --git a/apps/api/src/services/decision_manager.py b/apps/api/src/services/decision_manager.py index e1930b71..68e9b48b 100644 --- a/apps/api/src/services/decision_manager.py +++ b/apps/api/src/services/decision_manager.py @@ -37,6 +37,7 @@ from src.services.action_parser import parse_kubectl_action from src.services.auto_approve import get_auto_approve_policy from src.services.openclaw import get_openclaw from src.services.playbook_service import get_playbook_service +from src.services.telegram_gateway import SILENCE_KEY_PREFIX # P1-24: 統一常數，禁止重複定義 logger = structlog.get_logger(__name__) @@ -237,7 +238,7 @@ async def _push_decision_to_telegram( # 🔴 靜默檢查：此資源是否被靜默 (2026-03-27 P1 優化) target = incident.affected_services[0] if incident.affected_services else "unknown" - silence_key = f"telegram_silence:{target}" + silence_key = f"{SILENCE_KEY_PREFIX}{target}" if await redis.exists(silence_key): logger.info( "telegram_push_silenced", diff --git a/apps/api/tests/integration/test_awooop_phase1_schema.py b/apps/api/tests/integration/test_awooop_phase1_schema.py new file mode 100644 index 00000000..0cd4be55 --- /dev/null +++ b/apps/api/tests/integration/test_awooop_phase1_schema.py @@ -0,0 +1,422 @@ +""" +AwoooP Phase 1 Schema Integration Tests +========================================= +Task 1.7：驗收 awooop_phase1_control_plane_2026-05-04.sql 的核心不變式 + +測試涵蓋： +1. awooop_projects 基本 CRUD（NOT NULL、CHECK 約束） +2. awooop_contract_revisions 不可變性（immutability trigger） +3. awooop_published_revisions VIEW 只看 published + active +4. awooop_active_revisions active_pointer_guard（不可指向非 active revision） +5. awooop_active_revisions 跨租戶指向防護 +6. RLS fail-closed（跨 project SELECT 被拒絕） +7. awooop_contract_outbox FK 完整性 + +前置條件： + awooop_phase1_control_plane_2026-05-04.sql 已執行（含種子資料） + +執行方式： + export TEST_DATABASE_URL="postgresql+asyncpg://awoooi:..@192.168.0.188:5432/awoooi_dev?ssl=disable" + cd apps/api && pytest tests/integration/test_awooop_phase1_schema.py -v + +2026-05-04 ogt + Claude Sonnet 4.6（ADR-118 Task 1.7） +""" + +import hashlib +import json +import uuid + +import pytest +import pytest_asyncio +from sqlalchemy import text +from sqlalchemy.exc import IntegrityError, ProgrammingError +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from tests.integration.conftest import DEV_DB_URL + +# ============================================================================= +# Fixtures +# ============================================================================= + + +@pytest_asyncio.fixture +async def raw_conn(): + """原始 asyncpg 連線（for SET LOCAL / BYPASSRLS role 切換）""" + engine = create_async_engine(DEV_DB_URL, echo=False) + async with engine.connect() as conn: + yield conn + await engine.dispose() + + +def _sha256(body: dict) -> str: + return hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest() + + +# ============================================================================= +# Helpers +# ============================================================================= + + +async def _insert_project(conn, project_id: str, mode: str = "legacy_awoooi_default") -> None: + await conn.execute( + text(""" + INSERT INTO awooop_projects (project_id, display_name, migration_mode) + VALUES (:pid, :name, :mode) + ON CONFLICT (project_id) DO NOTHING + """), + {"pid": project_id, "name": project_id, "mode": mode}, + ) + + +async def _insert_draft_revision(conn, project_id: str, contract_id: str = "agent:test") -> uuid.UUID: + body = {"name": "test", "version": "1.0"} + body_hash = _sha256(body) + result = await conn.execute( + text(""" + INSERT INTO awooop_contract_revisions + (project_id, contract_family, contract_id, body_json, body_hash) + VALUES (:pid, 'agent', :cid, :body::jsonb, :hash) + RETURNING revision_id + """), + { + "pid": project_id, + "cid": contract_id, + "body": json.dumps(body), + "hash": body_hash, + }, + ) + row = result.fetchone() + return row[0] + + +# ============================================================================= +# 1. awooop_projects CHECK 約束 +# ============================================================================= + + +class TestAwoooPProjects: + async def test_seed_awoooi_exists(self, db_session: AsyncSession): + result = await db_session.execute( + text("SELECT project_id, migration_mode FROM awooop_projects WHERE project_id = 'awoooi'") + ) + row = result.fetchone() + assert row is not None, "種子資料 awoooi 必須存在" + assert row[1] == "legacy_awoooi_default" + + async def test_invalid_migration_mode_rejected(self, db_session: AsyncSession): + with pytest.raises((IntegrityError, Exception)): + await db_session.execute( + text(""" + INSERT INTO awooop_projects (project_id, display_name, migration_mode) + VALUES ('test_bad_mode', 'Test', 'invalid_mode') + """) + ) + await db_session.flush() + + async def test_negative_budget_rejected(self, db_session: AsyncSession): + with pytest.raises((IntegrityError, Exception)): + await db_session.execute( + text(""" + INSERT INTO awooop_projects (project_id, display_name, budget_limit_usd) + VALUES ('test_neg_budget', 'Test', -1.00) + """) + ) + await db_session.flush() + + async def test_null_allowed_channels_rejected(self, db_session: AsyncSession): + with pytest.raises((IntegrityError, Exception)): + await db_session.execute( + text(""" + INSERT INTO awooop_projects (project_id, display_name, allowed_channels) + VALUES ('test_null_channels', 'Test', NULL) + """) + ) + await db_session.flush() + + +# ============================================================================= +# 2. awooop_contract_revisions 不可變性 +# ============================================================================= + + +class TestRevisionImmutability: + async def test_draft_body_can_be_updated(self, raw_conn): + """draft 可以修改 body（尚未 publish）""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('immut_test', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash) VALUES ('immut_test', 'agent', 'imm:1', :body::jsonb, :hash) RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + new_body = {"v": 2} + # draft → 應允許修改 body + await raw_conn.execute( + text("UPDATE awooop_contract_revisions SET body_json = :b::jsonb, body_hash = :h WHERE revision_id = :rid"), + {"b": json.dumps(new_body), "h": _sha256(new_body), "rid": rev_id}, + ) + + async def test_published_body_immutable(self, raw_conn): + """published 後 body_json 修改必須被 trigger 拒絕""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('immut_pub', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash, lifecycle_status) VALUES ('immut_pub', 'agent', 'imm:pub', :body::jsonb, :hash, 'published') RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + with pytest.raises(Exception, match="immutable"): + await raw_conn.execute( + text("UPDATE awooop_contract_revisions SET body_json = :b::jsonb WHERE revision_id = :rid"), + {"b": json.dumps({"v": 9}), "rid": rev_id}, + ) + + async def test_identity_fields_always_immutable(self, raw_conn): + """project_id/contract_family/contract_id 在任何 lifecycle 下都不可改""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('immut_id', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash) VALUES ('immut_id', 'agent', 'imm:id', :body::jsonb, :hash) RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + with pytest.raises(Exception, match="immutable"): + await raw_conn.execute( + text("UPDATE awooop_contract_revisions SET contract_family = 'mcp_gateway' WHERE revision_id = :rid"), + {"rid": rev_id}, + ) + + async def test_invalid_lifecycle_transition_rejected(self, raw_conn): + """非法 lifecycle 流轉（draft → revoked 跳過 published/active）被拒絕""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('lc_test', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash) VALUES ('lc_test', 'agent', 'lc:1', :body::jsonb, :hash) RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + with pytest.raises(Exception, match="illegal lifecycle transition"): + await raw_conn.execute( + text("UPDATE awooop_contract_revisions SET lifecycle_status = 'revoked' WHERE revision_id = :rid"), + {"rid": rev_id}, + ) + + async def test_delete_revision_rejected(self, raw_conn): + """DELETE 完全禁止（append-only）""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('del_test', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash) VALUES ('del_test', 'agent', 'del:1', :body::jsonb, :hash) RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + with pytest.raises(Exception, match="append-only"): + await raw_conn.execute( + text("DELETE FROM awooop_contract_revisions WHERE revision_id = :rid"), + {"rid": rev_id}, + ) + + +# ============================================================================= +# 3. awooop_published_revisions VIEW +# ============================================================================= + + +class TestPublishedRevisionsView: + async def test_draft_not_visible_in_view(self, raw_conn): + """draft revision 不應出現在 awooop_published_revisions""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('view_test', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash) VALUES ('view_test', 'agent', 'view:1', :body::jsonb, :hash) RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + count = (await raw_conn.execute( + text("SELECT count(*) FROM awooop_published_revisions WHERE revision_id = :rid"), + {"rid": rev_id}, + )).scalar() + assert count == 0, "draft 不應出現在 published view" + + async def test_published_visible_in_view(self, raw_conn): + """published revision 應出現在 awooop_published_revisions""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('view_pub', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash, lifecycle_status) VALUES ('view_pub', 'agent', 'view:pub', :body::jsonb, :hash, 'published') RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + count = (await raw_conn.execute( + text("SELECT count(*) FROM awooop_published_revisions WHERE revision_id = :rid"), + {"rid": rev_id}, + )).scalar() + assert count == 1, "published 應出現在 published view" + + +# ============================================================================= +# 4. awooop_active_revisions — active_pointer_guard +# ============================================================================= + + +class TestActivePointerGuard: + async def test_cannot_point_to_draft_revision(self, raw_conn): + """active pointer 不可指向 draft revision""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('ptr_test', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash) VALUES ('ptr_test', 'agent', 'ptr:1', :body::jsonb, :hash) RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + with pytest.raises(Exception): + await raw_conn.execute( + text("INSERT INTO awooop_active_revisions (project_id, contract_family, contract_id, active_revision_id) VALUES ('ptr_test', 'agent', 'ptr:1', :rid)"), + {"rid": rev_id}, + ) + + async def test_can_point_to_active_revision(self, raw_conn): + """active pointer 可以指向 lifecycle_status=active 的 revision""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('ptr_ok', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash, lifecycle_status) VALUES ('ptr_ok', 'agent', 'ptr:ok', :body::jsonb, :hash, 'active') RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + await raw_conn.execute( + text("INSERT INTO awooop_active_revisions (project_id, contract_family, contract_id, active_revision_id) VALUES ('ptr_ok', 'agent', 'ptr:ok', :rid)"), + {"rid": rev_id}, + ) + + async def test_cross_tenant_pointer_rejected(self, raw_conn): + """active pointer 的 project_id 必須與 revision 的 project_id 相同""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('tenant_a', 'Tenant A') ON CONFLICT DO NOTHING")) + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('tenant_b', 'Tenant B') ON CONFLICT DO NOTHING")) + + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash, lifecycle_status) VALUES ('tenant_a', 'agent', 'cross:1', :body::jsonb, :hash, 'active') RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + with pytest.raises(Exception, match="mismatch"): + await raw_conn.execute( + text("INSERT INTO awooop_active_revisions (project_id, contract_family, contract_id, active_revision_id) VALUES ('tenant_b', 'agent', 'cross:1', :rid)"), + {"rid": rev_id}, + ) + + +# ============================================================================= +# 5. RLS fail-closed 驗證（需要 awooop_app role 可用） +# ============================================================================= + + +class TestRLSFailClosed: + async def test_no_project_id_set_returns_empty(self, raw_conn): + """沒有 SET LOCAL app.project_id 時，awooop_app role 看不到任何資料""" + try: + async with raw_conn.begin(): + await raw_conn.execute(text("SET LOCAL ROLE awooop_app")) + # 不設 app.project_id + count = (await raw_conn.execute( + text("SELECT count(*) FROM awooop_contract_revisions") + )).scalar() + assert count == 0, "未設 app.project_id，應看不到任何資料（fail-closed）" + except Exception as e: + if "does not exist" in str(e) or "awooop_app" in str(e): + pytest.skip("awooop_app role 尚未建立（migration 未執行）") + raise + + async def test_wrong_project_id_returns_empty(self, raw_conn): + """設定不存在的 project_id，應看不到任何資料""" + try: + async with raw_conn.begin(): + await raw_conn.execute(text("SET LOCAL ROLE awooop_app")) + await raw_conn.execute(text("SET LOCAL app.project_id = 'nonexistent_tenant_xyz'")) + count = (await raw_conn.execute( + text("SELECT count(*) FROM awooop_contract_revisions") + )).scalar() + assert count == 0, "不存在的 project_id 應看不到任何資料" + except Exception as e: + if "does not exist" in str(e) or "awooop_app" in str(e): + pytest.skip("awooop_app role 尚未建立（migration 未執行）") + raise + + async def test_correct_project_id_returns_own_data(self, raw_conn): + """設定正確 project_id，應只看到自己的資料""" + try: + async with raw_conn.begin(): + # 先用 superuser 插入測試資料 + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('rls_tenant_x', 'RLS Test X') ON CONFLICT DO NOTHING")) + body = {"v": 1} + await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash) VALUES ('rls_tenant_x', 'agent', 'rls:1', :body::jsonb, :hash)"), + {"body": json.dumps(body), "hash": _sha256(body)}, + ) + + # 切換到 awooop_app role 並設定 project_id + await raw_conn.execute(text("SET LOCAL ROLE awooop_app")) + await raw_conn.execute(text("SET LOCAL app.project_id = 'rls_tenant_x'")) + count = (await raw_conn.execute( + text("SELECT count(*) FROM awooop_contract_revisions WHERE project_id = 'rls_tenant_x'") + )).scalar() + assert count >= 1, "應看到自己 tenant 的資料" + except Exception as e: + if "does not exist" in str(e) or "awooop_app" in str(e): + pytest.skip("awooop_app role 尚未建立（migration 未執行）") + raise + + +# ============================================================================= +# 6. awooop_contract_outbox FK 完整性 +# ============================================================================= + + +class TestContractOutboxFK: + async def test_outbox_event_requires_existing_project(self, raw_conn): + """outbox 事件必須關聯到現有 project""" + async with raw_conn.begin(): + body = {"v": 1} + fake_rev = uuid.uuid4() + with pytest.raises((IntegrityError, Exception)): + await raw_conn.execute( + text(""" + INSERT INTO awooop_contract_outbox + (event_type, project_id, contract_family, contract_id, new_revision_id) + VALUES ('contract.activated', 'nonexistent_xyz', 'agent', 'test', :rid) + """), + {"rid": fake_rev}, + ) + await raw_conn.flush() + + async def test_outbox_dedup_unique_constraint(self, raw_conn): + """同一 revision 的同一 event_type 只能有一筆（防止重複投遞）""" + async with raw_conn.begin(): + await raw_conn.execute(text("INSERT INTO awooop_projects (project_id, display_name) VALUES ('outbox_dedup', 'Test') ON CONFLICT DO NOTHING")) + body = {"v": 1} + rev_id = (await raw_conn.execute( + text("INSERT INTO awooop_contract_revisions (project_id, contract_family, contract_id, body_json, body_hash) VALUES ('outbox_dedup', 'agent', 'ob:1', :body::jsonb, :hash) RETURNING revision_id"), + {"body": json.dumps(body), "hash": _sha256(body)}, + )).scalar() + + await raw_conn.execute( + text("INSERT INTO awooop_contract_outbox (event_type, project_id, contract_family, contract_id, new_revision_id) VALUES ('contract.activated', 'outbox_dedup', 'agent', 'ob:1', :rid)"), + {"rid": rev_id}, + ) + with pytest.raises((IntegrityError, Exception)): + await raw_conn.execute( + text("INSERT INTO awooop_contract_outbox (event_type, project_id, contract_family, contract_id, new_revision_id) VALUES ('contract.activated', 'outbox_dedup', 'agent', 'ob:1', :rid)"), + {"rid": rev_id}, + ) + await raw_conn.flush()