verification/api/tests/test_api_hypothesis.py

#!/usr/bin/env python
"""Comprehensive Hypothesis Property-Based Verification Suite for API Subsystem.

Verifies 10 mathematical and structural system invariants across randomized scenarios:
1. Probability & Risk Score Boundedness Invariant
2. Out-Of-Bounds Payload Input Validation Invariant
3. String Length Constraint Enforcement Invariant
4. Enum Query Parameter Robustness Invariant
5. Case Idempotency Key Deduplication Invariant
6. ABAC Cross-Tenant Isolation Invariant
7. Content-Type Enforcement Invariant
8. Cryptographic Audit Chain Invariant
9. Response Header Consistency Invariant
10. Malformed JSON Syntax Handling Invariant
"""
from __future__ import annotations

# ruff: noqa: E402
import os
import sys
from pathlib import Path

import torch

# Add backend directory to sys.path
backend_dir = Path(__file__).parent.parent.parent.parent / "backend"
sys.path.insert(0, str(backend_dir))

import pytest
from fastapi.testclient import TestClient
from hypothesis import HealthCheck, Phase, Verbosity, given, settings
from hypothesis import strategies as st

from app.application.services.idempotency import IdempotencyService
from app.application.services.model_registry import ModelRegistry
from app.application.services.model_service import ModelService
from app.config import get_settings
from app.main import DDoSProtectionMiddleware, app

client = TestClient(app)


@pytest.fixture(autouse=True)
def reset_ddos_limiter():
    DDoSProtectionMiddleware.reset_state()
    yield
    DDoSProtectionMiddleware.reset_state()


def _seed_global_model() -> None:
    """Ensure a verified champion model is active in ModelRegistry for API property tests."""
    from typing import Any, cast

    import numpy as np

    from app.application.services.model_service import NUM_FEATURES

    reg = ModelRegistry()
    if reg.get_champion_metadata(simulation_id="live_prod_v2") is None:
        svc = ModelService(get_settings())
        model = svc.create_model(input_dim=NUM_FEATURES, dp_compatible=True)
        X = np.random.RandomState(42).randn(64, NUM_FEATURES).astype(np.float32)
        y = np.random.RandomState(42).randint(0, 2, size=(64,)).astype(np.float32)
        trained, _, _ = svc.train_local(
            model, cast("Any", X), cast("Any", y), epochs=1, batch_size=32
        )
        trained.eval()
        reg.save_version(
            simulation_id="live_prod_v2",
            state_dict=trained.state_dict(),
            metrics={"auc_roc": 0.91, "f1_score": 0.82, "loss": 0.29},
            is_promoted=True,
            status="champion",
            provenance_type="TRAINING_RECORDED",
        )
    global_path = os.path.join(reg.storage_dir, "global_model.pt")
    if not os.path.exists(global_path):
        svc = ModelService(get_settings())
        model = svc.create_model(dp_compatible=True)
        torch.save(model.state_dict(), global_path)


_seed_global_model()

# Global Hypothesis Settings
settings.register_profile(
    "api_verification",
    max_examples=10,
    deadline=None,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
    verbosity=Verbosity.normal,
    phases=[Phase.explicit, Phase.reuse, Phase.generate, Phase.shrink]
)
settings.load_profile("api_verification")

# ------------------------------------------------------------------------------
# Property 1: Bounded Risk Score Invariant
# ------------------------------------------------------------------------------
@given(
    amount=st.floats(min_value=0.0, max_value=1e6, allow_nan=False, allow_infinity=False),
    merchant=st.text(min_size=1, max_size=64, alphabet=st.characters(whitelist_categories=('Lu', 'Ll', 'Nd'))),
    country=st.text(min_size=2, max_size=2, alphabet=st.characters(whitelist_categories=('Lu',))),
    device=st.sampled_from(["web_browser", "mobile_app", "pos_terminal"]),
    velocity=st.floats(min_value=0.0, max_value=50.0, allow_nan=False, allow_infinity=False),
    hour=st.integers(min_value=0, max_value=23),
    merchant_risk=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
    history_score=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
    chargebacks=st.integers(min_value=0, max_value=100),
    age_days=st.integers(min_value=0, max_value=3650)
)
def test_property_prediction_boundedness_invariant(
    amount, merchant, country, device, velocity, hour, merchant_risk, history_score, chargebacks, age_days
):
    """INVARIANT 1: Valid numerical payloads must produce fraud_probability ∈ [0.0, 1.0] and risk_score ∈ [0, 1000]."""
    payload = {
        "transaction_amount": amount,
        "merchant_category": merchant,
        "country_code": country,
        "device_type": device,
        "velocity": velocity,
        "hour_of_day": hour,
        "merchant_risk_score": merchant_risk,
        "customer_history_score": history_score,
        "chargeback_count": chargebacks,
        "account_age_days": age_days
    }
    r = client.post("/api/v1/predict", json=payload)
    assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
    body = r.json()
    assert 0.0 <= body["fraud_probability"] <= 1.0, f"Probability out of bounds: {body['fraud_probability']}"
    assert 0 <= body["risk_score"] <= 1000, f"Risk score out of bounds: {body['risk_score']}"

# ------------------------------------------------------------------------------
# Property 2: Out-Of-Bounds Validation Invariant
# ------------------------------------------------------------------------------
@given(
    bad_hour=st.one_of(st.integers(max_value=-1), st.integers(min_value=24, max_value=1000)),
    amount=st.floats(min_value=0.0, max_value=1e4, allow_nan=False, allow_infinity=False)
)
def test_property_out_of_bounds_validation_invariant(bad_hour, amount):
    """INVARIANT 2: Out-of-bounds hour_of_day values MUST return HTTP 422 Unprocessable Entity."""
    payload = {
        "transaction_amount": amount,
        "merchant_category": "grocery",
        "country_code": "US",
        "device_type": "mobile_app",
        "velocity": 1.0,
        "hour_of_day": bad_hour,
        "merchant_risk_score": 0.1,
        "customer_history_score": 0.9,
        "chargeback_count": 0,
        "account_age_days": 100
    }
    r = client.post("/api/v1/predict", json=payload)
    assert r.status_code == 422, f"Expected HTTP 422 for bad_hour={bad_hour}, got {r.status_code}"

# ------------------------------------------------------------------------------
# Property 3: String Length Bound Invariant
# ------------------------------------------------------------------------------
@given(
    oversized_str=st.text(min_size=257, max_size=500, alphabet=st.characters(whitelist_categories=('Lu', 'Ll', 'Nd')))
)
def test_property_string_length_bound_invariant(oversized_str):
    """INVARIANT 3: String fields exceeding max_length=256 characters MUST return HTTP 422."""
    payload = {
        "transaction_amount": 100.0,
        "merchant_category": oversized_str,
        "country_code": "US",
        "device_type": "web_browser",
        "velocity": 1.0,
        "hour_of_day": 12,
        "merchant_risk_score": 0.1,
        "customer_history_score": 0.9,
        "chargeback_count": 0,
        "account_age_days": 100
    }
    r = client.post("/api/v1/predict", json=payload)
    assert r.status_code == 422, f"Expected HTTP 422 for string length {len(oversized_str)}, got {r.status_code}"

# ------------------------------------------------------------------------------
# Property 4: Enum Query Parameter Guard Invariant
# ------------------------------------------------------------------------------
@given(
    severity_param=st.text(min_size=1, max_size=50, alphabet=st.characters(whitelist_categories=('Lu', 'Ll')))
)
def test_property_enum_query_parameter_guard_invariant(severity_param):
    """INVARIANT 4: Arbitrary query enum params must return HTTP 200 (if valid enum) or HTTP 422 (if invalid), never 500."""
    r = client.get(f"/api/v1/alerts?severity={severity_param}")
    assert r.status_code in (200, 422), f"Expected 200/422 for severity='{severity_param}', got {r.status_code}: {r.text}"

# ------------------------------------------------------------------------------
# Property 5: Case Idempotency Key Invariant
# ------------------------------------------------------------------------------
@given(
    title=st.text(min_size=5, max_size=50, alphabet=st.characters(whitelist_categories=('Lu', 'Ll', 'Nd', 'Zs'))),
    key=st.uuids().map(str)
)
def test_property_case_idempotency_key_invariant(title, key):
    """INVARIANT 5: Duplicate requests with identical Idempotency-Key return the exact same case_id."""
    IdempotencyService.get().clear()
    case_payload = {"title": title, "priority": "p2_high"}
    headers = {"Idempotency-Key": key}
    r1 = client.post("/api/v1/cases", json=case_payload, headers=headers)
    r2 = client.post("/api/v1/cases", json=case_payload, headers=headers)
    assert r1.status_code == 200 and r2.status_code == 200, f"Expected 200, got {r1.status_code} and {r2.status_code}"
    assert r1.json()["id"] == r2.json()["id"], f"Idempotency key failure: {r1.json()['id']} != {r2.json()['id']}"

# ------------------------------------------------------------------------------
# Property 6: ABAC Cross-Tenant Isolation Invariant
# ------------------------------------------------------------------------------
@given(
    user_bank=st.sampled_from(["bank_a", "bank_b", "bank_c"]),
    resource_bank=st.sampled_from(["bank_x", "bank_y", "bank_z"])
)
def test_property_abac_tenant_isolation_invariant(user_bank, resource_bank):
    """INVARIANT 6: Cross-tenant resource access (user_bank != resource_bank) is strictly denied (allowed: False)."""
    # ABACEvalRequest uses flat fields — not a nested user/resource dict.
    abac_payload = {
        "user_username": "test_user",
        "user_bank_id": user_bank,
        "user_roles": ["analyst"],
        "user_clearance": 2,
        "user_shift_hours": "00:00-24:00",
        "user_approval_tier": 50000.0,
        "resource_type": "api_route",
        "resource_id": "/api/v1/alerts",
        "resource_bank_id": resource_bank,
        "resource_amount": 0.0,
        "resource_classification": 1,
        "action": "read",
    }
    r = client.post("/api/v1/security/abac/evaluate", json=abac_payload)
    assert r.status_code == 200
    assert r.json()["allowed"] is False, f"ABAC isolation failed: user_bank={user_bank}, resource_bank={resource_bank}"

# ------------------------------------------------------------------------------
# Property 7: Content-Type Enforcement Invariant
# ------------------------------------------------------------------------------
@given(
    non_json_ct=st.sampled_from(["text/plain", "application/xml", "image/png", "application/octet-stream"])
)
def test_property_content_type_enforcement_invariant(non_json_ct):
    """INVARIANT 7: Non-JSON content types sent to mutating endpoints MUST return HTTP 415 Unsupported Media Type."""
    r = client.post("/api/v1/predict", content="sample body", headers={"Content-Type": non_json_ct})
    assert r.status_code == 415, f"Expected 415 for Content-Type='{non_json_ct}', got {r.status_code}"

# ------------------------------------------------------------------------------
# Property 8: Audit Chain Hash Integrity Invariant
# ------------------------------------------------------------------------------
@given(
    event_count=st.integers(min_value=1, max_value=5)
)
def test_property_audit_chain_hash_integrity_invariant(event_count):
    """INVARIANT 8: Cryptographic SHA-256 audit chain verification MUST remain valid across appends."""
    r = client.post("/api/v1/security/audit-chain/verify")
    assert r.status_code == 200
    assert r.json()["is_valid"] is True, "Audit chain cryptographic hash integrity check failed!"

# ------------------------------------------------------------------------------
# Property 9: Response Header Metadata Invariant
# ------------------------------------------------------------------------------
@given(
    endpoint=st.sampled_from(["/health", "/api/v1/alerts", "/api/v1/cases"])
)
def test_property_response_header_metadata_invariant(endpoint):
    """INVARIANT 9: API responses MUST contain X-API-Version and traceparent headers."""
    r = client.get(endpoint)
    assert "x-api-version" in r.headers, f"Missing X-API-Version header on {endpoint}"
    assert "traceparent" in r.headers, f"Missing traceparent header on {endpoint}"

# ------------------------------------------------------------------------------
# Property 10: Malformed Syntax Handling Invariant
# ------------------------------------------------------------------------------
@given(
    malformed_str=st.text(
        min_size=1,
        max_size=100,
        alphabet=st.characters(codec="utf-8", blacklist_characters=('"', '{', '}', ':')),
    )
)
def test_property_malformed_json_syntax_handling_invariant(malformed_str):
    """INVARIANT 10: Malformed non-JSON payloads MUST return HTTP 400 or 415/422, NEVER HTTP 500."""
    DDoSProtectionMiddleware.reset_state()
    r = client.post("/api/v1/predict", content=f"{{ {malformed_str} ", headers={"Content-Type": "application/json"})
    assert r.status_code in (400, 415, 422), f"Expected 400/415/422 for malformed payload, got {r.status_code}"
    assert r.status_code != 500, f"Malformed payload caused internal server error 500: {r.text}"


if __name__ == "__main__":
    print("Executing Fast Hypothesis Property-Based Verification Suite (10 examples per property)...")
    test_property_prediction_boundedness_invariant()
    test_property_out_of_bounds_validation_invariant()
    test_property_string_length_bound_invariant()
    test_property_enum_query_parameter_guard_invariant()
    test_property_case_idempotency_key_invariant()
    test_property_abac_tenant_isolation_invariant()
    test_property_content_type_enforcement_invariant()
    test_property_audit_chain_hash_integrity_invariant()
    test_property_response_header_metadata_invariant()
    test_property_malformed_json_syntax_handling_invariant()
    print("ALL 10 HYPOTHESIS PROPERTY INVARIANTS SUCCESSFULLY VERIFIED!")