Skip to content

Commit 3a39dfc

Browse files
committed
test(engine): add tests for fail-fast, timeout, and severity 'none' fix
6 new tests covering: - fail_fast=True aborts after first scanner failure - fail_fast=False continues past failures (existing behavior) - timeout kills slow scanners - timeout=None allows normal completion - severity_threshold "none" string parsed as no threshold - severity_threshold None value parsed as no threshold
1 parent 486a784 commit 3a39dfc

2 files changed

Lines changed: 108 additions & 0 deletions

File tree

argus/tests/test_config.py

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,22 @@ def test_reporting_config_parsed(self):
6464
assert config.reporting.severity_threshold == Severity.MEDIUM
6565
assert config.reporting.output_dir == "/tmp/results"
6666

67+
def test_severity_threshold_none_string_is_no_threshold(self):
68+
"""severity_threshold: 'none' in config means no threshold (not UNKNOWN)."""
69+
data = {
70+
"reporting": {"severity_threshold": "none"},
71+
}
72+
config = ArgusConfig.from_dict(data)
73+
assert config.reporting.severity_threshold is None
74+
75+
def test_severity_threshold_none_value_is_no_threshold(self):
76+
"""severity_threshold: null in YAML means no threshold."""
77+
data = {
78+
"reporting": {"severity_threshold": None},
79+
}
80+
config = ArgusConfig.from_dict(data)
81+
assert config.reporting.severity_threshold is None
82+
6783
def test_non_dict_scanner_entry_skipped(self):
6884
data = {
6985
"scanners": {

argus/tests/test_engine.py

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -280,3 +280,95 @@ def test_auto_backend_falls_back_to_local_no_docker(self, monkeypatch):
280280
assert len(summary.results) == 1
281281
# Docker unavailable, should fall back to local
282282
assert scanner.scan_called_with is not None
283+
284+
285+
class TestFailFast:
286+
"""Test --fail-fast abort-on-first-failure behavior."""
287+
288+
def _make_engine(self):
289+
data = {"scanners": {"a": {"enabled": True}, "b": {"enabled": True}}}
290+
return ArgusEngine(ArgusConfig.from_dict(data))
291+
292+
def test_fail_fast_stops_after_first_failure(self):
293+
engine = self._make_engine()
294+
295+
class FailingScanner:
296+
name = "a"
297+
def scan(self, path, config=None):
298+
raise RuntimeError("boom")
299+
def is_available(self):
300+
return True
301+
def install_command(self):
302+
return None
303+
304+
good = MockScanner("b", findings=[
305+
Finding(id="1", severity=Severity.LOW, title="f1"),
306+
])
307+
engine.register_scanner(FailingScanner())
308+
engine.register_scanner(good)
309+
310+
summary = engine.run(fail_fast=True)
311+
# Scanner "a" fails, "b" should never run
312+
assert len(summary.results) == 0
313+
assert good.scan_called_with is None
314+
315+
def test_without_fail_fast_continues_after_failure(self):
316+
engine = self._make_engine()
317+
318+
class FailingScanner:
319+
name = "a"
320+
def scan(self, path, config=None):
321+
raise RuntimeError("boom")
322+
def is_available(self):
323+
return True
324+
def install_command(self):
325+
return None
326+
327+
good = MockScanner("b", findings=[
328+
Finding(id="1", severity=Severity.LOW, title="f1"),
329+
])
330+
engine.register_scanner(FailingScanner())
331+
engine.register_scanner(good)
332+
333+
summary = engine.run(fail_fast=False)
334+
# Scanner "a" fails, "b" still runs
335+
assert len(summary.results) == 1
336+
assert good.scan_called_with is not None
337+
338+
339+
class TestTimeout:
340+
"""Test per-scanner timeout enforcement."""
341+
342+
def _make_engine(self):
343+
data = {"scanners": {"slow": {"enabled": True}}}
344+
return ArgusEngine(ArgusConfig.from_dict(data))
345+
346+
def test_timeout_raises_on_slow_scanner(self):
347+
import time
348+
349+
engine = self._make_engine()
350+
351+
class SlowScanner:
352+
name = "slow"
353+
def scan(self, path, config=None):
354+
time.sleep(5)
355+
return ScanResult(scanner=self.name)
356+
def is_available(self):
357+
return True
358+
def install_command(self):
359+
return None
360+
361+
engine.register_scanner(SlowScanner())
362+
summary = engine.run(timeout=1)
363+
# Scanner should time out and produce no results
364+
assert len(summary.results) == 0
365+
366+
def test_no_timeout_allows_completion(self):
367+
engine = self._make_engine()
368+
scanner = MockScanner("slow", findings=[
369+
Finding(id="1", severity=Severity.LOW, title="f1"),
370+
])
371+
engine.register_scanner(scanner)
372+
373+
summary = engine.run(timeout=None)
374+
assert len(summary.results) == 1

0 commit comments

Comments
 (0)