Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 26 additions & 16 deletions src/skillspector/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -631,7 +631,7 @@ def scan(
)
return
if detection.complete and not detection.has_root_skill and len(detection.skills) == 0:
console.print(
(err_console if format == FormatChoice.json else console).print(
"[yellow]Warning:[/yellow] --recursive specified but no sub-skills "
"detected. Scanning as single skill."
)
Expand All @@ -644,7 +644,7 @@ def scan(
"with a bounded scan and reporting partial coverage."
)
if detection.is_multi_skill:
console.print(
(err_console if format == FormatChoice.json else console).print(
f"[yellow]Warning:[/yellow] Found {len(detection.skills)} skills in "
f"this directory. Use --recursive to scan each independently."
)
Expand Down Expand Up @@ -1864,7 +1864,9 @@ def _scan_skill(
yara_dir = str(yara_rules_dir.resolve()) if yara_rules_dir else None
active_visited: set[str] = set()
if verbose:
console.print("[dim]Running scan...[/dim]")
(err_console if format == FormatChoice.json else console).print(
"[dim]Running scan...[/dim]"
)
logger.debug(
"Scan started: input_path=%s, format=%s, use_llm=%s, transitive=%s",
input_path,
Expand Down Expand Up @@ -2081,7 +2083,10 @@ def _scan_multi_skill(
if yara_dir is None and isinstance(legacy_kwargs.get("yara_rules_dir"), Path):
yara_dir = str(legacy_kwargs["yara_rules_dir"])
skills = detection.skills
console.print(f"[bold]Multi-skill directory detected:[/bold] {len(skills)} skills found\n")
status_console = err_console if format == FormatChoice.json and output is None else console
status_console.print(
f"[bold]Multi-skill directory detected:[/bold] {len(skills)} skills found\n"
)

shared_transitive_cache: dict[str, _CachedTransitiveResult] = {}
shared_transitive_traversal = _TransitiveTraversalState(
Expand Down Expand Up @@ -2130,7 +2135,7 @@ def _scan_multi_skill(
analysis_incomplete = True
aggregate_limitations.extend(shared_transitive_traversal.truncation_reasons)
break
console.print(
status_console.print(
f" [{i}/{len(skills)}] Scanning [bold]{skill.name}[/bold] ({skill.relative_path}/)"
)
try:
Expand Down Expand Up @@ -2204,7 +2209,7 @@ def _scan_multi_skill(
for source in _coerce_str_path_list(result.get("transitive_sources")):
transitive_sources.add(source)
severity = result.get("risk_severity") or "LOW"
console.print(f" Score: {score}/100 ({severity})\n")
status_console.print(f" Score: {score}/100 ({severity})\n")
except Exception as e:
error_message = str(e)[:1_024]
err_console.print(f" [red]Error:[/red] {error_message}\n")
Expand All @@ -2230,33 +2235,35 @@ def _scan_multi_skill(
)
analysis_incomplete = not bool(aggregate_completeness["is_complete"])

console.print("\n[bold]═══ Multi-Skill Summary ═══[/bold]\n")
console.print(
status_console.print("\n[bold]═══ Multi-Skill Summary ═══[/bold]\n")
status_console.print(
f" {'Skill':<30} {'Score':<8} {'Severity':<12} {'Findings':<10} {'Execution':<10}"
)
console.print(f" {'─' * 30} {'─' * 8} {'─' * 12} {'─' * 10} {'─' * 10}")
status_console.print(f" {'─' * 30} {'─' * 8} {'─' * 12} {'─' * 10} {'─' * 10}")

for skill, result in zip(processed_skills, results, strict=True):
if "error" in result:
console.print(f" {skill.name:<30} {'ERROR':<8} {'—':<12} {'—':<10} {'error':<10}")
status_console.print(
f" {skill.name:<30} {'ERROR':<8} {'—':<12} {'—':<10} {'error':<10}"
)
continue
score = result.get("risk_score", 0)
severity = result.get("risk_severity", "LOW")
finding_count = len(effective_findings(result))
execution = "failed" if result.get("execution_successful") is False else "successful"
console.print(
status_console.print(
f" {skill.name:<30} {score:<8} {severity:<12} {finding_count:<10} {execution:<10}"
)
if omitted_skill_count:
console.print(
status_console.print(
f" {'<omitted>':<30} {'—':<8} {'—':<12} {omitted_skill_count:<10} {'partial':<10}"
)
console.print(
status_console.print(
"[yellow]Recursive scan incomplete:[/yellow] one or more skills were omitted "
"after an aggregate safety limit."
)

if output and format == FormatChoice.json:
if format == FormatChoice.json:
combined: dict[str, object] = {
"multi_skill": True,
"skill_count": len(skills),
Expand Down Expand Up @@ -2346,8 +2353,11 @@ def _scan_multi_skill(
}
rendered = json.dumps(combined, indent=2)
_ensure_recursive_output_bound(rendered)
Path(output).write_text(rendered, encoding="utf-8")
console.print(f"[green]Combined report saved to:[/green] {output}")
if output:
Path(output).write_text(rendered, encoding="utf-8")
console.print(f"[green]Combined report saved to:[/green] {output}")
else:
print(rendered)
elif output and format == FormatChoice.sarif:
merged_sarif = _multi_skill_sarif_report(
processed_skills,
Expand Down
152 changes: 152 additions & 0 deletions tests/unit/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -841,6 +841,158 @@ def test_scan_multi_skill_json_output_unchanged(tmp_path: Path) -> None:
assert "skills" in data


def test_scan_multi_skill_json_stdout_is_machine_readable(
tmp_path: Path, capsys: pytest.CaptureFixture
) -> None:
"""Recursive JSON without --output emits only the combined document to stdout."""
s1 = SkillDirectory(path=tmp_path / "skill1", name="skill1", relative_path="skill1")
s2 = SkillDirectory(path=tmp_path / "skill2", name="skill2", relative_path="skill2")
detection = MultiSkillDetectionResult(
is_multi_skill=True, skills=[s1, s2], has_root_skill=False
)
results = [
{
"report_body": json.dumps({"issues": []}),
"risk_score": 10,
"risk_severity": "LOW",
"findings": [],
},
{
"report_body": json.dumps({"issues": []}),
"risk_score": 20,
"risk_severity": "LOW",
"findings": [],
},
]

with patch("skillspector.cli.graph.invoke", side_effect=results):
_scan_multi_skill(
detection,
FormatChoice.json,
None,
no_llm=True,
baseline=None,
show_suppressed=False,
transitive_enabled=False,
transitive_depth=1,
transitive_allow_prefix=(),
transitive_deny_prefix=(),
yara_dir=None,
verbose=True,
)

captured = capsys.readouterr()
payload = json.loads(captured.out)
assert payload["multi_skill"] is True
assert payload["skill_count"] == 2
assert payload["max_risk_score"] == 20
assert "Scanning" not in captured.out
assert "Multi-Skill Summary" not in captured.out
assert "Scanning" in captured.err
assert "Running scan" in captured.err
assert "Multi-Skill Summary" in captured.err


def test_scan_multi_skill_json_stdout_survives_child_failure(
tmp_path: Path, capsys: pytest.CaptureFixture
) -> None:
"""A failed child still emits one parseable combined document before exit 2."""
skills = [
SkillDirectory(path=tmp_path / name, name=name, relative_path=name)
for name in ("healthy", "broken")
]
detection = MultiSkillDetectionResult(is_multi_skill=True, skills=skills)
healthy = {
"report_body": json.dumps({"issues": []}),
"risk_score": 0,
"risk_severity": "LOW",
"findings": [],
}

with (
patch("skillspector.cli.graph.invoke", side_effect=[healthy, RuntimeError("boom")]),
pytest.raises(typer.Exit) as exit_info,
):
_scan_multi_skill(detection, FormatChoice.json, None, no_llm=True)

assert exit_info.value.exit_code == 2
captured = capsys.readouterr()
payload = json.loads(captured.out)
assert payload["execution_successful"] is False
assert payload["skills"][1] == {"name": "broken", "error": "boom"}
assert "Error: boom" in captured.err


def test_recursive_json_single_skill_advisory_does_not_pollute_stdout(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Recursive fallback advisories stay off stdout when JSON is requested."""
monkeypatch.setattr(
cli,
"detect_skills",
lambda _path: MultiSkillDetectionResult(
is_multi_skill=False,
skills=[],
has_root_skill=False,
),
)
monkeypatch.setattr(
cli,
"_scan_skill",
lambda **_kwargs: {
"report_body": json.dumps({"issues": []}),
"risk_score": 0,
},
)

result = runner.invoke(
app,
["scan", str(tmp_path), "--recursive", "--format", "json", "--no-llm"],
)

assert result.exit_code == 0, result.output
assert json.loads(result.stdout) == {"issues": []}
assert "Scanning as single" not in result.stdout
assert "Scanning as single" in result.stderr


def test_json_multi_skill_advisory_does_not_pollute_stdout(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""The non-recursive multi-skill advisory stays off JSON stdout."""
skills = [
SkillDirectory(path=tmp_path / name, name=name, relative_path=name)
for name in ("one", "two")
]
monkeypatch.setattr(
cli,
"detect_skills",
lambda _path: MultiSkillDetectionResult(
is_multi_skill=True,
skills=skills,
has_root_skill=False,
),
)
monkeypatch.setattr(
cli,
"_scan_skill",
lambda **_kwargs: {
"report_body": json.dumps({"issues": []}),
"risk_score": 0,
},
)

result = runner.invoke(
app,
["scan", str(tmp_path), "--format", "json", "--no-llm"],
)

assert result.exit_code == 0, result.output
assert json.loads(result.stdout) == {"issues": []}
assert "Use --recursive" not in result.stdout
assert "Use --recursive" in result.stderr


def test_recursive_detection_limit_reaches_canonical_incomplete_report(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
Expand Down
Loading