OSVauco/test/a2h2a_flow_test.py

import os
import uuid
import httpx
import asyncio
import traceback
import logging
from datetime import datetime, timezone

# --- Test Configuration ---
LOG_FILE = "/home/chris_christiansen/.gemini/tmp/osvauco/test_a2h2a.log"
BASE_URL = "http://localhost:8000"  # Replace with your local server address
TEST_USER_EMAIL = "chris.christiansen@vauco.no"  # Must match authorized_approver in the ticket

# --- Logging Setup ---
# Clear the log file before starting
if os.path.exists(LOG_FILE):
    os.remove(LOG_FILE)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(LOG_FILE),
        logging.StreamHandler()
    ]
)

async def main():
    """Runs an end-to-end test of the A2H2A approval flow."""
    logging.info("--- Starting A2H2A End-to-End Test ---")
    raw_token = None
    ticket_id = f"test-ticket-{uuid.uuid4()}"

    try:
        # 1. Define a sample ticket
        ticket_payload = {
            "ticket_id": ticket_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "severity": "MEDIUM",
            "category": "TESTING",
            "source": {
                "reporter": "test-harness",
                "trigger": "e2e-test-run",
                "affected_service": "opax-mcp",
                "project_id": os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5"),
                "region": "us-central1"
            },
            "context": {
                "summary": "This is an end-to-end test ticket.",
                "evidence_logs": []
            },
            "proposed_action": {
                "action_type": "NO_OP",
                "runbook_reference": "docs/RUNBOOK.md",
                "execution_tool": "example_tool_1",
                "parameters": {"param1": "value1", "param2": True},
                "rollback_plan": "No rollback needed for a test."
            },
            "governance": {
                "authorized_approver": TEST_USER_EMAIL,
                "timeout_minutes": 5
            }
        }

        # 2. Create the ticket in test mode to get the raw token
        logging.info(f"1. Creating ticket {ticket_id}...")
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{BASE_URL}/api/v1/a2h2a/tickets?test_mode=true",
                json=ticket_payload,
                timeout=10
            )
            response.raise_for_status()
            creation_data = response.json()
            raw_token = creation_data.get('raw_token')
            logging.info(f"   Ticket created successfully. Status: {creation_data.get('status')}")
            if not raw_token:
                raise ValueError("Raw token not returned in test mode.")

        # 3. Simulate accessing the review URL (with IAP header)
        logging.info(f"2. Simulating access to review URL...")
        headers = {"X-Goog-Authenticated-User-Email": f"accounts.google.com:{TEST_USER_EMAIL}"}
        async with httpx.AsyncClient() as client:
            review_url = f"{BASE_URL}/a2h2a/review/{ticket_id}?token={raw_token}"
            response = await client.get(review_url, headers=headers, timeout=10)
            response.raise_for_status()
            logging.info(f"   Review page accessed successfully (Status: {response.status_code}).")

        # 4. Simulate approving the ticket
        logging.info(f"3. Approving the ticket...")
        async with httpx.AsyncClient() as client:
            form_data = {"ticket_id": ticket_id, "token": raw_token}
            response = await client.post(
                f"{BASE_URL}/a2h2a/approve",
                data=form_data,
                headers=headers,
                timeout=10
            )
            response.raise_for_status()
            approval_data = response.json()
            logging.info(f"   Ticket approved successfully. Final status: {approval_data.get('status')}")

        # 5. Verify the final state in Firestore (requires google-cloud-firestore)
        logging.info("4. Verifying final state in Firestore...")
        from google.cloud import firestore
        db = firestore.AsyncClient(project=os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5"))

        # Check ticket status
        ticket_ref = db.collection('a2h2a_tickets').document(ticket_id)
        ticket_doc = await ticket_ref.get()
        if ticket_doc.exists:
            ticket_data = ticket_doc.to_dict()
            final_status = ticket_data.get('governance', {}).get('approval_status')
            if final_status == 'APPROVED':
                logging.info(f"   ✅ Ticket status is correctly set to APPROVED.")
            else:
                logging.error(f"   ❌ ERROR: Final ticket status is '{final_status}', not 'APPROVED'.")
        else:
            logging.error("   ❌ ERROR: Ticket document not found in Firestore.")

        # Check audit logs
        audit_ref = db.collection('a2h2a_audit_events').where("ticket_id", "==", ticket_id)
        audit_docs = [doc async for doc in audit_ref.stream()]
        event_types = {doc.to_dict().get('event_type') for doc in audit_docs}
        logging.info(f"   Found {len(event_types)} unique audit events: {event_types}")
        if "PROPOSAL_CREATED" in event_types and "TICKET_APPROVED" in event_types:
            logging.info("   ✅ Correct audit events were created.")
        else:
            logging.error("   ❌ ERROR: Missing required audit events.")

    except httpx.ConnectError as e:
        logging.error("\n❌ CONNECTION ERROR: Could not connect to the server.")
        logging.error(f"   Ensure the OPAX server is running at {BASE_URL}.")
        logging.error(f"   Error details: {e}")
    except httpx.HTTPStatusError as e:
        logging.error(f"\n❌ HTTP STATUS ERROR: {e.response.status_code} on request to {e.request.url}")
        logging.error(f"   Response body: {e.response.text}")
    except Exception as e:
        logging.error("\n❌ UNEXPECTED ERROR during test execution:")
        logging.error(traceback.format_exc())

    logging.info("--- Test Complete ---")

if __name__ == "__main__":
    logging.info("Starting test script...")
    asyncio.run(main())
    logging.info("Test script finished.")