music-assistant-server

1"""Tests for scoping pip-audit findings to the vulnerabilities a pull request introduces."""
2
3from __future__ import annotations
4
5import json
6from pathlib import Path
7
8import pytest
9
10from scripts.audit_changed_packages import (
11    audit_status,
12    changed_packages,
13    introduced_packages,
14    main,
15    normalize,
16    requirement_versions,
17    resolved_findings,
18    vulnerable_packages,
19)
20
21# Resolved dependency set of the target branch, as the audit workflow writes it
22BASE_CLOSURE = """
23aiohttp==3.14.1
24clean-package==1.0.0
25"""
26
27
28def audit_report(*vulnerable: str, skipped: str | None = None) -> str:
29    """
30    Return a pip-audit JSON report where the given packages carry a vulnerability.
31
32    :param vulnerable: Packages to report a vulnerability for, as `name` or `name==version`.
33    :param skipped: Package name pip-audit could not audit, if any.
34    """
35    dependencies = []
36    for entry in vulnerable:
37        name, _, version = entry.partition("==")
38        dependencies.append(
39            {
40                "name": name,
41                "version": version or "1.0.0",
42                "vulns": [{"id": "GHSA-test", "fix_versions": []}],
43            }
44        )
45    dependencies.append({"name": "clean-package", "version": "1.0.0", "vulns": []})
46    if skipped:
47        dependencies.append({"name": skipped, "skip_reason": "not found on PyPI"})
48    return json.dumps({"dependencies": dependencies, "fixes": []})
49
50
51@pytest.mark.parametrize(
52    ("name", "expected"),
53    [
54        ("aiohttp", "aiohttp"),
55        ("Zeroconf", "zeroconf"),
56        ("aiohttp_fast_zlib", "aiohttp-fast-zlib"),
57        ("zope.interface", "zope-interface"),
58        ("a__b", "a-b"),
59    ],
60)
61def test_normalize(name: str, expected: str) -> None:
62    """Package names are compared in their PEP 503 normalized form."""
63    assert normalize(name) == expected
64
65
66def test_changed_packages_reads_requirement_lines() -> None:
67    """Package names are extracted regardless of the version specifier used."""
68    requirements = """
69        aiohttp==3.14.1
70        music-assistant-models>=1.2
71        some_pkg[extra]~=2.0 ; python_version >= '3.14'
72        aiolibdatachannel @ git+https://github.com/example/aiolibdatachannel@v1
73
74        # a comment
75        --index-url https://example.org/simple
76    """
77    assert changed_packages(requirements) == {
78        "aiohttp",
79        "music-assistant-models",
80        "some-pkg",
81        "aiolibdatachannel",
82    }
83
84
85def test_requirement_versions_reads_pinned_versions() -> None:
86    """A resolved dependency set pins one version per package, except for URL requirements."""
87    closure = """
88        # This file was autogenerated by uv
89        --extra-index-url https://example.org/simple
90        aiohttp==3.14.1
91        some_pkg[extra]==2.0
92        torch==2.13.0+cpu ; sys_platform == 'linux'
93        aiolibdatachannel @ git+https://github.com/example/aiolibdatachannel@v1
94    """
95    assert requirement_versions(closure) == {
96        "aiohttp": {"3.14.1"},
97        "some-pkg": {"2.0"},
98        "torch": {"2.13.0+cpu"},
99        "aiolibdatachannel": set(),
100    }
101
102
103def test_requirement_versions_collects_every_pinned_version() -> None:
104    """A package can appear more than once, as the marker-split torch requirements do."""
105    closure = """
106        torch==2.13.0+cpu ; sys_platform == 'linux' and platform_machine == 'x86_64'
107        torch==2.13.0 ; sys_platform != 'linux' or platform_machine != 'x86_64'
108    """
109    assert requirement_versions(closure) == {"torch": {"2.13.0+cpu", "2.13.0"}}
110
111
112def test_vulnerable_packages_ignores_clean_and_skipped() -> None:
113    """Only packages with findings count; skipped ones carry no vulns key."""
114    report = audit_report("aiohttp==3.14.3", skipped="torch")
115    assert vulnerable_packages(report) == {("aiohttp", "3.14.3")}
116
117
118def test_introduced_packages_compares_name_and_version() -> None:
119    """A finding is introduced unless the target branch resolves to that exact version."""
120    resolved = {"aiohttp": {"3.14.1"}, "aiolibdatachannel": set()}
121    findings = {
122        ("aiohttp", "3.14.1"),
123        ("aiohttp", "3.14.3"),
124        ("transformers", "5.14.1"),
125        ("aiolibdatachannel", "0.1.0"),
126    }
127    assert introduced_packages(findings, resolved) == {
128        ("aiohttp", "3.14.3"),
129        ("transformers", "5.14.1"),
130    }
131
132
133def test_introduced_packages_reads_a_url_pin_from_the_changed_requirements() -> None:
134    """A package the target branch pins by URL is only introduced when its requirement changes."""
135    findings = {("aiolibdatachannel", "0.1.0")}
136    resolved: dict[str, set[str]] = {"aiolibdatachannel": set()}
137
138    assert introduced_packages(findings, resolved) == set()
139    assert introduced_packages(findings, resolved, {"aiolibdatachannel"}) == findings
140
141
142def test_resolved_findings_keeps_what_the_set_installs() -> None:
143    """Only findings the resolved set carries count; a URL requirement matches on name alone."""
144    resolved = {"aiohttp": {"3.14.1"}, "aiolibdatachannel": set()}
145    findings = {
146        ("aiohttp", "3.14.1"),
147        ("aiohttp", "3.14.3"),
148        ("transformers", "5.14.1"),
149        ("aiolibdatachannel", "0.1.0"),
150    }
151    assert resolved_findings(findings, resolved) == {
152        ("aiohttp", "3.14.1"),
153        ("aiolibdatachannel", "0.1.0"),
154    }
155
156
157def test_no_findings_passes() -> None:
158    """A clean audit passes even when the pull request changes dependencies."""
159    assert audit_status(audit_report(), "aiohttp==3.14.1", BASE_CLOSURE) == "pass"
160
161
162def test_findings_the_target_branch_has_too_are_preexisting() -> None:
163    """Vulnerabilities the target branch already installs describe it, not the pull request."""
164    report = audit_report("aiohttp==3.14.1")
165
166    assert audit_status(report, "aiohttp==3.14.1", BASE_CLOSURE) == "preexisting"
167    assert audit_status(report, "aiohttp==3.14.1", BASE_CLOSURE, BASE_CLOSURE) == "preexisting"
168
169
170def test_findings_in_a_bumped_package_fail() -> None:
171    """A version the target branch does not install is one this pull request brings in."""
172    assert audit_status(audit_report("aiohttp==3.14.3"), "aiohttp==3.14.3", BASE_CLOSURE) == "fail"
173
174
175def test_findings_in_a_new_transitive_package_fail() -> None:
176    """A vulnerability pulled in indirectly gates too, even with no direct requirement for it."""
177    assert audit_status(audit_report("transformers==5.14.1"), "beat-this==1.1.0", BASE_CLOSURE) == (
178        "fail"
179    )
180
181
182def test_a_release_published_during_the_run_is_not_gated() -> None:
183    """The environment is installed before either branch is resolved; that skew must not gate."""
184    # Both branches resolve to the release that landed after the environment was installed
185    closure = "aiohttp==3.14.3\nclean-package==1.0.0\n"
186    report = audit_report("aiohttp==3.14.1")
187    changed = "music-assistant-models==1.1.183"
188
189    assert audit_status(report, changed, closure) == "fail"
190    assert audit_status(report, changed, closure, closure) == "preexisting"
191
192
193def test_findings_not_gated_on_are_named(capsys: pytest.CaptureFixture[str]) -> None:
194    """What the pull request's resolution carries instead tells an artifact from a mistake."""
195    closure = "aiohttp==3.14.3\nclean-package==1.0.0\n"
196    report = audit_report("aiohttp==3.14.1", "transformers==5.14.1")
197
198    assert audit_status(report, "music-assistant-models==1.1.183", closure, closure) == (
199        "preexisting"
200    )
201    captured = capsys.readouterr()
202    assert captured.out == ""
203    assert "`aiohttp 3.14.1` (resolved to 3.14.3)" in captured.err
204    assert "`transformers 5.14.1` (not in the resolved set)" in captured.err
205
206
207def test_findings_are_named_alongside_a_gating_one(capsys: pytest.CaptureFixture[str]) -> None:
208    """A pull request that introduces one finding can have another set aside at the same time."""
209    head_closure = "aiohttp==3.14.3\nclean-package==1.0.0\n"
210    report = audit_report("aiohttp==3.14.3", "transformers==5.14.1")
211
212    assert audit_status(report, "aiohttp==3.14.3", BASE_CLOSURE, head_closure) == "fail"
213    captured = capsys.readouterr()
214    assert "`transformers 5.14.1` (not in the resolved set)" in captured.err
215    assert "aiohttp" not in captured.err
216
217
218def test_gated_findings_are_not_named(capsys: pytest.CaptureFixture[str]) -> None:
219    """Nothing is set aside when the pull request's own resolution carries every finding."""
220    head_closure = "aiohttp==3.14.3\nclean-package==1.0.0\n"
221
222    assert audit_status(audit_report("aiohttp==3.14.3"), "", BASE_CLOSURE, head_closure) == "fail"
223    assert capsys.readouterr().err == ""
224
225
226def test_findings_the_pull_request_resolves_to_still_fail() -> None:
227    """A bumped version the pull request's own resolution carries is genuinely introduced."""
228    head_closure = "aiohttp==3.14.3\nclean-package==1.0.0\n"
229    report = audit_report("aiohttp==3.14.3")
230
231    assert audit_status(report, "aiohttp==3.14.3", BASE_CLOSURE, head_closure) == "fail"
232
233
234def test_swapping_a_url_pin_for_a_vulnerable_release_fails() -> None:
235    """Replacing a URL requirement with a published version brings that version in."""
236    base_closure = "aiolibdatachannel @ git+https://github.com/example/aiolibdatachannel@v1"
237    report = audit_report("aiolibdatachannel==0.1.0")
238
239    assert audit_status(report, "aiolibdatachannel==0.1.0", base_closure) == "fail"
240    assert audit_status(report, "", base_closure) == "preexisting"
241
242
243def test_a_changed_url_pin_fails_against_both_resolutions() -> None:
244    """A URL requirement pins no version, so the pull request's resolution matches on name."""
245    base_closure = "aiolibdatachannel @ git+https://github.com/example/aiolibdatachannel@v1"
246    head_closure = "aiolibdatachannel @ git+https://github.com/example/aiolibdatachannel@v2"
247    report = audit_report("aiolibdatachannel==0.1.0")
248
249    assert audit_status(report, head_closure, base_closure, head_closure) == "fail"
250
251
252def test_an_unreadable_target_branch_set_raises() -> None:
253    """A resolved set no package can be read from must fail loudly rather than gate everything."""
254    with pytest.raises(ValueError, match="No packages"):
255        audit_status(audit_report("aiohttp"), "", '{"packages": [{"name": "aiohttp"}]}')
256
257
258def test_an_unreadable_pull_request_set_raises() -> None:
259    """The same holds the other way around: it would clear every finding instead."""
260    with pytest.raises(ValueError, match="No packages"):
261        audit_status(
262            audit_report("aiohttp==3.14.3"),
263            "aiohttp==3.14.3",
264            BASE_CLOSURE,
265            '{"packages": [{"name": "aiohttp"}]}',
266        )
267
268
269def test_findings_without_a_target_branch_set_compare_changed_requirements() -> None:
270    """Resolving the target branch can fail; the changed requirements remain a usable basis."""
271    assert audit_status(audit_report("aiohttp"), "some-pkg==1.0") == "preexisting"
272    assert audit_status(audit_report("aiohttp"), "aiohttp==3.14.1") == "fail"
273
274
275def test_findings_without_dependency_changes_are_preexisting() -> None:
276    """A pull request that changes no dependency can never introduce a vulnerability."""
277    assert audit_status(audit_report("aiohttp"), "") == "preexisting"
278
279
280def test_findings_match_across_name_spellings() -> None:
281    """The comparison survives the underscore/dash spellings both sides use."""
282    assert audit_status(audit_report("aiohttp_fast_zlib"), "aiohttp-fast-zlib==0.3.0") == "fail"
283
284
285def test_main_prints_status(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
286    """The status is printed for the reporting workflow to pick up."""
287    audit = tmp_path / "audit.json"
288    audit.write_text(audit_report("aiohttp==3.14.3"))
289    requirements = tmp_path / "new_deps.txt"
290    requirements.write_text("aiohttp==3.14.3\n")
291    base_closure = tmp_path / "base_closure.txt"
292    base_closure.write_text(BASE_CLOSURE)
293
294    assert main([str(audit), str(requirements), str(base_closure)]) == 0
295    assert capsys.readouterr().out.strip() == "fail"
296
297
298def test_main_reads_the_resolutions_in_order(
299    tmp_path: Path, capsys: pytest.CaptureFixture[str]
300) -> None:
301    """The target branch is the third argument, the pull request's own set the fourth."""
302    audit = tmp_path / "audit.json"
303    audit.write_text(audit_report("aiohttp==3.14.1"))
304    requirements = tmp_path / "new_deps.txt"
305    requirements.write_text("music-assistant-models==1.1.183\n")
306    without = tmp_path / "without.txt"
307    without.write_text("aiohttp==3.14.3\n")
308    carrying = tmp_path / "carrying.txt"
309    carrying.write_text("aiohttp==3.14.1\n")
310
311    assert main([str(audit), str(requirements), str(without), str(carrying)]) == 0
312    assert capsys.readouterr().out.strip() == "fail"
313
314    assert main([str(audit), str(requirements), str(carrying), str(without)]) == 0
315    assert capsys.readouterr().out.strip() == "preexisting"
316
317
318def test_main_keeps_the_status_alone_on_stdout(
319    tmp_path: Path, capsys: pytest.CaptureFixture[str]
320) -> None:
321    """The workflow captures stdout into its status variable, so the note goes to stderr."""
322    audit = tmp_path / "audit.json"
323    audit.write_text(audit_report("aiohttp==3.14.1"))
324    requirements = tmp_path / "new_deps.txt"
325    requirements.write_text("music-assistant-models==1.1.183\n")
326    closure = tmp_path / "closure.txt"
327    closure.write_text("aiohttp==3.14.3\nclean-package==1.0.0\n")
328
329    assert main([str(audit), str(requirements), str(closure), str(closure)]) == 0
330    captured = capsys.readouterr()
331    assert captured.out == "preexisting\n"
332    assert "`aiohttp 3.14.1` (resolved to 3.14.3)" in captured.err
333
334
335def test_main_without_a_base_closure_file(
336    tmp_path: Path, capsys: pytest.CaptureFixture[str]
337) -> None:
338    """The resolved set is absent when the target branch could not be resolved."""
339    audit = tmp_path / "audit.json"
340    audit.write_text(audit_report("transformers==5.14.1"))
341    requirements = tmp_path / "new_deps.txt"
342    requirements.write_text("beat-this==1.1.0\n")
343
344    assert main([str(audit), str(requirements), str(tmp_path / "missing.txt")]) == 0
345    assert capsys.readouterr().out.strip() == "preexisting"
346
347
348def test_main_with_an_empty_base_closure_file(
349    tmp_path: Path, capsys: pytest.CaptureFixture[str]
350) -> None:
351    """An empty resolved set carries no comparison and must not raise."""
352    audit = tmp_path / "audit.json"
353    audit.write_text(audit_report("transformers==5.14.1"))
354    base_closure = tmp_path / "base_closure.txt"
355    base_closure.write_text("")
356
357    assert main([str(audit), str(tmp_path / "missing.txt"), str(base_closure)]) == 0
358    assert capsys.readouterr().out.strip() == "preexisting"
359
360
361def test_main_without_a_requirements_file(
362    tmp_path: Path, capsys: pytest.CaptureFixture[str]
363) -> None:
364    """The requirements file is absent when the analysis found no dependency changes."""
365    audit = tmp_path / "audit.json"
366    audit.write_text(audit_report("aiohttp"))
367
368    assert main([str(audit), str(tmp_path / "missing.txt")]) == 0
369    assert capsys.readouterr().out.strip() == "preexisting"
370
371
372@pytest.mark.parametrize(
373    ("report", "expected"),
374    [
375        ("not json at all", json.JSONDecodeError),
376        # A pip-audit release that renames the key must not read as "no findings"
377        ('{"deps": [], "fixes": []}', KeyError),
378    ],
379)
380def test_an_unusable_report_raises(tmp_path: Path, report: str, expected: type[Exception]) -> None:
381    """An unreadable audit must fail loudly; the workflow gates on a non-zero exit."""
382    audit = tmp_path / "audit.json"
383    audit.write_text(report)
384
385    with pytest.raises(expected):
386        main([str(audit), str(tmp_path / "missing.txt")])
387
388
389def test_a_missing_report_raises(tmp_path: Path) -> None:
390    """pip-audit writing no report at all must fail loudly rather than pass."""
391    with pytest.raises(FileNotFoundError):
392        main([str(tmp_path / "missing.json"), str(tmp_path / "missing.txt")])
393