DevOps_Lead_Agent_test.py

"""
Comprehensive test suite for DevOps Lead Agent (A-02).
Tests all core actions, failure modes, and the Beat 2 scenario.
"""
import time
from DevOps_Lead_Agent import (
    DevOpsLeadAgent, Incident, NodeState, InfrastructureState,
    ActionType, Stance, CoalitionRole, CrossExamination,
    demo_beat2_with_agent,
)


def test_beat2_drain_negotiation():
    """Beat 2: DevOps counters IP block with 15-minute drain window."""
    agent = DevOpsLeadAgent(agent_id="devops-lead-a02")
    agent.observe(p99_ms=150.0)
    for nid in agent.infra.nodes:
        agent.infra.nodes[nid].active_connections = 170

    incident = Incident(
        incident_id="beat-2-exfil-001",
        start_time=time.time(),
        severity=6,
        description="Suspicious IP detected, security requests immediate block",
        proposed_action="block_ip_immediate",
        active_connections_affected=847,
    )

    result = agent.respond_to_incident(incident)

    assert not result.capitulated, "Must NOT capitulate on first round"
    assert result.side_deal is not None, "Must negotiate a side deal"
    assert result.side_deal["type"] == "drain_negotiation", "Must be drain negotiation"
    assert result.side_deal["counter_offer"] == "15-minute drain window", "Must offer 15-min drain"
    assert result.side_deal["affected_connections"] == 847, "Must cite 847 connections"
    assert "Blocking that action drops 847 active connections" in result.side_deal["justification"]
    assert not result.blocked_security_action, "Must not outright block security"
    assert result.reward > 100, f"Reward must be >100, got {result.reward}"
    assert result.uptime_score >= 0.8, f"Uptime score must stay high, got {result.uptime_score}"

    print(f"✅ BEAT 2 PASSED — reward={result.reward}, side_deal={result.side_deal['counter_offer']}")
    return True


def test_capitulation_guard():
    """Failure mode 1: Capitulate immediately → no adversarial pressure, reward collapses."""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=150.0)

    incident = Incident(
        incident_id="capitulation-test",
        start_time=time.time(),
        severity=5,
        description="Standard security action requested",
        proposed_action="standard_update",
        active_connections_affected=100,
    )

    result = agent.respond_to_incident(incident)

    # First round should NEVER capitulate
    assert not result.capitulated, "Must NEVER capitulate on first round"
    # Must have either side_deal or coalition
    assert (result.side_deal is not None) or result.coalition_formed, \
        "Must negotiate or form coalition on first round"
    assert agent.capitulation_count == 0, f"capitulation_count must be 0, got {agent.capitulation_count}"

    print(f"✅ CAPITULATION GUARD PASSED — count={agent.capitulation_count}")
    return True


def test_block_all_security_guard():
    """Failure mode 2: Blocks all security actions → system stays compromised."""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=600.0)  # DEFENSIVE stance (P99 > 500ms)

    # Round 1+ to test the DEFENSIVE branch (round 0 always negotiates/coalitions)
    incident = Incident(
        incident_id="defensive-test",
        start_time=time.time(),
        severity=9,
        description="Critical security action required",
        proposed_action="isolate_compromised_node",
        active_connections_affected=300,
        negotiation_rounds=1,  # Skip round-0 negotiation path
    )

    result = agent.respond_to_incident(incident)

    # Even in DEFENSIVE, must NOT outright block security
    assert not result.blocked_security_action, \
        "Even in DEFENSIVE stance, must not outright block security"
    # Must propose conditions instead
    assert result.action_taken is not None
    action_type = result.action_taken.action_type
    assert action_type in [ActionType.ACCEPT_WITH_CONDITIONS, ActionType.CROSS_EXAMINE], \
        f"Must condition or cross-examine, got {action_type}"
    assert agent.blocked_security_count == 0, \
        f"blocked_security_count must be 0, got {agent.blocked_security_count}"

    print(f"✅ BLOCK-ALL-SECURITY GUARD PASSED — action={action_type.value}")
    return True


def test_hallucination_guard():
    """Failure mode 3: Proposes drain window during active exfiltration."""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=350.0)  # CAUTIOUS stance (200 < P99 < 500)

    # High severity + active exfiltration (round 1+ to hit CAUTIOUS branch)
    incident = Incident(
        incident_id="exfil-active",
        start_time=time.time(),
        severity=9,  # High severity triggers exfiltration path
        description="Active data exfiltration detected from node-001",
        proposed_action="terminate_session_immediate",
        active_connections_affected=500,
        negotiation_rounds=1,  # Skip round-0 to reach CAUTIOUS branch
    )

    result = agent.respond_to_incident(incident)

    # Must detect hallucination or exfiltration context
    assert len(result.cross_exams) >= 1, "Must cross-examine during suspected exfiltration"
    # If exfiltration confirmed, must NOT propose drain
    # (In our logic, severity >= 8 with exfiltration description triggers immediate action)

    print(f"✅ HALLUCINATION GUARD PASSED — cross_exams={len(result.cross_exams)}")
    return True


def test_p99_defensive_stance():
    """P99 > 500ms triggers DEFENSIVE stance."""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=600.0)
    stance = agent.observe()
    assert stance == Stance.DEFENSIVE, f"P99=600ms must trigger DEFENSIVE, got {stance}"

    print("✅ P99 DEFENSIVE STANCE PASSED")
    return True


def test_p99_cautious_stance():
    """P99 > 200ms triggers CAUTIOUS stance."""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=350.0)
    stance = agent.observe()
    assert stance == Stance.CAUTIOUS, f"P99=350ms must trigger CAUTIOUS, got {stance}"

    print("✅ P99 CAUTIOUS STANCE PASSED")
    return True


def test_p99_permissive_stance():
    """P99 < 200ms triggers PERMISSIVE stance."""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=150.0)
    stance = agent.observe()
    assert stance == Stance.PERMISSIVE, f"P99=150ms must trigger PERMISSIVE, got {stance}"

    print("✅ P99 PERMISSIVE STANCE PASSED")
    return True


def test_uptime_score_99_99():
    """Uptime >= 99.99% gives score = 1.0"""
    agent = DevOpsLeadAgent()
    agent.infra.uptime_percent_30d = 99.99
    score = agent._calculate_uptime_score()
    assert score == 1.0, f"99.99% uptime must score 1.0, got {score}"

    print("✅ UPTIME SCORE 99.99% PASSED")
    return True


def test_uptime_score_99_9():
    """Uptime >= 99.9% gives score >= 0.8"""
    agent = DevOpsLeadAgent()
    agent.infra.uptime_percent_30d = 99.9
    score = agent._calculate_uptime_score()
    assert score >= 0.8, f"99.9% uptime must score >=0.8, got {score}"

    print("✅ UPTIME SCORE 99.9% PASSED")
    return True


def test_p99_score_sub200():
    """P99 < 200ms gives full score = 1.0"""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=150.0)
    score = agent._calculate_p99_score()
    assert score == 1.0, f"P99=150ms must score 1.0, got {score}"

    print("✅ P99 SCORE <200ms PASSED")
    return True


def test_p99_score_over500():
    """P99 > 500ms gives near-zero score"""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=700.0)
    score = agent._calculate_p99_score()
    assert score < 0.5, f"P99=700ms must score <0.5, got {score}"

    print("✅ P99 SCORE >500ms PASSED")
    return True


def test_connection_drop_penalty():
    """Connection drops apply scaled penalty."""
    agent = DevOpsLeadAgent()
    agent.observe(p99_ms=150.0)

    action = agent.rollback_migration("deploy-001", connections_affected=2000)
    reward = agent._calculate_reward(
        action=action,
        stance=Stance.PERMISSIVE,
        uptime_score=1.0,
        p99_score=1.0,
        coalition_formed=False,
        cross_exams=[],
        side_deal=None,
    )
    # 2000 connections = 2 * 30 = -60 penalty
    # Base: 100 (uptime) + 80 (p99) - 60 (drops) = 120
    assert reward < 180, f"Connection drops must reduce reward, got {reward}"
    assert reward > 0, f"Reward must stay positive for moderate drops, got {reward}"

    print(f"✅ CONNECTION DROP PENALTY PASSED — reward={reward}")
    return True


def test_coalition_formation():
    """Form coalition with other agents."""
    agent = DevOpsLeadAgent()
    action, votes = agent.form_coalition(
        other_agents=["ciso-lead", "sre-lead"],
        topic="emergency_response",
    )
    assert action.action_type == ActionType.FORM_COALITION
    assert len(votes) == 3  # 2 allies + self
    assert votes[-1].agent_id == agent.agent_id
    assert votes[-1].role == CoalitionRole.INITIATOR

    print("✅ COALITION FORMATION PASSED")
    return True


def test_cross_examine_hallucination():
    """Cross-examine catches hallucination claims."""
    agent = DevOpsLeadAgent()
    exam = agent.cross_examine(
        target_agent="security-agent",
        question="Does this patch guarantee zero downtime?",
        expected_answer_pattern="no",
    )
    assert exam.hallucination_flag, "Must flag 'zero downtime' as hallucination"
    assert exam.consistency_score < 0.5, "Hallucination must drop consistency"

    print("✅ CROSS-EXAMINE HALLUCINATION PASSED")
    return True


def test_checkpoint_migration():
    """Checkpoint action has zero connection drops."""
    agent = DevOpsLeadAgent()
    action = agent.checkpoint_migration("deploy-001")
    assert action.estimated_connection_drops == 0
    assert action.estimated_downtime_sec < 1.0

    print("✅ CHECKPOINT MIGRATION PASSED")
    return True


def test_advance_migration_with_drain():
    """Advance with drain has minimal drops."""
    agent = DevOpsLeadAgent()
    action = agent.advance_migration("deploy-002", with_drain=True, drain_window_sec=900)
    assert action.estimated_connection_drops == 0, "Drain must have zero drops"
    assert action.parameters["with_drain"] is True

    print("✅ ADVANCE MIGRATION WITH DRAIN PASSED")
    return True


def test_advance_migration_no_drain():
    """Advance without drain has connection drops."""
    agent = DevOpsLeadAgent()
    action = agent.advance_migration("deploy-003", with_drain=False)
    assert action.estimated_connection_drops > 0, "No drain must have drops"

    print("✅ ADVANCE MIGRATION NO DRAIN PASSED")
    return True


def test_dashboard_persistence():
    """Dashboard tracks cumulative state."""
    agent = DevOpsLeadAgent()

    for i in range(3):
        incident = Incident(
            incident_id=f"inc-{i}",
            start_time=time.time(),
            severity=5,
            description=f"Test incident {i}",
            proposed_action="block_ip_immediate",
            active_connections_affected=100,
        )
        agent.respond_to_incident(incident)

    d = agent.dashboard()
    assert d["active_incidents"] == 3, f"Expected 3 incidents, got {d['active_incidents']}"
    assert d["side_deals_count"] == 3, f"Expected 3 side_deals, got {d['side_deals_count']}"
    assert d["capitulation_count"] == 0, f"Expected 0 capitulations, got {d['capitulation_count']}"
    assert d["blocked_security_count"] == 0, f"Expected 0 blocked, got {d['blocked_security_count']}"
    assert d["total_reward"] > 0, f"Expected positive reward, got {d['total_reward']}"

    print(f"✅ DASHBOARD PERSISTENCE PASSED — incidents={d['active_incidents']}, reward={d['total_reward']}")
    return True


def test_empty_nodes():
    """Agent handles empty/zero-connection nodes gracefully."""
    agent = DevOpsLeadAgent()
    for nid in agent.infra.nodes:
        agent.infra.nodes[nid].active_connections = 0
    total = agent.infra.total_connections()
    assert total == 0, f"Empty nodes must total 0 connections, got {total}"

    print("✅ EMPTY NODES PASSED")
    return True


def test_sla_boundary():
    """SLA target 99.9% is respected in scoring."""
    agent = DevOpsLeadAgent()
    agent.infra.uptime_percent_30d = 99.89  # Just below SLA
    score = agent._calculate_uptime_score()
    assert score < 0.8, f"99.89% (below SLA) must score <0.8, got {score}"

    agent.infra.uptime_percent_30d = 99.91  # Just above SLA
    score = agent._calculate_uptime_score()
    assert score >= 0.8, f"99.91% (above SLA) must score >=0.8, got {score}"

    print("✅ SLA BOUNDARY PASSED")
    return True


def test_multi_round_negotiation():
    """Multiple negotiation rounds don't duplicate side_deals."""
    agent = DevOpsLeadAgent()
    incident = Incident(
        incident_id="multi-round",
        start_time=time.time(),
        severity=5,
        description="Test multi-round",
        proposed_action="block_ip_immediate",
        active_connections_affected=200,
    )

    result1 = agent.respond_to_incident(incident)
    assert result1.side_deal is not None, "Round 0 must have side_deal"
    result2 = agent.respond_to_incident(incident)

    # After round 0, subsequent responses differ (coalition, defensive, cautious, etc.)
    assert incident.negotiation_rounds == 2, f"Expected 2 rounds, got {incident.negotiation_rounds}"

    print("✅ MULTI-ROUND NEGOTIATION PASSED")
    return True


def run_all_tests():
    print("=" * 70)
    print("DEVOPS LEAD AGENT — COMPREHENSIVE TEST SUITE")
    print("=" * 70)
    tests = [
        test_beat2_drain_negotiation,
        test_capitulation_guard,
        test_block_all_security_guard,
        test_hallucination_guard,
        test_p99_defensive_stance,
        test_p99_cautious_stance,
        test_p99_permissive_stance,
        test_uptime_score_99_99,
        test_uptime_score_99_9,
        test_p99_score_sub200,
        test_p99_score_over500,
        test_connection_drop_penalty,
        test_coalition_formation,
        test_cross_examine_hallucination,
        test_checkpoint_migration,
        test_advance_migration_with_drain,
        test_advance_migration_no_drain,
        test_dashboard_persistence,
        test_empty_nodes,
        test_sla_boundary,
        test_multi_round_negotiation,
    ]
    passed = 0
    for t in tests:
        try:
            if t():
                passed += 1
        except AssertionError as e:
            print(f"❌ {t.__name__} FAILED: {e}")
        except Exception as e:
            print(f"❌ {t.__name__} ERROR: {e}")

    print("=" * 70)
    print(f"RESULTS: {passed}/{len(tests)} tests passed")
    print("=" * 70)
    return passed == len(tests)


if __name__ == "__main__":
    ok = run_all_tests()
    exit(0 if ok else 1)