File size: 1,521 Bytes
4524449 75874b2 4524449 75874b2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 | import asyncio
import json
from pathlib import Path
from unittest.mock import patch
from app import pipeline
def test_httpx_batches_targets_and_filters_scope():
calls = []
async def fake(args, timeout):
assert "-l" in args and "-u" not in args
assert "-nfs" in args
payload = Path(args[args.index("-l") + 1]).read_text().splitlines()
calls.append(payload)
assert timeout >= len(payload)
return [json.dumps({"url": url, "status_code": 200}) for url in payload], "completed"
targets = [f"https://example.org/page{i}" for i in range(7)]
targets.append("https://not-example.org/")
with patch.object(pipeline, "execute", fake):
result = asyncio.run(pipeline.probe(targets, "example.org", "Test", limit=10))
assert len(calls) == 1
assert result["submitted"] == 7
assert len(result["probes"]) == 7
assert all(p["url"].startswith("https://example.org/") for p in result["probes"])
def test_httpx_preserves_failed_batch_and_partial_probe_results():
async def fake(args, timeout):
return ['{"url": "https://example.org/a", "status_code": 200}',
"[diagnostic] Deadline exceeded"], "timeout"
with patch.object(pipeline, "execute", fake):
result = asyncio.run(pipeline.probe(["https://example.org/a"],
"example.org", "Test"))
assert result["status"] == "timeout"
assert len(result["probes"]) == 1
assert "Deadline exceeded" in result["reason"] |