End-to-End Testing APIs — Testing Complete Workflows Across Services

import requests
import pytest

BASE = "https://staging.durga-antivirus.com"

class TestThreatManagementWorkflow:
    @pytest.fixture
    def session(self):
        s = requests.Session()
        resp = s.post(f"{BASE}/api/auth/login", json={
            "username": "e2e-tester",
            "password": "e2e-test-password"
        })
        assert resp.status_code == 200
        s.headers["Authorization"] = f"Bearer {resp.json()['access_token']}"
        yield s
        # Cleanup: close any investigations
        s.post(f"{BASE}/api/auth/logout")

    def test_full_threat_management_workflow(self, session):
        investigation_id = None
        threat_id = None
        try:
            # Step 1: Create an investigation
            inv_resp = session.post(f"{BASE}/api/v1/investigations", json={
                "title": "E2E Test Investigation",
                "priority": "high",
                "description": "Automated E2E test"
            })
            assert inv_resp.status_code == 201
            investigation_id = inv_resp.json()["id"]

            # Step 2: Add threat indicators
            indicator_resp = session.post(
                f"{BASE}/api/v1/investigations/{investigation_id}/indicators",
                json={
                    "type": "ip_address",
                    "value": "185.220.101.1",
                    "confidence": "high"
                }
            )
            assert indicator_resp.status_code == 201
            indicator_id = indicator_resp.json()["id"]

            # Step 3: Run threat analysis
            analysis_resp = session.post(
                f"{BASE}/api/v1/investigations/{investigation_id}/analyze"
            )
            assert analysis_resp.status_code == 200
            assert "threat_score" in analysis_resp.json()

            # Step 4: Create threat from investigation
            threat_resp = session.post(f"{BASE}/api/v1/threats", json={
                "investigation_id": investigation_id,
                "name": "E2E Test Threat",
                "severity": "high",
                "source_ip": "185.220.101.1"
            })
            assert threat_resp.status_code == 201
            threat_id = threat_resp.json()["id"]

            # Step 5: Verify threat appears in list
            list_resp = session.get(f"{BASE}/api/v1/threats?severity=high")
            assert list_resp.status_code == 200
            threat_ids = [t["id"] for t in list_resp.json()["threats"]]
            assert threat_id in threat_ids

            print(f"E2E workflow completed: investigation={investigation_id}, threat={threat_id}")

        finally:
            # Cleanup
            if threat_id:
                session.delete(f"{BASE}/api/v1/threats/{threat_id}")
            if investigation_id:
                session.delete(f"{BASE}/api/v1/investigations/{investigation_id}")

import time

class TestAsyncWorkflow:
    def wait_for_status(self, session, url, target_status, timeout=30, interval=2):
        start = time.time()
        while time.time() - start < timeout:
            resp = session.get(url)
            assert resp.status_code == 200
            if resp.json()["status"] == target_status:
                return resp.json()
            time.sleep(interval)
        raise TimeoutError(f"Status did not change to {target_status} within {timeout}s")

    def test_async_threat_scan(self, session):
        # Step 1: Submit scan job
        scan_resp = session.post(f"{BASE}/api/v1/scans", json={
            "target": "example.com",
            "scan_type": "full",
            "callback_url": "https://e2e-test.example.com/callback"
        })
        assert scan_resp.status_code == 202
        scan_id = scan_resp.json()["scan_id"]
        assert scan_resp.json()["status"] == "queued"

        # Step 2: Poll for completion
        result = self.wait_for_status(
            session,
            f"{BASE}/api/v1/scans/{scan_id}",
            "completed"
        )

        # Step 3: Verify scan results
        assert "vulnerabilities" in result
        assert "scan_duration" in result
        assert result["scan_duration"] > 0
        print(f"Async scan {scan_id} completed in {result['scan_duration']}s")

        # Step 4: Verify results endpoint
        results_resp = session.get(f"{BASE}/api/v1/scans/{scan_id}/results")
        assert results_resp.status_code == 200
        assert "findings" in results_resp.json()

import requests

class TestCrossServiceWorkflow:
    """E2E test involving multiple API services."""

    def test_threat_alert_notification_workflow(self, session):
        # Step 1: Auth service — login
        # Step 2: Threat service — create critical threat
        threat_resp = session.post(f"{BASE}/api/v1/threats", json={
            "name": "Critical E2E Test",
            "severity": "critical",
            "source_ip": "10.0.0.99"
        })
        assert threat_resp.status_code == 201
        threat_id = threat_resp.json()["id"]

        # Step 3: Notification service — verify alert was created
        # (May need to poll for async processing)
        time.sleep(2)
        alerts_resp = session.get(
            f"{BASE}/api/v1/alerts",
            params={"threat_id": threat_id}
        )
        assert alerts_resp.status_code == 200
        alerts = alerts_resp.json()["alerts"]
        assert len(alerts) > 0
        assert alerts[0]["severity"] == "critical"
        assert alerts[0]["type"] == "threat_detected"

        # Step 4: Notification service — verify email notification
        email_resp = session.get(
            f"{BASE}/api/v1/notifications",
            params={"type": "email", "reference_id": threat_id}
        )
        email_log = email_resp.json()
        assert len(email_log) > 0
        assert "security-team@durga.com" in email_log[0]["recipients"]

        # Cleanup
        session.delete(f"{BASE}/api/v1/threats/{threat_id}")