lambdagent/tests/test_governance.py

"""End-to-end tests for the M3-M0 governance vertical slice."""
from __future__ import annotations

import os

os.environ.setdefault("AGENTPAAS_DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("AGENTPAAS_TESTING", "1")

import pytest

from agentpaas.db.models import Database, now_utc
from agentpaas.governance import service
from agentpaas.governance.agents import execute_governance_agent


@pytest.fixture
def db():
    value = Database("sqlite:///:memory:")
    value.execute(
        "INSERT INTO tenants (id,name,plan,status,created_at) VALUES (?,?,?,?,?)",
        ("tenant-a", "Tenant A", "free", "active", now_utc()),
    )
    value.execute(
        "INSERT INTO tenants (id,name,plan,status,created_at) VALUES (?,?,?,?,?)",
        ("tenant-b", "Tenant B", "free", "active", now_utc()),
    )
    value.commit()
    return value


def test_reference_slice_is_complete_valid_and_idempotent(db):
    first = service.seed_tenant_isolation(db, "tenant-a")
    second = service.seed_tenant_isolation(db, "tenant-a")

    assert first["project"]["id"] == second["project"]["id"]
    assert second["validation"]["valid"] is True
    assert second["validation"]["findings"] == []
    assert second["overview"]["layers"]["M3"] == 8
    assert second["overview"]["layers"]["M0"] == 3
    assert second["overview"]["totals"]["pending_reviews"] == 0


def test_impact_crosses_business_software_and_governance_profiles(db):
    seeded = service.seed_tenant_isolation(db, "tenant-a")
    project_id = seeded["project"]["id"]
    elements = service.list_elements(db, "tenant-a", project_id, query="TenantIdColumn")["items"]
    column = next(item for item in elements if item["name"] == "TenantIdColumn")

    result = service.impact(db, "tenant-a", column["id"], max_depth=5)
    names = {item["name"] for item in result["nodes"]}
    assert {"Tenant", "IAMDataOwner", "TenantIsolationPolicy", "TenantIdCompleteness"} <= names
    assert any(edge["predicate"] == "REPRESENTS" for edge in result["edges"])


def test_tenant_scope_blocks_cross_tenant_reads(db):
    seeded = service.seed_tenant_isolation(db, "tenant-a")
    project_id = seeded["project"]["id"]
    with pytest.raises(service.GovernanceError) as exc:
        service.overview(db, "tenant-b", project_id)
    assert exc.value.status == 404


def test_ddl_agent_persists_facts_candidates_evidence_and_trace(db):
    project = service.create_project(db, "tenant-a", {
        "name": "DDL ingest", "namespace": "https://example.test/ddl",
    })
    ddl = """
    CREATE TABLE dcp_customer (
      id BIGINT PRIMARY KEY,
      mobile VARCHAR(32) NOT NULL,
      created_at TIMESTAMP
    );
    """
    response = execute_governance_agent(db, "tenant-a", project["id"], {
        "source_type": "ddl",
        "source_name": "customer.sql",
        "environment": "production",
        "content": ddl,
    })

    assert response["status"] == "completed"
    assert response["persisted"]["elements"] == 9
    assert response["persisted"]["assertions"] == 11
    assert response["persisted"]["issues"] == 2
    assert {entry["term_name"] for entry in response["trace"]} >= {
        "governance-source-normalizer", "asset-discovery-agent",
        "business-semantics-agent", "governance-controls-agent",
        "governance-synthesis-agent",
    }

    pending = service.list_assertions(
        db, "tenant-a", project["id"], review_status="pending"
    )
    assert len(pending) == 1
    assert pending[0]["predicate"] == "REPRESENTS"
    assert pending[0]["assertion_type"] == "inferred"
    assert pending[0]["evidence_ids"]

    reviewed = service.review_assertion(
        db, "tenant-a", pending[0]["id"], "accepted", "data-steward", "术语已确认"
    )
    assert reviewed["review_status"] == "accepted"
    assert reviewed["reviewer"] == "data-steward"

    runs = db.fetchall(
        "SELECT status,trace_json FROM governance_agent_runs WHERE tenant_id=?",
        ("tenant-a",),
    )
    assert len(runs) == 1 and runs[0]["status"] == "completed"


def test_validation_detects_untyped_published_m0_fact(db):
    project = service.create_project(db, "tenant-a", {
        "name": "Validation", "namespace": "https://example.test/validation",
    })
    fact = service.upsert_element(db, "tenant-a", project["id"], {
        "layer": "M0", "profile": "software", "kind": "PhysicalTable",
        "name": "prod.orphan", "status": "published",
    })
    result = service.validate_project(db, "tenant-a", project["id"], persist=True)
    assert result["valid"] is False
    assert any(item["code"] == "M0_UNKNOWN_TYPE" and item["target_id"] == fact["id"]
               for item in result["findings"])