Spaces:
Running
Running
File size: 3,958 Bytes
cc1ad5a | 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 | from __future__ import annotations
from fastapi.testclient import TestClient
from agents.credit_risk.schema import CreditRiskOutput
from agents.kyc_identity.schema import KycIdentityOutput
from agents.risk_consultant.schema import RiskConsultantOutput
from agents.sanctions_pep.schema import SanctionsPepOutput
from agents.transaction_fraud.schema import TransactionFraudOutput
from api import app
client = TestClient(app)
ORIGIN = {"Origin": "https://example.com"}
VALID_FRAUD = {
"transaction_id": "api-fraud-001",
"amount": 9500.0,
"hour_of_day": 3,
"is_international": True,
"merchant_category": "electronics",
"transaction_velocity_1h": 8,
"amount_vs_avg_ratio": 4.5,
"is_new_device": True,
"distance_from_home_km": 650.0,
"failed_attempts_before": 2,
"account_age_days": 15,
}
VALID_CREDIT = {
"applicant_id": "api-credit-001",
"credit_score": 720,
"debt_to_income_ratio": 0.24,
"employment_months": 48,
"num_open_accounts": 5,
"payment_history_missed": 0,
"loan_amount": 26000.0,
"revolving_utilization": 0.28,
"recent_hard_inquiries": 1,
"collateral_value": 32000.0,
"loan_purpose": "auto",
}
VALID_KYC = {
"application_id": "api-kyc-001",
"id_document_age_days": 365,
"address_match_score": 0.95,
"name_vs_id_match_score": 0.97,
"selfie_liveness_score": 0.96,
"num_accounts_same_address": 1,
"phone_age_days": 400,
"email_domain_risk": 3,
"ip_country_vs_id_country_match": True,
"velocity_applications_7d": 1,
}
VALID_SANCTIONS = {"query_name": "John Smith", "country": ""}
VALID_CONSULTANT = {"question": "What velocity rules are most effective for card fraud?"}
def test_valid_agent_endpoints_and_cors() -> None:
fraud_response = client.post("/api/v1/fraud/predict", json=VALID_FRAUD, headers=ORIGIN)
credit_response = client.post("/api/v1/credit/predict", json=VALID_CREDIT, headers=ORIGIN)
kyc_response = client.post("/api/v1/kyc/predict", json=VALID_KYC, headers=ORIGIN)
sanctions_response = client.post("/api/v1/sanctions/screen", json=VALID_SANCTIONS, headers=ORIGIN)
consultant_response = client.post("/api/v1/consultant/ask", json=VALID_CONSULTANT, headers=ORIGIN)
assert fraud_response.status_code == 200
assert credit_response.status_code == 200
assert kyc_response.status_code == 200
assert sanctions_response.status_code == 200
assert consultant_response.status_code == 200
TransactionFraudOutput.model_validate(fraud_response.json())
CreditRiskOutput.model_validate(credit_response.json())
KycIdentityOutput.model_validate(kyc_response.json())
SanctionsPepOutput.model_validate(sanctions_response.json())
RiskConsultantOutput.model_validate(consultant_response.json())
assert fraud_response.headers["access-control-allow-origin"] in {"*", "https://example.com"}
def test_validation_errors_and_health() -> None:
fraud_bad = client.post("/api/v1/fraud/predict", json={"transaction_id": "missing-fields"})
credit_bad = client.post("/api/v1/credit/predict", json={"applicant_id": "missing-fields"})
kyc_bad = client.post("/api/v1/kyc/predict", json={"application_id": "missing-fields"})
sanctions_bad = client.post("/api/v1/sanctions/screen", json={})
consultant_bad = client.post("/api/v1/consultant/ask", json={})
assert fraud_bad.status_code == 422
assert credit_bad.status_code == 422
assert kyc_bad.status_code == 422
assert sanctions_bad.status_code == 422
assert consultant_bad.status_code == 422
root_health = client.get("/health")
fraud_schema = client.get("/api/v1/fraud/schema")
assert root_health.status_code == 200
assert root_health.json() == {"status": "ok", "agents": ["fraud", "credit", "kyc", "sanctions", "consultant"]}
assert fraud_schema.status_code == 200
assert any(field["name"] == "transaction_id" for field in fraud_schema.json()["fields"])
|