Function test_engine_fanout_partial_failure

tests/test_recon_engine.py:111–152

One runner failing must not block the others.

(monkeypatch)

def test_engine_fanout_partial_failure(monkeypatch):
    """One runner failing must not block the others."""
    from recon_engine import ReconEngine, ReconType, ReconResult

    engine = ReconEngine.__new__(ReconEngine)
    engine._lock = __import__("threading").Lock()
    engine.active_scans = {}
    engine._tool_paths = {}
    from collections import deque
    engine._scan_history = deque(maxlen=100)

    def ok_runner(target):
        return ReconResult(recon_type=ReconType.TLS_AUDIT, target=target, status="ok")

    def failing_runner(target):
        raise RuntimeError("simulated failure")

    def slow_ok_runner(target):
        time.sleep(0.05)
        return ReconResult(recon_type=ReconType.CONTENT_DISCOVERY, target=target, status="ok")

    monkeypatch.setattr(engine, "_run_tls_audit", ok_runner)
    monkeypatch.setattr(engine, "_run_dns_passive", failing_runner)
    monkeypatch.setattr(engine, "_run_content_discovery", slow_ok_runner)

    from recon_engine import ReconScanState
    state = ReconScanState(
        scan_id="TEST",
        target="https://example.com",
        recon_types=[ReconType.TLS_AUDIT, ReconType.DNS_PASSIVE, ReconType.CONTENT_DISCOVERY],
    )
    engine.active_scans["TEST"] = state

    engine._run_scan("TEST", "https://example.com",
                    [ReconType.TLS_AUDIT, ReconType.DNS_PASSIVE, ReconType.CONTENT_DISCOVERY],
                    timeout=10)

    assert state.status == "completed"
    assert state.results[ReconType.TLS_AUDIT].status == "ok"
    assert state.results[ReconType.DNS_PASSIVE].status == "error"
    assert "simulated failure" in state.results[ReconType.DNS_PASSIVE].error_message
    assert state.results[ReconType.CONTENT_DISCOVERY].status == "ok"


def test_assert_handoff_scope_same_domain():