# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Unit tests for the report node (risk scoring, output_format, report_body).""" from __future__ import annotations import json import pytest from skillspector.models import Finding from skillspector.nodes.report import ( _DIMINISHING_WEIGHTS, _MAX_OCCURRENCES_PER_RULE, _SEVERITY_POINTS, _compute_risk_score, report, ) from skillspector.sarif_models import validate_sarif_report from skillspector.state import SkillspectorState, llm_call_record from skillspector.suppression import Baseline, SuppressionRule def _finding( rule_id: str, severity: str = "LOW", message: str = "test", confidence: float = 1.0, file: str = "SKILL.md", ) -> Finding: return Finding( rule_id=rule_id, message=message, severity=severity, confidence=confidence, file=file, start_line=1, ) # --- Risk score computation tests --- class TestComputeRiskScoreBasic: """Tests for basic scoring behavior with single findings.""" def test_empty_findings_yields_zero(self) -> None: score, band, rec = _compute_risk_score([], False) assert score == 0 assert band == "LOW" assert rec == "SAFE" @pytest.mark.parametrize( "severity,expected_points", [ ("CRITICAL", 50), ("HIGH", 25), ("MEDIUM", 10), ("LOW", 5), ], ) def test_single_finding_full_confidence_scores_base_points( self, severity: str, expected_points: int ) -> None: findings = [_finding("R1", severity, confidence=1.0)] score, _, _ = _compute_risk_score(findings, False) assert score == expected_points def test_single_finding_partial_confidence_scales_score(self) -> None: findings = [_finding("R1", "HIGH", confidence=0.5)] score, _, _ = _compute_risk_score(findings, False) assert score == 12 # 25 * 1.0 * 0.5 = 12.5 -> int(12.5) = 12 def test_unknown_severity_defaults_to_low_points(self) -> None: f = _finding("R1", "LOW") f.severity = "" score, _, _ = _compute_risk_score([f], False) assert score == 5 class TestComputeRiskScoreDiminishingReturns: """Tests for per-rule diminishing returns logic.""" def test_same_rule_twice_second_scores_half(self) -> None: findings = [ _finding("TM1", "MEDIUM", confidence=1.0), _finding("TM1", "MEDIUM", confidence=1.0), ] score, _, _ = _compute_risk_score(findings, False) # 10*1.0 + 10*0.5 = 15 assert score == 15 def test_same_rule_three_times_third_scores_quarter(self) -> None: findings = [ _finding("TM1", "MEDIUM", confidence=1.0), _finding("TM1", "MEDIUM", confidence=1.0), _finding("TM1", "MEDIUM", confidence=1.0), ] score, _, _ = _compute_risk_score(findings, False) # 10*1.0 + 10*0.5 + 10*0.25 = 17.5 -> 17 assert score == 17 def test_same_rule_beyond_cap_contributes_zero(self) -> None: findings = [_finding("TM1", "MEDIUM", confidence=1.0) for _ in range(10)] score, _, _ = _compute_risk_score(findings, False) # Only first 3 count: 10*1.0 + 10*0.5 + 10*0.25 = 17.5 -> 17 assert score == 17 def test_different_rules_each_score_independently(self) -> None: findings = [ _finding("TM1", "MEDIUM", confidence=1.0), _finding("EA2", "MEDIUM", confidence=1.0), _finding("SQP1", "MEDIUM", confidence=1.0), ] score, _, _ = _compute_risk_score(findings, False) # Each is first occurrence: 10*1.0 + 10*1.0 + 10*1.0 = 30 assert score == 30 def test_mixed_rules_diminishing_applies_per_rule(self) -> None: findings = [ _finding("TM1", "MEDIUM", confidence=1.0), _finding("TM1", "MEDIUM", confidence=1.0), _finding("EA2", "HIGH", confidence=1.0), _finding("EA2", "HIGH", confidence=1.0), ] score, _, _ = _compute_risk_score(findings, False) # TM1: 10*1.0 + 10*0.5 = 15 # EA2: 25*1.0 + 25*0.5 = 37.5 # Total: 52.5 -> 52 assert score == 52 class TestComputeRiskScoreExecutableMultiplier: """Tests for the executable scripts multiplier.""" def test_executable_multiplier_applies(self) -> None: findings = [_finding("R1", "HIGH", confidence=1.0, file="run.py")] component_metadata = [{"path": "run.py", "executable": True}] score, _, _ = _compute_risk_score(findings, True, component_metadata) # 25 * 1.3 = 32.5 -> 32 assert score == 32 def test_executable_multiplier_caps_at_100(self) -> None: findings = [ _finding("C1", "CRITICAL", confidence=1.0), _finding("C2", "CRITICAL", confidence=1.0), _finding("C3", "CRITICAL", confidence=1.0), ] score, _, _ = _compute_risk_score(findings, True) # 50 + 50 + 50 = 150, * 1.3 = 195, capped at 100 assert score == 100 class TestComputeRiskScoreEdgeCases: """Tests for edge cases identified in code review.""" def test_zero_confidence_finding_does_not_consume_weight_slot(self) -> None: """A finding with confidence=0 should be skipped entirely.""" findings = [ _finding("TM1", "HIGH", confidence=0.0), _finding("TM1", "HIGH", confidence=1.0), ] score, _, _ = _compute_risk_score(findings, False) # Zero-confidence skipped, second TM1 is first real occurrence: 25*1.0*1.0 = 25 assert score == 25 def test_negative_confidence_clamped_to_zero_and_skipped(self) -> None: findings = [_finding("R1", "HIGH", confidence=-0.5)] score, _, _ = _compute_risk_score(findings, False) assert score == 0 def test_confidence_above_one_clamped(self) -> None: findings = [_finding("R1", "HIGH", confidence=1.5)] score, _, _ = _compute_risk_score(findings, False) # Clamped to 1.0: 25 * 1.0 * 1.0 = 25 assert score == 25 def test_none_rule_id_bucketed_as_unknown(self) -> None: """Findings with empty/None rule_id all share one bucket.""" f1 = _finding("", "MEDIUM", confidence=1.0) f1.rule_id = "" f2 = _finding("", "MEDIUM", confidence=1.0) f2.rule_id = "" score, _, _ = _compute_risk_score([f1, f2], False) # Both go to "UNKNOWN" bucket: 10*1.0 + 10*0.5 = 15 assert score == 15 def test_same_rule_mixed_severities(self) -> None: """Same rule_id with different severities still uses per-rule diminishing.""" findings = [ _finding("TM1", "CRITICAL", confidence=1.0), _finding("TM1", "LOW", confidence=1.0), ] score, _, _ = _compute_risk_score(findings, False) # First TM1: 50*1.0, second TM1: 5*0.5 = 2.5 -> total 52.5 -> 52 assert score == 52 def test_same_rule_low_before_critical_sorted_correctly(self) -> None: """LOW before CRITICAL in input order must still score as if CRITICAL came first. Without severity sorting, LOW gets the full weight (5*1.0=5) and CRITICAL gets the diminished weight (50*0.5=25), yielding 30. With sorting, CRITICAL gets full weight (50*1.0=50) and LOW gets diminished (5*0.5=2.5), yielding 52. """ findings = [ _finding("TM1", "LOW", confidence=1.0), _finding("TM1", "CRITICAL", confidence=1.0), ] score, _, _ = _compute_risk_score(findings, False) # Sorted: CRITICAL first (50*1.0) + LOW second (5*0.5=2.5) = 52.5 -> 52 assert score == 52 def test_exact_band_boundary_21_is_medium(self) -> None: findings = [ _finding("R1", "MEDIUM", confidence=1.0), _finding("R2", "MEDIUM", confidence=1.0), _finding("R3", "LOW", confidence=0.2), ] score, band, _ = _compute_risk_score(findings, False) # 10 + 10 + 5*1.0*0.2 = 21 assert score == 21 assert band == "MEDIUM" def test_exact_band_boundary_20_is_low(self) -> None: findings = [ _finding("R1", "MEDIUM", confidence=1.0), _finding("R2", "MEDIUM", confidence=1.0), ] score, band, _ = _compute_risk_score(findings, False) # 10 + 10 = 20 assert score == 20 assert band == "LOW" class TestComputeRiskScoreBands: """Tests for severity band assignment.""" def test_score_0_to_20_is_low(self) -> None: findings = [_finding("R1", "MEDIUM", confidence=1.0)] score, band, rec = _compute_risk_score(findings, False) assert score == 10 assert band == "LOW" assert rec == "SAFE" def test_score_21_to_50_is_medium(self) -> None: findings = [ _finding("R1", "HIGH", confidence=1.0), _finding("R2", "LOW", confidence=1.0), ] score, band, rec = _compute_risk_score(findings, False) # 25 + 5 = 30 assert score == 30 assert band == "MEDIUM" assert rec == "CAUTION" def test_score_51_to_80_is_high(self) -> None: findings = [ _finding("R1", "CRITICAL", confidence=1.0), _finding("R2", "MEDIUM", confidence=1.0), ] score, band, rec = _compute_risk_score(findings, False) # 50 + 10 = 60 assert score == 60 assert band == "HIGH" assert rec == "DO_NOT_INSTALL" def test_score_81_plus_is_critical(self) -> None: findings = [ _finding("R1", "CRITICAL", confidence=1.0), _finding("R2", "CRITICAL", confidence=1.0), ] score, band, rec = _compute_risk_score(findings, False) # 50 + 50 = 100 assert score == 100 assert band == "CRITICAL" assert rec == "DO_NOT_INSTALL" class TestComputeRiskScoreRealWorldScenarios: """Tests simulating real-world scanning scenarios from issue #134.""" def test_multi_file_skill_same_rule_does_not_saturate(self) -> None: """A skill using subprocess in 10 files should NOT hit 100.""" findings = [ _finding("TM1", "MEDIUM", confidence=0.5, file=f"step{i}.py") for i in range(10) ] score, band, _ = _compute_risk_score(findings, False) # Only 3 count: 10*1.0*0.5 + 10*0.5*0.5 + 10*0.25*0.5 = 5 + 2.5 + 1.25 = 8.75 -> 8 assert score == 8 assert band == "LOW" def test_diverse_rules_still_accumulate_meaningfully(self) -> None: """Different genuine vulnerabilities should still produce a high score.""" findings = [ _finding("RCE1", "CRITICAL", confidence=0.9), _finding("SQLI", "CRITICAL", confidence=0.85), _finding("XSS", "HIGH", confidence=0.9), _finding("SSRF", "HIGH", confidence=0.8), ] score, band, _ = _compute_risk_score(findings, False) # RCE1: 50*1.0*0.9 = 45 # SQLI: 50*1.0*0.85 = 42.5 # XSS: 25*1.0*0.9 = 22.5 # SSRF: 25*1.0*0.8 = 20 # Total: 130 -> capped at 100 assert score == 100 assert band == "CRITICAL" def test_single_critical_vulnerability_scores_appropriately(self) -> None: """One genuine CRITICAL should register strongly.""" findings = [_finding("RCE1", "CRITICAL", confidence=0.95)] score, band, _ = _compute_risk_score(findings, False) # 50 * 1.0 * 0.95 = 47.5 -> 47 assert score == 47 assert band == "MEDIUM" def test_constants_are_consistent(self) -> None: """Verify module-level constants are in expected ranges.""" assert _MAX_OCCURRENCES_PER_RULE == len(_DIMINISHING_WEIGHTS) assert all(0 < w <= 1.0 for w in _DIMINISHING_WEIGHTS) assert _DIMINISHING_WEIGHTS[0] >= _DIMINISHING_WEIGHTS[-1] for sev in ("CRITICAL", "HIGH", "MEDIUM", "LOW"): assert sev in _SEVERITY_POINTS # --- Report node integration tests --- class TestReportNode: """Tests for the full report() node function.""" def test_report_empty_findings_zero_risk(self) -> None: state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": "/tmp/skill", "output_format": "sarif", } result = report(state) assert result["risk_score"] == 0 assert result["risk_severity"] == "LOW" assert result["risk_recommendation"] == "SAFE" assert "report_body" in result assert "sarif_report" in result def test_report_critical_finding_medium_band(self) -> None: """One CRITICAL finding at confidence 1.0 yields score 50, MEDIUM band.""" state: SkillspectorState = { "filtered_findings": [_finding("P5", "CRITICAL", confidence=1.0)], "component_metadata": [ { "path": "SKILL.md", "type": "markdown", "lines": 10, "executable": False, "size_bytes": 100, } ], "has_executable_scripts": False, "manifest": {"name": "test"}, "skill_path": "/tmp/skill", "output_format": "json", } result = report(state) assert result["risk_score"] == 50 assert result["risk_severity"] == "MEDIUM" assert result["risk_recommendation"] == "CAUTION" def test_report_high_severity_do_not_install(self) -> None: """Score >= 51 yields severity HIGH and DO_NOT_INSTALL.""" state: SkillspectorState = { "filtered_findings": [ _finding("P5", "CRITICAL", confidence=1.0), _finding("E2", "MEDIUM", confidence=1.0), ], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": None, "output_format": "json", } result = report(state) # 50 + 10 = 60 => HIGH band assert result["risk_score"] == 60 assert result["risk_severity"] == "HIGH" assert result["risk_recommendation"] == "DO_NOT_INSTALL" def test_report_executable_scripts_multiplier(self) -> None: """has_executable_scripts applies 1.3x to risk score.""" state: SkillspectorState = { "filtered_findings": [ _finding("E2", "HIGH", confidence=1.0, file="run.py"), _finding("PE3", "HIGH", confidence=1.0, file="run.py"), ], "component_metadata": [ { "path": "run.py", "type": "python", "lines": 5, "executable": True, "size_bytes": 200, } ], "has_executable_scripts": True, "manifest": {}, "skill_path": "/tmp/skill", "output_format": "json", } result = report(state) # (25 + 25) * 1.3 = 65 assert result["risk_score"] == 65 assert result["risk_severity"] == "HIGH" assert result["risk_recommendation"] == "DO_NOT_INSTALL" def test_report_output_format_json(self) -> None: """output_format json produces valid JSON with expected structure.""" state: SkillspectorState = { "filtered_findings": [_finding("P1", "HIGH", confidence=1.0)], "component_metadata": [ { "path": "a.md", "type": "markdown", "lines": 1, "executable": False, "size_bytes": 10, } ], "has_executable_scripts": False, "manifest": {"name": "my-skill"}, "skill_path": "/path/to/skill", "output_format": "json", } result = report(state) body = result["report_body"] data = json.loads(body) assert data["skill"]["name"] == "my-skill" assert "risk_assessment" in data assert "score" in data["risk_assessment"] assert "severity" in data["risk_assessment"] assert "recommendation" in data["risk_assessment"] assert "components" in data assert "issues" in data assert len(data["issues"]) == 1 assert data["issues"][0]["id"] == "P1" def test_report_output_format_markdown(self) -> None: """output_format markdown produces expected headings.""" state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": None, "output_format": "markdown", } result = report(state) body = result["report_body"] assert "# SkillSpector Security Report" in body assert "## Risk Assessment" in body assert "## Components" in body assert "## Issues" in body def test_report_output_format_terminal(self) -> None: """output_format terminal produces Rich-formatted output.""" state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {"name": "cli-test"}, "skill_path": "/foo", "output_format": "terminal", } result = report(state) body = result["report_body"] assert "SkillSpector" in body assert "Risk Assessment" in body assert "cli-test" in body def test_report_output_format_sarif(self) -> None: """output_format sarif produces valid SARIF JSON.""" state: SkillspectorState = { "filtered_findings": [_finding("E2", "HIGH", "env harvest", confidence=1.0)], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": None, "output_format": "sarif", } result = report(state) body = result["report_body"] data = json.loads(body) assert "runs" in data assert data.get("$schema") or "runs" in data def test_report_default_output_format_is_sarif(self) -> None: """When output_format is missing, report uses sarif.""" state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, } result = report(state) body = result["report_body"] json.loads(body) assert "sarif_report" in result def test_report_dedup_affects_score_only_not_report_output(self) -> None: """Deduplication reduces score but all affected files appear in the report.""" duplicated = [ Finding( rule_id="TM1", message="shell injection", severity="HIGH", confidence=0.8, file=f"step{i}.py", start_line=10, matched_text="subprocess.run(cmd, shell=True)", ) for i in range(4) ] state: SkillspectorState = { "filtered_findings": duplicated, "component_metadata": [], "has_executable_scripts": False, "manifest": {"name": "multi-file"}, "skill_path": "/tmp/skill", "output_format": "json", } result = report(state) body = json.loads(result["report_body"]) reported_files = {issue["location"]["file"] for issue in body["issues"]} assert reported_files == {"step0.py", "step1.py", "step2.py", "step3.py"} assert len(body["issues"]) == 4 assert result["risk_score"] < 4 * 25 def test_report_baseline_suppresses_finding_and_lowers_score() -> None: """A baseline-suppressed CRITICAL finding does not count toward the risk score.""" baseline = Baseline(rules=[SuppressionRule(rule_id="P5", reason="false positive")]) state: SkillspectorState = { "filtered_findings": [_finding("P5", "CRITICAL")], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": None, "output_format": "json", "baseline": baseline, } result = report(state) assert result["risk_score"] == 0 assert result["risk_severity"] == "LOW" assert result["risk_recommendation"] == "SAFE" # Suppressed findings stay in SARIF but are marked with `suppressions` # (audit trail) so consumers exclude them from counts. sarif_results = result["sarif_report"]["runs"][0]["results"] assert len(sarif_results) == 1 assert sarif_results[0]["suppressions"][0]["kind"] == "external" assert len(result["suppressed_findings"]) == 1 def test_report_baseline_keeps_unmatched_finding() -> None: """Findings not matched by the baseline are kept and scored normally.""" baseline = Baseline(rules=[SuppressionRule(rule_id="SQP-1", reason="nit")]) state: SkillspectorState = { "filtered_findings": [_finding("P5", "CRITICAL"), _finding("SQP-1", "MEDIUM")], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": None, "output_format": "json", "baseline": baseline, } result = report(state) assert result["risk_score"] == 50 # only the CRITICAL counts assert len(result["suppressed_findings"]) == 1 def test_report_json_includes_suppressed_section() -> None: """JSON output reports suppressed_count and a suppressed array.""" baseline = Baseline(rules=[SuppressionRule(rule_id="P5", reason="fp")]) state: SkillspectorState = { "filtered_findings": [_finding("P5", "CRITICAL")], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": None, "output_format": "json", "baseline": baseline, } data = json.loads(report(state)["report_body"]) assert data["suppressed_count"] == 1 assert data["issues"] == [] assert data["suppressed"][0]["suppression_reason"] == "fp" def test_report_markdown_show_suppressed_lists_rows() -> None: """Markdown lists suppressed findings only when show_suppressed is set.""" baseline = Baseline(rules=[SuppressionRule(rule_id="P5", reason="fp")]) base_state: SkillspectorState = { "filtered_findings": [_finding("P5", "CRITICAL")], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": None, "output_format": "markdown", "baseline": baseline, } hidden = report({**base_state})["report_body"] assert "## Suppressed (1)" in hidden assert "--show-suppressed" in hidden shown = report({**base_state, "show_suppressed": True})["report_body"] assert "## Suppressed (1)" in shown assert "fp" in shown def test_report_no_baseline_unchanged() -> None: """Without a baseline, scoring is unchanged and nothing is suppressed.""" state: SkillspectorState = { "filtered_findings": [_finding("P5", "CRITICAL")], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "skill_path": None, "output_format": "json", } result = report(state) assert result["risk_score"] == 50 assert result["suppressed_findings"] == [] # --------------------------------------------------------------------------- # LLM degradation signal (use_llm requested but every LLM call failed) # --------------------------------------------------------------------------- def _meta_from_json_report(state: SkillspectorState) -> dict: """Run the report node in JSON mode and return the metadata block.""" return json.loads(report(state)["report_body"])["metadata"] def test_report_llm_degraded_when_all_calls_failed(monkeypatch: pytest.MonkeyPatch) -> None: """use_llm requested + every LLM call failed -> llm_available False, llm_degraded True.""" # Pre-flight reports available (binary/creds present); the failure is at runtime. monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "json", "use_llm": True, "llm_call_log": [ llm_call_record("semantic_security_discovery", ok=False, error="claude empty stdout"), llm_call_record("semantic_developer_intent", ok=False, error="claude empty stdout"), llm_call_record("semantic_quality_policy", ok=False, error="boom"), ], } meta = _meta_from_json_report(state) assert meta["llm_requested"] is True assert meta["llm_available"] is False # degraded -> not actually available assert meta["llm_degraded"] is True assert meta["llm_calls_attempted"] == 3 assert meta["llm_calls_succeeded"] == 0 # Distinct error reasons are surfaced (deduped). assert "claude empty stdout" in meta["llm_error"] assert "static analysis only" in meta["llm_error"] def test_report_not_degraded_when_some_calls_succeeded(monkeypatch: pytest.MonkeyPatch) -> None: """At least one successful LLM call -> not degraded, llm_available stays True.""" monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "json", "use_llm": True, "llm_call_log": [ llm_call_record("semantic_security_discovery", ok=True), llm_call_record("semantic_quality_policy", ok=False, error="boom"), ], } meta = _meta_from_json_report(state) assert meta["llm_available"] is True assert "llm_degraded" not in meta assert meta["llm_calls_attempted"] == 2 assert meta["llm_calls_succeeded"] == 1 def test_report_not_degraded_when_no_llm_calls(monkeypatch: pytest.MonkeyPatch) -> None: """use_llm True but no LLM calls attempted (e.g. empty skill) -> not degraded.""" monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "json", "use_llm": True, "llm_call_log": [], } meta = _meta_from_json_report(state) assert meta["llm_available"] is True assert "llm_degraded" not in meta assert "llm_calls_attempted" not in meta def test_report_no_llm_failures_not_counted_as_degraded(monkeypatch: pytest.MonkeyPatch) -> None: """use_llm False -> failures (if any) never mark the scan degraded.""" monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "json", "use_llm": False, "llm_call_log": [llm_call_record("meta_analyzer", ok=False, error="boom")], } meta = _meta_from_json_report(state) assert "llm_degraded" not in meta def test_report_terminal_shows_degraded_warning(monkeypatch: pytest.MonkeyPatch) -> None: """Terminal output surfaces a visible degraded-scan warning.""" monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {"name": "t"}, "output_format": "terminal", "use_llm": True, "llm_call_log": [llm_call_record("semantic_quality_policy", ok=False, error="boom")], } body = report(state)["report_body"] assert "Degraded scan" in body assert "STATIC analysis only" in body def test_report_markdown_shows_degraded_warning(monkeypatch: pytest.MonkeyPatch) -> None: """Markdown output surfaces a visible degraded-scan warning.""" monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "markdown", "use_llm": True, "llm_call_log": [llm_call_record("meta_analyzer", ok=False, error="boom")], } body = report(state)["report_body"] assert "Degraded scan" in body def test_report_sarif_carries_degradation_notification() -> None: """The default SARIF output surfaces degradation via a tool-execution notification.""" state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "sarif", "use_llm": True, "llm_call_log": [ llm_call_record("semantic_security_discovery", ok=False, error="claude empty stdout"), ], } result = report(state) run = result["sarif_report"]["runs"][0] assert "invocations" in run invocation = run["invocations"][0] assert invocation["executionSuccessful"] is True # scan completed; LLM sub-stage degraded notification = invocation["toolExecutionNotifications"][0] assert notification["level"] == "warning" assert "STATIC analysis only" in notification["message"]["text"] # The serialized report_body carries it too, and the doc stays schema-valid. body = json.loads(result["report_body"]) assert body["runs"][0]["invocations"][0]["toolExecutionNotifications"] validate_sarif_report(result["sarif_report"]) def test_report_sarif_no_invocations_when_not_degraded() -> None: """A healthy scan's SARIF output is unchanged (no invocations block).""" state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "sarif", "use_llm": True, "llm_call_log": [llm_call_record("semantic_security_discovery", ok=True)], } result = report(state) assert "invocations" not in result["sarif_report"]["runs"][0] # --------------------------------------------------------------------------- # Fail-closed: a degraded deep scan must not be able to report SAFE # --------------------------------------------------------------------------- def test_degraded_scan_floors_recommendation_at_caution() -> None: """No findings would normally be SAFE; a degraded LLM stage forces CAUTION.""" state: SkillspectorState = { "filtered_findings": [], # static score 0 -> would be SAFE "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "json", "use_llm": True, "llm_call_log": [llm_call_record("semantic_security_discovery", ok=False, error="boom")], } result = report(state) assert result["risk_score"] == 0 # score is left honest assert result["risk_recommendation"] == "CAUTION" # but never SAFE when degraded def test_non_degraded_clean_scan_stays_safe() -> None: """Without degradation, a clean scan still reports SAFE (no over-flooring).""" state: SkillspectorState = { "filtered_findings": [], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "json", "use_llm": True, "llm_call_log": [llm_call_record("semantic_security_discovery", ok=True)], } result = report(state) assert result["risk_recommendation"] == "SAFE" def test_degraded_scan_does_not_downgrade_a_blocking_verdict() -> None: """A degraded scan that is already DO_NOT_INSTALL stays blocking (floor only lifts SAFE).""" state: SkillspectorState = { "filtered_findings": [_finding("P5", "CRITICAL"), _finding("P6", "CRITICAL")], "component_metadata": [], "has_executable_scripts": False, "manifest": {}, "output_format": "json", "use_llm": True, "llm_call_log": [llm_call_record("meta_analyzer", ok=False, error="boom")], } result = report(state) assert result["risk_recommendation"] == "DO_NOT_INSTALL" def test_report_executable_scripts_multiplier() -> None: """1.3x multiplier applied only to findings from executable files.""" # 2 HIGH findings in run.py = 2 × 25 × 1.3 = 65 (float-based accumulation) state: SkillspectorState = { "filtered_findings": [ _finding("E2", "HIGH", file="run.py"), _finding("PE3", "HIGH", file="run.py"), ], "component_metadata": [ {"path": "run.py", "type": "python", "lines": 5, "executable": True, "size_bytes": 200} ], "has_executable_scripts": True, "manifest": {}, "skill_path": "/tmp/skill", "output_format": "json", } result = report(state) assert result["risk_score"] == 65 assert result["risk_severity"] == "HIGH" assert result["risk_recommendation"] == "DO_NOT_INSTALL" def test_report_doc_findings_no_multiplier() -> None: """Findings from non-executable files (markdown/docs) are not multiplied.""" # 2 HIGH in SKILL.md (non-executable) = 2 × 25 = 50 (no 1.3x) state: SkillspectorState = { "filtered_findings": [ _finding("P1", "HIGH", file="SKILL.md"), _finding("P2", "HIGH", file="SKILL.md"), ], "component_metadata": [ { "path": "SKILL.md", "type": "markdown", "lines": 10, "executable": False, "size_bytes": 500, }, {"path": "run.py", "type": "python", "lines": 5, "executable": True, "size_bytes": 200}, ], "has_executable_scripts": True, "manifest": {}, "skill_path": "/tmp/skill", "output_format": "json", } result = report(state) # Without the multiplier: 2 HIGH = 50, not 65 assert result["risk_score"] == 50 assert result["risk_severity"] == "MEDIUM"