ontorefactor-governance/tests/test_agents.py

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