from pathlib import Path import pytest from ontorefactor_governance.agents import execute_governance_agent from ontorefactor_governance.db import Database from ontorefactor_governance.service import GovernanceError, create_project, list_assertions, list_elements def project(db): return create_project(db, "tenant-a", {"name": "Agent import", "namespace": "urn:test:agent"}) def test_lambdagent_pipeline_discovers_and_persists_assets(): db = Database("sqlite:///:memory:") item = project(db) result = execute_governance_agent(db, "tenant-a", item["id"], { "source_type": "ddl", "source_name": "customer.sql", "environment": "production", "content": """ CREATE TABLE dcp_customer ( id INTEGER PRIMARY KEY, mobile VARCHAR(32) NOT NULL ); """, }) assert result["status"] == "completed" assert result["result"]["summary"] == {"elements": 7, "assertions": 8, "issues": 2} assert result["persisted"] == {"elements": 7, "assertions": 8, "issues": 2} assert len(result["trace"]) >= 5 assert list_elements(db, "tenant-a", item["id"])["total"] == 7 assert len(list_assertions(db, "tenant-a", item["id"])) == 8 assert any(value["review_status"] == "pending" for value in list_assertions(db, "tenant-a", item["id"])) db.close() def test_invalid_ddl_creates_a_failed_audited_run(): db = Database("sqlite:///:memory:") item = project(db) with pytest.raises(GovernanceError): execute_governance_agent(db, "tenant-a", item["id"], {"source_type": "ddl", "content": "SELECT 1"}) run = db.fetchone("SELECT status,error FROM governance_agent_runs") assert run["status"] == "failed" assert run["error"] db.close() def test_runtime_source_has_no_agentpaas_dependency(): source_root = Path(__file__).parents[1] / "src" / "ontorefactor_governance" source = "\n".join(path.read_text(encoding="utf-8") for path in source_root.glob("*.py")) assert "import agentpaas" not in source assert "from agentpaas" not in source