Spaces:
Configuration error
Configuration error
File size: 5,586 Bytes
2567e7e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | import pytest
from app.core.security import UserContext
from app.core.document_indexer import DocumentIndexer
from app.core.data_store import DataStore
from app.agent.tools import (
tool_document_search,
tool_structured_data_lookup,
tool_calculate_cancellation_fee,
tool_calculate_service_credit,
tool_prepare_state_action
)
from app.agent.proactive_detector import ProactiveIssueDetector
from app.agent.agent_engine import AgentEngine
@pytest.fixture
def core_services():
indexer = DocumentIndexer()
data_store = DataStore()
engine = AgentEngine(indexer, data_store)
return indexer, data_store, engine
# 1. Access Control Tests
def test_access_control_customer_isolation(core_services):
indexer, data_store, _ = core_services
customer_ctx = UserContext(account_id="ACCT-001", is_internal=False, role="customer")
# Should allow own account
assert customer_ctx.can_access_account("ACCT-001") is True
# Should block other account
assert customer_ctx.can_access_account("ACCT-002") is False
# Document access check: Northstar agreement allowed, LumenWorks agreement denied
assert customer_ctx.can_access_document("05_Northstar_Logistics_Enterprise_Agreement.pdf", "ACCT-001") is True
assert customer_ctx.can_access_document("06_LumenWorks_Service_Agreement.pdf", "ACCT-002") is False
# Search check: Hides LumenWorks agreement from Northstar user search results
search_res = indexer.search_documents("service credit", customer_ctx)
filenames = [d["filename"] for d in search_res]
assert "06_LumenWorks_Service_Agreement.pdf" not in filenames
def test_access_control_internal_full_scope(core_services):
_, data_store, _ = core_services
ops_ctx = UserContext(account_id="ACCT-001", is_internal=True, role="operations_lead")
# Internal user can view all accounts & orders
all_accounts = data_store.get_accounts(ops_ctx)
all_orders = data_store.get_orders(ops_ctx)
assert len(all_accounts) == 4
assert len(all_orders) == 6
# 2. Business Rules & Precedence Tests
def test_northstar_cancellation_fee_waiver(core_services):
indexer, data_store, _ = core_services
ctx = UserContext(account_id="ACCT-001", is_internal=False, role="customer")
# ORD-1001: Booked 2 hours ago (>30 mins). SOP v4 requires 250 fee, but Northstar Contract waives fee!
result = tool_calculate_cancellation_fee("ORD-1001", ctx, data_store, indexer)
assert result["order_status"] == "BOOKED"
assert result["contract_fee_waived"] is True
assert result["final_cancellation_fee_inr"] == 0
assert "05_Northstar_Logistics_Enterprise_Agreement.pdf" in result["governing_source"]
def test_lumenworks_service_credit_threshold(core_services):
indexer, data_store, _ = core_services
ctx = UserContext(account_id="ACCT-002", is_internal=False, role="customer")
# For LumenWorks, 3-hour late pickup is INELIGIBLE because contract requires >4 hours delay!
result = tool_calculate_service_credit("ORD-2002", ctx, data_store, indexer)
assert result["account_name"] == "LumenWorks"
# Delay is 4.5 hours for ORD-2002 (window ended 06:30, snapshot 11:00) -> Eligible for INR 300 fixed credit
assert result["eligible"] is True
assert result["calculated_credit_inr"] == 300.0
assert "06_LumenWorks_Service_Agreement.pdf" in result["governing_source"]
# 3. Proactive Issue Detection Tests
def test_proactive_sla_breach_detection(core_services):
_, data_store, _ = core_services
ops_ctx = UserContext(is_internal=True, role="operations_lead")
detector = ProactiveIssueDetector(data_store)
issues = detector.detect_all_issues(ops_ctx)
sla_breaches = issues["sla_breaches"]
# TKT-501 (Northstar P1 outage) should be flagged as breached (15m SLA target, 30m elapsed)
tkt_501_breach = [b for b in sla_breaches if b["ticket_id"] == "TKT-501"]
assert len(tkt_501_breach) == 1
assert tkt_501_breach[0]["breached"] is True
assert tkt_501_breach[0]["target_sla_minutes"] == 15
def test_proactive_security_alert(core_services):
_, data_store, _ = core_services
ops_ctx = UserContext(is_internal=True, role="operations_lead")
detector = ProactiveIssueDetector(data_store)
issues = detector.detect_all_issues(ops_ctx)
sec_alerts = issues["security_alerts"]
# TKT-505 (API key exposure) should be flagged
tkt_505 = [s for s in sec_alerts if s["ticket_id"] == "TKT-505"]
assert len(tkt_505) == 1
assert "API key" in tkt_505[0]["description"] or "API key" in tkt_505[0]["subject"]
# 4. Agent Engine Multi-step Query & Confirmation Tests
def test_agent_cancellation_query(core_services):
_, _, engine = core_services
ctx = UserContext(account_id="ACCT-001", is_internal=False, role="customer")
prompt = "Can Northstar cancel ORD-1001 without a cancellation fee? Explain why."
res = engine.process_query(prompt, ctx)
assert res["status"] == "SUCCESS"
assert "INR 0" in res["answer"] or "waived" in res["answer"].lower()
assert len(res["trace_steps"]) >= 3
assert len(res["citations"]) > 0
def test_state_changing_action_confirmation(core_services):
_, _, engine = core_services
ctx = UserContext(is_internal=True, role="operations_lead")
prompt = "Escalate ticket TKT-501 to Tier-2 Operations immediately"
res = engine.process_query(prompt, ctx)
assert res["status"] == "PENDING_CONFIRMATION"
assert res["pending_action"]["confirmation_required"] is True
assert res["pending_action"]["action_name"] == "escalate_ticket"
|