Wuhp
Preserve program identity, broaden compatible root-scope pipelines and remove 100-target cap
75874b2 | 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"] |