DataOps-platform/tests/test_data_rule_runtime_sbom.py

from __future__ import annotations

import hashlib
import json
from pathlib import Path

from jsonschema import Draft7Validator
from referencing import Registry, Resource

ROOT = Path(__file__).resolve().parents[1]
SBOM = ROOT / "docs" / "security" / "data-rule-runtime-sbom.json"
SCHEMAS = ROOT / "docs" / "security" / "cyclonedx-1.5-schema"


def _load(name: str):
    return json.loads((SCHEMAS / name).read_text(encoding="utf-8"))


def test_data_rule_runtime_sbom_validates_against_cyclonedx_15_schema():
    expected_schema_digests = {
        "bom-1.5.schema.json": (
            "a00fcba23a44b72179ac1f288be4c3529"
            "b59f6e1b3719709a40685f177516b46"
        ),
        "spdx.schema.json": (
            "c87aa7bb5eb503d40b52ec6bf00de8045"
            "df15da7a13cea48d290cf6d36a8d2ea"
        ),
        "jsf-0.82.schema.json": (
            "2faf5eb3651f2ae5f46091a131770d8d8"
            "47bbd121139d19c85fc7051bfa58c46"
        ),
    }
    for name, expected in expected_schema_digests.items():
        assert hashlib.sha256((SCHEMAS / name).read_bytes()).hexdigest() == (
            expected
        )
    document = json.loads(SBOM.read_text(encoding="utf-8"))
    schema = _load("bom-1.5.schema.json")
    registry = Registry().with_resources(
        [
            (
                "http://cyclonedx.org/schema/spdx.schema.json",
                Resource.from_contents(_load("spdx.schema.json")),
            ),
            (
                "http://cyclonedx.org/schema/jsf-0.82.schema.json",
                Resource.from_contents(_load("jsf-0.82.schema.json")),
            ),
        ]
    )

    Draft7Validator.check_schema(schema)
    errors = sorted(
        Draft7Validator(
            schema,
            registry=registry,
        ).iter_errors(document),
        key=lambda error: list(error.absolute_path),
    )

    assert errors == []


def test_data_rule_runtime_sbom_pins_official_oss_dependencies():
    document = json.loads(SBOM.read_text(encoding="utf-8"))
    components = {
        component["name"]: component
        for component in document["components"]
    }

    assert document["bomFormat"] == "CycloneDX"
    assert document["specVersion"] == "1.5"
    assert set(components) == {
        "minio",
        "polars",
        "polars-runtime-32",
        "psutil",
        "pyarrow",
    }
    for name, version, license_id, source in (
        (
            "polars",
            "1.42.1",
            "MIT",
            "https://github.com/pola-rs/polars",
        ),
        (
            "polars-runtime-32",
            "1.42.1",
            "MIT",
            "https://github.com/pola-rs/polars",
        ),
        (
            "pyarrow",
            "21.0.0",
            "Apache-2.0",
            "https://github.com/apache/arrow",
        ),
        (
            "psutil",
            "5.9.8",
            "BSD-3-Clause",
            "https://github.com/giampaolo/psutil",
        ),
        (
            "minio",
            "7.2.10",
            "Apache-2.0",
            "https://github.com/minio/minio-py",
        ),
    ):
        component = components[name]
        assert component["version"] == version
        assert component["licenses"] == [
            {"license": {"id": license_id}}
        ]
        assert component["externalReferences"] == [
            {"type": "vcs", "url": source}
        ]
        assert component["properties"][0]["name"] == "dataops:purpose"
        assert not {"license", "officialSource", "purpose"} & set(component)

    requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8")
    for requirement in (
        "polars==1.42.1",
        "pyarrow==21.0.0",
        "psutil==5.9.8",
        "minio==7.2.10",
    ):
        assert requirement in requirements