Spaces:
Configuration error
Configuration error
| 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 | |
| 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" | |