group_xinghuo_jinrong/tests/test_sprint1_compliance_rules.py

from pathlib import Path
from uuid import uuid4

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

from tests.advisor_test_utils import login_staff_token, STAFF_ADVISOR, STAFF_COMPLIANCE

def login_token(username: str, password: str) -> str:
    _ = password
    actor = STAFF_COMPLIANCE if username == "compliance_test" else STAFF_ADVISOR
    return login_staff_token(client, actor_id=actor)

def test_compliance_rule_migration_sql_defines_core_tables():
    migration = (
        Path(__file__).resolve().parents[1]
        / "scripts"
        / "agent"
        / "migrate-advisor-agent-sprint1-3.sql"
    )
    text = migration.read_text(encoding="utf-8")
    for table in ("compliance_rule", "compliance_check_log"):
        assert f"CREATE TABLE IF NOT EXISTS {table}" in text
    for column in (
        "rule_code",
        "rule_type",
        "pattern",
        "severity",
        "category",
        "suggestion",
        "is_active",
        "priority",
        "created_by",
        "updated_by",
        "check_id",
        "trace_id",
        "advisor_id",
        "input_text",
        "risk_level",
        "hit_details",
        "ai_analysis",
        "ai_degraded",
    ):
        assert column in text

def test_compliance_rule_service_creates_updates_lists_and_soft_deletes_rule():
    from app.model.advisor_schemas import ComplianceRuleCreate, ComplianceRuleUpdate
    from app.repository.compliance_rule_repository import ComplianceRuleRepository
    from app.service.compliance_rule_service import ComplianceRuleService

    pattern = f"guaranteed-return-{uuid4().hex}"
    service = ComplianceRuleService(ComplianceRuleRepository())

    created = service.create_rule(
        ComplianceRuleCreate(
            rule_type="keyword",
            pattern=pattern,
            severity="block",
            category="return_promise",
            suggestion="Do not promise returns; use risk disclosure instead.",
            priority=10,
        ),
        actor_id="compliance_test",
    )

    assert created.id > 0
    assert created.rule_code.startswith("CR-")
    assert created.is_active is True

    listed = service.list_rules(keyword=pattern)
    assert listed.total == 1
    assert listed.items[0].pattern == pattern

    updated = service.update_rule(
        created.id,
        ComplianceRuleUpdate(severity="warn", suggestion="Use risk disclosure wording."),
        actor_id="admin_test",
    )
    assert updated.severity == "warn"
    assert updated.updated_by == "admin_test"

    deleted = service.soft_delete_rule(created.id, actor_id="admin_test")
    assert deleted.is_active is False
    assert service.list_rules(keyword=pattern, is_active=True).total == 0
    assert service.list_rules(keyword=pattern, is_active=False).total == 1

def test_compliance_rule_api_enforces_permission_and_exposes_crud_flow():
    advisor_token = login_token("advisor_test", "advisor_test")
    compliance_token = login_token("compliance_test", "compliance_test")
    pattern = f"promise-profit-{uuid4().hex}"

    denied = client.post(
        "/api/advisor-agent/compliance/rules",
        json={
            "rule_type": "keyword",
            "pattern": pattern,
            "severity": "block",
            "category": "return_promise",
            "suggestion": "Do not promise returns.",
        },
        headers={"Authorization": f"Bearer {advisor_token}", "X-Trace-Id": "trace-rule-denied"},
    )
    assert denied.status_code == 403
    assert denied.json()["error_code"] == "AUTH_403_PERMISSION"

    created = client.post(
        "/api/advisor-agent/compliance/rules",
        json={
            "rule_type": "keyword",
            "pattern": pattern,
            "severity": "block",
            "category": "return_promise",
            "suggestion": "Do not promise returns.",
            "priority": 20,
        },
        headers={"Authorization": f"Bearer {compliance_token}", "X-Trace-Id": "trace-rule-create"},
    )
    assert created.status_code == 200
    rule = created.json()["data"]
    assert rule["pattern"] == pattern
    assert rule["rule_code"].startswith("CR-")

    listed = client.get(
        f"/api/advisor-agent/compliance/rules?keyword={pattern}",
        headers={"Authorization": f"Bearer {compliance_token}", "X-Trace-Id": "trace-rule-list"},
    )
    assert listed.status_code == 200
    assert listed.json()["data"]["total"] == 1

    updated = client.put(
        f"/api/advisor-agent/compliance/rules/{rule['id']}",
        json={"severity": "warn", "suggestion": "Use risk disclosure wording."},
        headers={"Authorization": f"Bearer {compliance_token}", "X-Trace-Id": "trace-rule-update"},
    )
    assert updated.status_code == 200
    assert updated.json()["data"]["severity"] == "warn"

    deleted = client.delete(
        f"/api/advisor-agent/compliance/rules/{rule['id']}",
        headers={"Authorization": f"Bearer {compliance_token}", "X-Trace-Id": "trace-rule-delete"},
    )
    assert deleted.status_code == 200
    assert deleted.json()["data"]["is_active"] is False