image-tools/tests/test_verify.py

"""Whether a candidate image improves on the pinned one (#1308).

Trivy lists the packages of both, netOrk's matcher in kb-server judges both in
one call (so against one state of its data, kb-server#155), and the verdict is
"improved" when a critical or high CVE with a fix is closed by a patch, or
Debian made a point release. Everything else is "same": the pin stays.

A matcher that cannot be asked is never an improvement: no pin moves on a guess.
"""

from __future__ import annotations

import json
import subprocess

import pytest

from image_tools.verify import Verdict, judge, render, verify

CLOSED = {
    "cve": "CVE-2026-1111",
    "package": "openssl",
    "severity": "high",
    "installed": "3.5.1-1",
    "fixed": "3.5.1-1+deb13u1",
}
NEW = {"cve": "CVE-2026-2222", "package": "nginx", "severity": "critical", "installed": "1.26.3-4"}


def _answer(closed=(), new=()):
    return {
        "closed": list(closed),
        "new": list(new),
        "current": {"critical": 1, "high": 4},
        "candidate": {"critical": 1, "high": 3},
    }


class TestJudge:
    def test_a_closed_cve_is_an_improvement(self):
        verdict = judge(_answer(closed=[CLOSED]), "13.7", "13.7")

        assert verdict.improved
        assert verdict.closed == [CLOSED]
        assert not verdict.point_release

    def test_nothing_closed_is_the_same(self):
        assert not judge(_answer(), "13.7", "13.7").improved

    def test_a_point_release_is_an_improvement_on_its_own(self):
        verdict = judge(_answer(), "13.7", "13.8")

        assert verdict.improved
        assert verdict.point_release == ("13.7", "13.8")

    def test_new_cves_do_not_block_they_are_reported(self):
        verdict = judge(_answer(closed=[CLOSED], new=[NEW]), "13.7", "13.7")

        assert verdict.improved
        assert verdict.new == [NEW]

    def test_new_cves_alone_are_no_improvement(self):
        assert not judge(_answer(new=[NEW]), "13.7", "13.7").improved

    @pytest.mark.parametrize("severity", ["medium", "low", "unknown"])
    def test_only_critical_and_high_count(self, severity):
        """kb-server already filters; a weaker one slipping through moves nothing."""
        closed = dict(CLOSED, severity=severity)

        assert not judge(_answer(closed=[closed]), "13.7", "13.7").improved

    def test_a_closed_cve_without_a_fix_is_not_a_patch(self):
        """Gone without a fixed version means the data changed, not the image."""
        closed = dict(CLOSED, fixed="")

        assert not judge(_answer(closed=[closed]), "13.7", "13.7").improved


class TestVerify:
    def _deps(self, answer=None, error=None, versions=("13.7", "13.7"), failing=((), ())):
        calls = {"sbom": [], "match": []}

        def sbom(ref):
            calls["sbom"].append(ref)
            return {"ArtifactName": ref}

        def match(current, candidate):
            calls["match"].append((current, candidate))
            if error:
                raise error
            return answer

        debian = dict(zip(("cur", "cand"), versions, strict=True))
        cis = dict(zip(("cur", "cand"), map(set, failing), strict=True))
        return calls, {
            "sbom": sbom,
            "match": match,
            "debian_version": lambda ref: debian["cur" if ref == "cur" else "cand"],
            "cis_failing": lambda ref: cis["cur" if ref == "cur" else "cand"],
        }

    def test_both_go_to_the_matcher_in_one_call(self):
        calls, seams = self._deps(answer=_answer(closed=[CLOSED]))

        verdict = verify("cur", "cand", **seams)

        assert calls["sbom"] == ["cur", "cand"]
        assert calls["match"] == [({"ArtifactName": "cur"}, {"ArtifactName": "cand"})]
        assert verdict.improved

    def test_a_matcher_that_cannot_be_asked_moves_nothing(self):
        _, seams = self._deps(error=OSError("connection refused"))

        verdict = verify("cur", "cand", **seams)

        assert not verdict.improved
        assert "connection refused" in verdict.error

    def test_even_a_point_release_waits_for_the_matcher(self):
        _, seams = self._deps(error=OSError("down"), versions=("13.7", "13.8"))

        assert not verify("cur", "cand", **seams).improved

    def test_a_cis_control_the_pin_fails_and_the_candidate_passes_is_an_improvement(self):
        _, seams = self._deps(
            answer=_answer(), failing=({"docker:4.1", "docker:4.8"}, {"docker:4.8"})
        )

        verdict = verify("cur", "cand", **seams)

        assert verdict.improved
        assert verdict.cis_closed == ["docker:4.1"]

    def test_a_cis_control_the_candidate_newly_fails_is_no_improvement(self):
        """The gate in check_image.sh stops such a build before; here it closes nothing."""
        _, seams = self._deps(answer=_answer(), failing=((), {"docker:4.8"}))

        assert not verify("cur", "cand", **seams).improved

    def test_a_cis_measurement_that_cannot_be_taken_moves_nothing(self):
        _, seams = self._deps(answer=_answer(closed=[CLOSED]))

        def broken(ref):
            raise subprocess.CalledProcessError(1, "docker")

        seams["cis_failing"] = broken
        assert not verify("cur", "cand", **seams).improved


def test_the_report_names_what_closed_and_what_is_new():
    verdict = judge(_answer(closed=[CLOSED], new=[NEW]), "13.7", "13.8", ["docker:4.1"])
    text = render(verdict, "cur", "cand")

    assert "CVE-2026-1111" in text and "openssl" in text and "3.5.1-1+deb13u1" in text
    assert "CVE-2026-2222" in text and "nginx" in text
    assert "13.7 → 13.8" in text
    assert "docker:4.1" in text
    assert "improved" in text.lower()


def test_the_verdict_is_json_for_the_workflow():
    verdict = judge(_answer(closed=[CLOSED]), "13.7", "13.7")

    data = json.loads(verdict.to_json())

    assert data["improved"] is True
    assert data["closed"][0]["cve"] == "CVE-2026-1111"


def test_a_verdict_has_no_surprise_fields():
    assert set(Verdict.__dataclass_fields__) == {
        "improved",
        "closed",
        "new",
        "point_release",
        "cis_closed",
        "counts",
        "error",
    }