You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by sb...@apache.org on 2026/09/09 18:04:00 UTC
(tooling-trusted-releases) branch sbp updated: Add a task for generating heatmaps
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a commit to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
The following commit(s) were added to refs/heads/sbp by this push:
new 7e06fda9 Add a task for generating heatmaps
7e06fda9 is described below
commit 7e06fda9e9d5ce3176da2382fcbd479b83231e78
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Wed Sep 9 19:03:27 2026 +0100
Add a task for generating heatmaps
---
atr/catalog_site.py | 42 +++++-
atr/models/args.py | 5 +
atr/models/sql.py | 3 +
atr/sbom/heatmap.py | 62 +++++----
atr/shared/catalog.py | 4 +
atr/tasks/__init__.py | 3 +
atr/tasks/heatmap.py | 85 ++++++++++++
atr/tasks/task.py | 1 +
atr/worker.py | 10 ++
tests/unit/test_heatmap_tasks.py | 283 +++++++++++++++++++++++++++++++++++++++
tests/unit/test_sbom_heatmap.py | 43 +++++-
11 files changed, 508 insertions(+), 33 deletions(-)
diff --git a/atr/catalog_site.py b/atr/catalog_site.py
index 2e357816..2a048869 100644
--- a/atr/catalog_site.py
+++ b/atr/catalog_site.py
@@ -35,6 +35,7 @@ import pathlib
import re
import shutil
import urllib.parse
+import uuid
from collections.abc import Awaitable, Container, Iterable, Iterator, Sequence
from typing import Any, Final
@@ -55,6 +56,7 @@ import atr.models.results as results
import atr.models.safe as safe
import atr.models.sql as sql
import atr.paths as paths
+import atr.sbom.heatmap as heatmap
import atr.shared.catalog as catalog
import atr.util as util
@@ -233,16 +235,27 @@ async def regenerate_project(data: db.Session, project_key: str) -> None:
log.warning(f"Catalog site: committee {project.committee_key} not found, skipping regeneration")
return
site_dir = paths.get_catalog_site_dir()
+ await _write_assets(site_dir)
await _write_committee_index(data, committee, site_dir)
# A project sharing its committee's key is the committee page, so it's written already
if project.key != committee.key:
await _write_project(data, committee, project, site_dir)
# A first release, or a last one archived, moves a committee on or off the front page
# and between the Incubator's two columns, so the indexes are rebuilt every time too.
- # The assets are not: they only change with a deploy, so the full rebuild owns them.
await _write_index_pages(data, site_dir, committees)
+def _applicable_heatmap(
+ snapshot: results.SBOMHeatmap | None, version: api.CatalogVersion
+) -> results.SBOMHeatmap | None:
+ if (snapshot is None) or (snapshot.analysis_version != heatmap.ANALYSIS_VERSION):
+ return None
+ sboms = catalog.sbom_urls(version)
+ if sboms and (sboms == {sbom.artifact_path: sbom.sbom_url for sbom in snapshot.sboms}):
+ return snapshot
+ return None
+
+
def _artifact_classifier_combos(
version: api.CatalogVersion, pmc: str, project: str
) -> Iterator[tuple[dict[str, str], str]]:
@@ -399,6 +412,22 @@ def _heatmap_url(url: str | None) -> str | None:
return url if (parsed.scheme in {"http", "https"}) and parsed.netloc else None
+async def _heatmaps(
+ data: db.Session, project_key: str, releases: Sequence[sql.Release]
+) -> dict[str, results.SBOMHeatmap]:
+ published = {release.version for release in releases if release.phase == sql.ReleasePhase.RELEASE}
+ completed = (
+ await data.task(task_type=sql.TaskType.SBOM_HEATMAP, status=sql.TaskStatus.COMPLETED, project_key=project_key)
+ .order_by(sql.Task.completed, sql.Task.id)
+ .all()
+ )
+ return {
+ task.version_key: task.result
+ for task in completed
+ if (task.version_key in published) and isinstance(task.result, results.SBOMHeatmap)
+ }
+
+
def _htaccess_cond(qualifier: str, value: str) -> str:
"""A RewriteCond matching `<qualifier>=<value>` anywhere in the query string."""
# Match the raw (percent-encoded) query string, regex-escaped so its metacharacters
@@ -519,7 +548,12 @@ async def _write_assets(site_dir: safe.StatePath) -> None:
for rel in _ASSETS:
destination = assets_root / rel
await aiofiles.os.makedirs(destination.parent, exist_ok=True)
- await asyncio.to_thread(shutil.copyfile, _STATIC_DIR / rel, destination)
+ temporary = destination.with_name(f".{destination.name}.{uuid.uuid4()}.tmp")
+ try:
+ await asyncio.to_thread(shutil.copyfile, _STATIC_DIR / rel, temporary)
+ await aiofiles.os.replace(temporary, destination)
+ finally:
+ await asyncio.to_thread(temporary.unlink, missing_ok=True)
async def _write_attic_index(
@@ -735,9 +769,11 @@ async def _write_project(
root=root,
),
)
+ heatmaps = await _heatmaps(data, project.key, releases)
for version in assembled.versions:
document = release_documents.get(str(version.version))
- await _write_release(project_dir, committee, project, version, f"{root}../", document)
+ snapshot = _applicable_heatmap(heatmaps.get(str(version.version)), version)
+ await _write_release(project_dir, committee, project, version, f"{root}../", document, snapshot)
await _prune_directories(project_dir, {str(version.version) for version in assembled.versions})
diff --git a/atr/models/args.py b/atr/models/args.py
index 0a32f81b..c782e16d 100644
--- a/atr/models/args.py
+++ b/atr/models/args.py
@@ -182,6 +182,11 @@ class ReleaseFinalise(schema.Strict):
audit_until: str = schema.description("Inclusive audit log datetime boundary for the release log")
+class SBOMHeatmap(schema.Strict):
+ project_key: safe.ProjectKey
+ version_key: safe.VersionKey
+
+
class Send(schema.Strict):
"""Arguments for the task to send an email."""
diff --git a/atr/models/sql.py b/atr/models/sql.py
index e6c0ea95..c1c5130e 100644
--- a/atr/models/sql.py
+++ b/atr/models/sql.py
@@ -285,6 +285,7 @@ class TaskType(enum.StrEnum):
SBOM_CONVERT = "sbom_convert"
SBOM_GENERATE = "sbom_generate"
SBOM_GENERATE_CYCLONEDX = "sbom_generate_cyclonedx"
+ SBOM_HEATMAP = "sbom_heatmap"
SBOM_OSV_SCAN = "sbom_osv_scan"
SBOM_QS_SCORE = "sbom_qs_score"
SBOM_TOOL_SCORE = "sbom_tool_score"
@@ -349,6 +350,8 @@ class TaskType(enum.StrEnum):
return "SBOM generation and augmentation"
case TaskType.SBOM_GENERATE_CYCLONEDX:
return "SBOM generation"
+ case TaskType.SBOM_HEATMAP:
+ return "SBOM heatmap analysis"
case TaskType.SBOM_OSV_SCAN:
return "SBOM vulnerability scan"
case TaskType.SBOM_QS_SCORE:
diff --git a/atr/sbom/heatmap.py b/atr/sbom/heatmap.py
index c07962cd..2b9e8934 100644
--- a/atr/sbom/heatmap.py
+++ b/atr/sbom/heatmap.py
@@ -148,6 +148,13 @@ def _health(row: results.SBOMHeatmapRow, now: datetime.datetime) -> tuple[float
return (sum(observed) / 4 if (len(observed) == 4) else None), len(observed)
+def _issues(data: Any) -> dict[str, Any]:
+ if (not isinstance(data, dict)) or (not data):
+ return {}
+ maintainers = data.get("active_maintainers")
+ return {"active_maintainers": len(maintainers) if isinstance(maintainers, list) else None}
+
+
async def _mirrors(
session: aiohttp.ClientSession, packages: dict[str, dict[str, Any]]
) -> dict[str, tuple[dict[str, Any], dict[str, Any]]]:
@@ -155,12 +162,11 @@ async def _mirrors(
mirrors = {}
for project_id in sorted(project_id for project_id in ids if project_id):
name = urllib.parse.quote(project_id.removeprefix("github.com/"), safe="")
- repository = await _request(session, f"https://repos.ecosyste.ms/api/v1/hosts/GitHub/repositories/{name}")
- issues = await _request(session, f"https://issues.ecosyste.ms/api/v1/hosts/GitHub/repositories/{name}")
- mirrors[project_id] = (
- repository if isinstance(repository, dict) else {},
- issues if isinstance(issues, dict) else {},
+ repository = _repository(
+ await _request(session, f"https://repos.ecosyste.ms/api/v1/hosts/GitHub/repositories/{name}")
)
+ issues = _issues(await _request(session, f"https://issues.ecosyste.ms/api/v1/hosts/GitHub/repositories/{name}"))
+ mirrors[project_id] = (repository, issues)
return mirrors
@@ -178,7 +184,7 @@ def _observations(
) -> None:
version = _advisories(row, versions)
row.latest_release_at = package.get("latest_release_published_at")
- row.rankings_average = _number((package.get("rankings") or {}).get("average"))
+ row.rankings_average = package.get("rankings_average")
row.repository_url = package.get("repository_url")
repository = package.get("repo_metadata") or {}
issues = package.get("issue_metadata") or {}
@@ -219,15 +225,14 @@ async def _packages(session: aiohttp.ClientSession, keys: list[str]) -> dict[str
if purl is not None:
key = str(packageurl.PackageURL(purl.type, purl.namespace, purl.name))
packages[key] = {
- field: package.get(field)
- for field in (
- "latest_release_published_at",
- "rankings",
- "repository_url",
- "repo_metadata",
- "issue_metadata",
- )
+ "latest_release_published_at": package.get("latest_release_published_at"),
+ "rankings_average": _number((package.get("rankings") or {}).get("average")),
+ "repository_url": package.get("repository_url"),
+ "repo_metadata": _repository(package.get("repo_metadata")),
+ "issue_metadata": _issues(package.get("issue_metadata")),
}
+ del package
+ del records
return packages
@@ -261,6 +266,22 @@ def _purl(text: Any) -> packageurl.PackageURL | None:
return None
+def _repository(data: Any) -> dict[str, Any]:
+ if (not isinstance(data, dict)) or (not data):
+ return {}
+ archived = data.get("archived")
+ files = (data.get("metadata") or {}).get("files")
+ names = ("security", "code_of_conduct", "contributing")
+ governance = None
+ if isinstance(files, dict) and all(name in files for name in names):
+ governance = sum(bool(files[name]) for name in names)
+ return {
+ "archived": archived if isinstance(archived, bool) else None,
+ "dds": _number((data.get("commit_stats") or {}).get("dds")),
+ "governance_files": governance,
+ }
+
+
def _repository_id(url: str) -> str | None:
parsed = urllib.parse.urlsplit(url)
if (parsed.scheme not in {"http", "https"}) or (
@@ -274,15 +295,10 @@ def _repository_id(url: str) -> str | None:
def _repository_observations(row: results.SBOMHeatmapRow, repository: dict[str, Any], issues: dict[str, Any]) -> None:
- archived = repository.get("archived")
- row.archived = archived if isinstance(archived, bool) else None
- row.dds = _number((repository.get("commit_stats") or {}).get("dds"))
- maintainers = issues.get("active_maintainers")
- row.active_maintainers = len(maintainers) if isinstance(maintainers, list) else None
- files = (repository.get("metadata") or {}).get("files")
- names = ("security", "code_of_conduct", "contributing")
- if isinstance(files, dict) and all(name in files for name in names):
- row.governance_files = sum(bool(files[name]) for name in names)
+ row.archived = repository.get("archived")
+ row.dds = repository.get("dds")
+ row.active_maintainers = issues.get("active_maintainers")
+ row.governance_files = repository.get("governance_files")
async def _request(session: aiohttp.ClientSession, url: str, body: dict[str, Any] | None = None) -> Any:
diff --git a/atr/shared/catalog.py b/atr/shared/catalog.py
index f5298e8a..450a5aae 100644
--- a/atr/shared/catalog.py
+++ b/atr/shared/catalog.py
@@ -154,6 +154,10 @@ def cycle_groups(
]
+def sbom_urls(version: models.api.CatalogVersion) -> dict[str, str]:
+ return {artifact.artifact_path: artifact.sbom_url for artifact in version.artifacts if artifact.sbom_url}
+
+
def version_order(
version_method: sql.VersionMethod, items: Iterable[tuple[str, datetime.datetime | None, int | None]]
) -> list[str]:
diff --git a/atr/tasks/__init__.py b/atr/tasks/__init__.py
index 3d0557da..045057c1 100644
--- a/atr/tasks/__init__.py
+++ b/atr/tasks/__init__.py
@@ -51,6 +51,7 @@ import atr.tasks.checks.targz as targz
import atr.tasks.checks.zipformat as zipformat
import atr.tasks.distribution as distribution
import atr.tasks.gha as gha
+import atr.tasks.heatmap as heatmap
import atr.tasks.keys as keys
import atr.tasks.maintenance as maintenance
import atr.tasks.message as message
@@ -432,6 +433,8 @@ def resolve(task_type: sql.TaskType) -> Callable[..., Awaitable[results.Results
return sbom.generate
case sql.TaskType.SBOM_GENERATE_CYCLONEDX:
return sbom.generate_cyclonedx
+ case sql.TaskType.SBOM_HEATMAP:
+ return heatmap.analyse
case sql.TaskType.SBOM_OSV_SCAN:
return sbom.osv_scan
case sql.TaskType.SBOM_QS_SCORE:
diff --git a/atr/tasks/heatmap.py b/atr/tasks/heatmap.py
new file mode 100644
index 00000000..c7043a91
--- /dev/null
+++ b/atr/tasks/heatmap.py
@@ -0,0 +1,85 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+import datetime
+from typing import Any
+
+import sqlmodel
+
+import atr.catalog_site as catalog_site
+import atr.constants as constants
+import atr.db as db
+import atr.models.args as args
+import atr.models.results as results
+import atr.models.safe as safe
+import atr.models.sql as sql
+import atr.sbom.heatmap
+import atr.shared.catalog as catalog
+import atr.tasks.checks as checks
+
+
+@checks.with_model(args.SBOMHeatmap)
+async def analyse(task_args: args.SBOMHeatmap) -> results.SBOMHeatmap:
+ async with db.session() as data:
+ release = await data.release(
+ project_key=str(task_args.project_key),
+ version=str(task_args.version_key),
+ phase=sql.ReleasePhase.RELEASE,
+ ).demand(ValueError("Heatmap analysis requires a published release"))
+ artifacts = await data.artifact(project_key=release.project_key, version=release.version, _release=True).all()
+ versions = catalog.assemble(
+ release.project.version_method, artifacts, [], datetime.datetime.now(datetime.UTC)
+ ).versions
+ sboms = {path: url for version in versions for path, url in catalog.sbom_urls(version).items()}
+ if not sboms:
+ raise ValueError("Release has no published SBOMs")
+ return await atr.sbom.heatmap.analyse(sboms)
+
+
+async def complete(data: db.Session, task_id: int, task_args: dict[str, Any]) -> None:
+ identity = args.SBOMHeatmap.model_validate(task_args)
+ via = sql.validate_instrumented_attribute
+ await data.execute(
+ sqlmodel.delete(sql.Task).where(
+ via(sql.Task.task_type) == sql.TaskType.SBOM_HEATMAP,
+ via(sql.Task.project_key) == str(identity.project_key),
+ via(sql.Task.version_key) == str(identity.version_key),
+ via(sql.Task.status) == sql.TaskStatus.COMPLETED,
+ via(sql.Task.id) != task_id,
+ )
+ )
+ await catalog_site.queue_regeneration(data, constants.SYSTEM_SERVICE_UID, str(identity.project_key))
+
+
+async def queue(data: db.Session, project_key: safe.ProjectKey, version_key: safe.VersionKey) -> sql.Task:
+ existing = await data.task(
+ task_type=sql.TaskType.SBOM_HEATMAP,
+ project_key=str(project_key),
+ version_key=str(version_key),
+ status_in=[sql.TaskStatus.QUEUED, sql.TaskStatus.ACTIVE],
+ ).get()
+ if existing is not None:
+ return existing
+ task = sql.Task(
+ task_type=sql.TaskType.SBOM_HEATMAP,
+ task_args=args.SBOMHeatmap(project_key=project_key, version_key=version_key).model_dump(),
+ asf_uid=constants.SYSTEM_SERVICE_UID,
+ project_key=str(project_key),
+ version_key=str(version_key),
+ )
+ data.add(task)
+ return task
diff --git a/atr/tasks/task.py b/atr/tasks/task.py
index 5c66ed2f..092521ed 100644
--- a/atr/tasks/task.py
+++ b/atr/tasks/task.py
@@ -70,6 +70,7 @@ RECURRING_TASK_TYPES: Final[frozenset[sql.TaskType]] = frozenset(
TASK_TYPE_TIMEOUT_SECONDS: Final[dict[sql.TaskType, int]] = {
sql.TaskType.CATALOG_SITE_GENERATE: 1800,
sql.TaskType.RELEASE_FINALISE: 600,
+ sql.TaskType.SBOM_HEATMAP: 600,
}
diff --git a/atr/worker.py b/atr/worker.py
index 117f8761..3011fe54 100644
--- a/atr/worker.py
+++ b/atr/worker.py
@@ -44,6 +44,7 @@ import atr.models.safe as safe
import atr.models.sql as sql
import atr.tasks as tasks
import atr.tasks.checks as checks
+import atr.tasks.heatmap as heatmap
import atr.tasks.task as task
import atr.util as util
@@ -276,11 +277,18 @@ async def _task_next_claim() -> tuple[int, str, list[str] | dict[str, Any], str]
async with db.session() as data:
async with data.begin():
# Get the ID of the oldest queued task
+ heatmap_active = (
+ sqlmodel.select(sql.Task.id)
+ .where(sql.Task.task_type == sql.TaskType.SBOM_HEATMAP, sql.Task.status == task.ACTIVE)
+ .correlate(None)
+ .exists()
+ )
oldest_queued_task = (
sqlmodel.select(sql.Task.id)
.where(
sqlmodel.and_(
sql.Task.status == task.QUEUED,
+ sqlmodel.or_(sql.Task.task_type != sql.TaskType.SBOM_HEATMAP, ~heatmap_active),
sqlmodel.or_(
via(sql.Task.scheduled).is_(None),
via(sql.Task.scheduled) <= datetime.datetime.now(datetime.UTC),
@@ -428,6 +436,8 @@ async def _task_result_process(
task_type, task_args, asf_uid, added, started = row
with contextlib.suppress(ValueError):
task_type = sql.TaskType(task_type)
+ if task_type == sql.TaskType.SBOM_HEATMAP:
+ await heatmap.complete(data, task_id, task_args)
if task_type in task.RECURRING_TASK_TYPES:
# A successful recurring run leaves only a log line behind
task_obj = sql.Task(
diff --git a/tests/unit/test_heatmap_tasks.py b/tests/unit/test_heatmap_tasks.py
new file mode 100644
index 00000000..e9321842
--- /dev/null
+++ b/tests/unit/test_heatmap_tasks.py
@@ -0,0 +1,283 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+import asyncio
+import datetime
+import os
+import unittest.mock as mock
+
+import pytest
+import sqlalchemy.event
+import sqlalchemy.ext.asyncio
+import sqlmodel
+
+import atr.catalog_site as catalog_site
+import atr.constants as constants
+import atr.db as db
+import atr.models.results as results
+import atr.models.safe as safe
+import atr.models.sql as sql
+import atr.paths as paths
+import atr.sbom.heatmap
+import atr.tasks as tasks
+import atr.tasks.heatmap as heatmap
+import atr.worker as worker
+
+_PROJECT = safe.ProjectKey("example")
+_VERSION = safe.VersionKey("1.0")
+_SBOM_URL = "https://downloads.apache.org/example/1.0/source.zip.cdx.json"
+_NOW = datetime.datetime(2026, 9, 9, tzinfo=datetime.UTC)
+
+
+@pytest.fixture
+async def database(monkeypatch, tmp_path):
+ engine = sqlalchemy.ext.asyncio.create_async_engine("sqlite+aiosqlite:///" + str(tmp_path / "tasks.sqlite"))
+ sqlalchemy.event.listen(engine.sync_engine, "connect", _foreign_keys)
+ async with engine.begin() as connection:
+ await connection.execute(sqlalchemy.text("PRAGMA journal_mode=WAL"))
+ await connection.run_sync(sqlmodel.SQLModel.metadata.create_all)
+ maker = sqlalchemy.ext.asyncio.async_sessionmaker(bind=engine, class_=db.Session, expire_on_commit=False)
+ monkeypatch.setattr(db, "session", maker)
+ monkeypatch.setattr(paths, "get_catalog_site_dir", lambda: safe.StatePath(tmp_path))
+ async with maker() as data:
+ data.add(sql.Committee(key="example", name="Example", catalog_reviewed=True))
+ await data.flush()
+ data.add(sql.Project(key="example", name="Apache Example", committee_key="example"))
+ await data.flush()
+ data.add(sql.Release(project_key="example", version="1.0", phase=sql.ReleasePhase.RELEASE, created=_NOW))
+ await data.flush()
+ data.add(
+ sql.Artifact(
+ project_key="example",
+ version="1.0",
+ release_key="example-1.0",
+ artifact_path="source.zip",
+ sbom_path="source.zip.cdx.json",
+ download_path_suffix="example/1.0",
+ )
+ )
+ await data.commit()
+ yield maker
+ await engine.dispose()
+
+
+@pytest.mark.parametrize("change", ["added", "removed", "replaced", "archived", "withdrawn", "analysis_version"])
+async def test_catalogue_removes_inapplicable_reports(database, tmp_path, change) -> None:
+ async with database() as data:
+ await _saved(data)
+ await catalog_site.regenerate_project(data, "example")
+ assert (tmp_path / "example/1.0/heatmap.json").is_file()
+ artifact = await data.artifact(project_key="example", version="1.0").get()
+ release = await data.release(project_key="example", version="1.0").get()
+ match change:
+ case "added":
+ data.add(
+ sql.Artifact(
+ project_key="example",
+ version="1.0",
+ release_key=release.key,
+ artifact_path="binary.zip",
+ sbom_path="binary.zip.cdx.json",
+ download_path_suffix="example/1.0",
+ )
+ )
+ case "removed":
+ artifact.sbom_path = None
+ case "replaced":
+ artifact.sbom_path = "replacement.cdx.json"
+ case "archived":
+ release.is_archived = True
+ case "withdrawn":
+ release.phase = sql.ReleasePhase.RELEASE_CANDIDATE_DRAFT
+ case "analysis_version":
+ saved = await data.task(task_type=sql.TaskType.SBOM_HEATMAP).get()
+ saved.result = saved.result.model_copy(update={"analysis_version": 0})
+ await data.commit()
+ await catalog_site.regenerate_project(data, "example")
+ assert not (tmp_path / "example/1.0/heatmap.json").exists()
+ assert not (tmp_path / "example/1.0/heatmap.html").exists()
+ assert "heatmap.html" not in (tmp_path / "example/1.0/index.html").read_text()
+
+
+async def test_completion_and_publication_are_atomic(database, monkeypatch) -> None:
+ async with database() as data:
+ previous = await _saved(data)
+ current = await heatmap.queue(data, _PROJECT, _VERSION)
+ await data.commit()
+ current_id = current.id
+ claimed = await worker._task_next_claim()
+ assert claimed[0] == current_id
+ with monkeypatch.context() as patch:
+ patch.setattr(catalog_site, "queue_regeneration", mock.AsyncMock(side_effect=RuntimeError("unavailable")))
+ with pytest.raises(RuntimeError, match="unavailable"):
+ await worker._task_result_process(current_id, _snapshot(), sql.TaskStatus.COMPLETED)
+ async with database() as data:
+ assert (await data.task(id=current_id).get()).status == sql.TaskStatus.ACTIVE
+ assert await data.task(id=previous.id).get() is not None
+ assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is None
+ await worker._task_result_process(current_id, _snapshot(), sql.TaskStatus.COMPLETED)
+ async with database() as data:
+ assert await data.task(id=previous.id).get() is None
+ assert (await data.task(id=current_id).get()).result == _snapshot()
+ assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is not None
+
+
+@pytest.mark.parametrize("status", [sql.TaskStatus.FAILED, sql.TaskStatus.BROKEN])
+async def test_failed_refresh_and_late_completion_preserve_the_previous_result(database, status) -> None:
+ async with database() as data:
+ previous = await _saved(data)
+ current = await heatmap.queue(data, _PROJECT, _VERSION)
+ await data.commit()
+ current_id = current.id
+ await worker._task_next_claim()
+ await worker._task_result_process(current_id, None, status, "failed refresh")
+ await worker._task_result_process(current_id, _snapshot(), sql.TaskStatus.COMPLETED)
+ async with database() as data:
+ assert (await data.task(id=current_id).get()).status == status
+ assert (await data.task(id=previous.id).get()).result == _snapshot()
+ assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is None
+
+
+async def test_heatmap_waiters_do_not_block_other_tasks(database) -> None:
+ async with database() as data:
+ older = await heatmap.queue(data, _PROJECT, _VERSION)
+ older.scheduled = datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=1)
+ newer = await heatmap.queue(data, _PROJECT, safe.VersionKey("2.0"))
+ await data.commit()
+ older_id, newer_id = older.id, newer.id
+ assert (await worker._task_next_claim())[0] == newer_id
+ async with database() as data:
+ older = await data.task(id=older_id).get()
+ older.scheduled = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=1)
+ ordinary = sql.Task(task_type=sql.TaskType.SBOM_QS_SCORE, task_args={}, asf_uid=constants.SYSTEM_SERVICE_UID)
+ data.add(ordinary)
+ await data.commit()
+ ordinary_id = ordinary.id
+ assert (await worker._task_next_claim())[0] == ordinary_id
+ assert await worker._task_next_claim() is None
+ async with database() as data:
+ assert (await data.task(id=older_id).get()).status == sql.TaskStatus.QUEUED
+
+
+@pytest.mark.parametrize("status", [sql.TaskStatus.COMPLETED, sql.TaskStatus.FAILED])
+async def test_heatmaps_are_claimed_exclusively_across_projects(database, status) -> None:
+ async with database() as data:
+ data.add(sql.Project(key="other", name="Other", committee_key="example"))
+ await data.flush()
+ first = await heatmap.queue(data, _PROJECT, _VERSION)
+ second = await heatmap.queue(data, safe.ProjectKey("other"), _VERSION)
+ await data.commit()
+ ids = {first.id, second.id}
+ claims = await asyncio.gather(worker._task_next_claim(), worker._task_next_claim())
+ assert claims.count(None) == 1
+ claimed = next(claim for claim in claims if claim is not None)
+ assert claimed[0] in ids
+ assert await worker._task_next_claim() is None
+ result = _snapshot() if (status == sql.TaskStatus.COMPLETED) else None
+ await worker._task_result_process(claimed[0], result, status, "failed" if result is None else None)
+ assert (await worker._task_next_claim())[0] == (ids - {claimed[0]}).pop()
+
+
+@pytest.mark.parametrize("archived", [False, True])
+async def test_queued_analysis_is_stored_and_published(database, monkeypatch, tmp_path, archived) -> None:
+ sbom_url = _SBOM_URL.replace("downloads.apache.org", "archive.apache.org/dist") if archived else _SBOM_URL
+ snapshot = _snapshot(sbom_url)
+ analyse = mock.AsyncMock(return_value=snapshot)
+ monkeypatch.setattr(atr.sbom.heatmap, "analyse", analyse)
+ async with database() as data:
+ release = await data.release(project_key="example", version="1.0").get()
+ release.is_archived = archived
+ queued = await heatmap.queue(data, _PROJECT, _VERSION)
+ assert await heatmap.queue(data, _PROJECT, _VERSION) is queued
+ assert queued.asf_uid == constants.SYSTEM_SERVICE_UID
+ assert queued.inputs_hash is None
+ await data.commit()
+ task_id = queued.id
+ claimed = await worker._task_next_claim()
+ assert claimed[0] == task_id
+ async with database() as data:
+ assert (await heatmap.queue(data, _PROJECT, _VERSION)).id == task_id
+ await worker._task_process(*claimed)
+ analyse.assert_awaited_once_with({"source.zip": sbom_url})
+ async with database() as data:
+ saved = await data.task(id=task_id).get()
+ assert saved.status == sql.TaskStatus.COMPLETED
+ assert saved.result == snapshot
+ regeneration = await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get()
+ assert regeneration.project_key == "example"
+ regeneration_args = regeneration.task_args
+ await tasks.resolve(sql.TaskType.CATALOG_SITE_GENERATE)(regeneration_args)
+ assert results.SBOMHeatmap.model_validate_json((tmp_path / "example/1.0/heatmap.json").read_text()) == snapshot
+ assert "heatmap-table.js" in (tmp_path / "example/1.0/heatmap.html").read_text()
+ assert "heatmap.html" in (tmp_path / "example/1.0/index.html").read_text()
+ assert (tmp_path / "assets/js/src/heatmap-table.js").is_file()
+ assert (tmp_path / "assets/css/heatmap.css").is_file()
+ assert not list(tmp_path.rglob("*.tmp"))
+
+
+@pytest.mark.parametrize("published", [False, True])
+async def test_unpublished_release_or_missing_sbom_is_not_analysed(database, monkeypatch, published) -> None:
+ analyse = mock.AsyncMock()
+ monkeypatch.setattr(atr.sbom.heatmap, "analyse", analyse)
+ async with database() as data:
+ if published:
+ artifact = await data.artifact(project_key="example", version="1.0").get()
+ artifact.sbom_path = None
+ else:
+ release = await data.release(project_key="example", version="1.0").get()
+ release.phase = sql.ReleasePhase.RELEASE_CANDIDATE_DRAFT
+ queued = await heatmap.queue(data, _PROJECT, _VERSION)
+ await data.commit()
+ task_args = queued.task_args
+ with pytest.raises(ValueError, match="published"):
+ await tasks.resolve(sql.TaskType.SBOM_HEATMAP)(task_args)
+ analyse.assert_not_awaited()
+
+
+def _foreign_keys(connection, _record) -> None:
+ cursor = connection.cursor()
+ cursor.execute("PRAGMA foreign_keys=ON")
+ cursor.close()
+
+
+async def _saved(data):
+ saved = sql.Task(
+ task_type=sql.TaskType.SBOM_HEATMAP,
+ task_args={"project_key": "example", "version_key": "1.0"},
+ asf_uid=constants.SYSTEM_SERVICE_UID,
+ project_key="example",
+ version_key="1.0",
+ status=sql.TaskStatus.COMPLETED,
+ completed=_NOW,
+ started=_NOW,
+ pid=os.getpid(),
+ result=_snapshot(),
+ )
+ data.add(saved)
+ await data.commit()
+ return saved
+
+
+def _snapshot(sbom_url=_SBOM_URL):
+ return results.SBOMHeatmap(
+ analysis_version=atr.sbom.heatmap.ANALYSIS_VERSION,
+ generated_at=_NOW.isoformat(),
+ sboms=[results.SBOMHeatmapSbom(artifact_path="source.zip", sbom_url=sbom_url, sha256="a" * 64)],
+ rows=[],
+ sources={},
+ attribution="Example",
+ )
diff --git a/tests/unit/test_sbom_heatmap.py b/tests/unit/test_sbom_heatmap.py
index ece3a4e9..55588e50 100644
--- a/tests/unit/test_sbom_heatmap.py
+++ b/tests/unit/test_sbom_heatmap.py
@@ -184,15 +184,22 @@ async def test_analyse_keeps_versions_membership_and_unknown_packages(monkeypatc
assert results.ResultsAdapter.validate_json(result.model_dump_json()) == result
+@pytest.mark.parametrize("metadata", [None, {}, {"unused": "present"}])
async def test_analyse_labels_gitbox_inference_and_deduplicates_repository_calls(
monkeypatch: pytest.MonkeyPatch,
+ metadata,
) -> None:
source = document({"purl": _KEY + "@1"}, {"purl": "pkg:maven/org.example/other@2"})
monkeypatch.setattr(heatmap.utilities, "fetch", mock.AsyncMock(return_value=source))
mirror_url = "https://gitbox.apache.org/repos/asf/example.git"
packages = [
- package(repository_url=mirror_url, repo_metadata=None, issue_metadata=None),
- package(purl="pkg:maven/org.example/other", repository_url=mirror_url, repo_metadata=None, issue_metadata=None),
+ package(repository_url=mirror_url, repo_metadata=metadata, issue_metadata=metadata),
+ package(
+ purl="pkg:maven/org.example/other",
+ repository_url=mirror_url,
+ repo_metadata=metadata,
+ issue_metadata=metadata,
+ ),
]
fake = session(
monkeypatch,
@@ -232,9 +239,9 @@ async def test_analyse_labels_gitbox_inference_and_deduplicates_repository_calls
assert entry.source_repo == _PROJECT
assert entry.repository_source == ("deps.dev" if (entry.key == _KEY) else "gitbox_mirror")
assert entry.repository_url == mirror_url
- assert entry.active_maintainers == 0
- assert entry.health_inputs == 4
- assert entry.health is not None
+ assert entry.active_maintainers == (None if metadata else 0)
+ assert entry.health_inputs == (1 if metadata else 4)
+ assert (entry.health is None) == bool(metadata)
assert entry.maintained == 10
assert entry.scorecard_date == "2026-08-24"
@@ -305,13 +312,35 @@ def test_missing_governance_is_not_zero_and_issue_tracker_is_not_source() -> Non
data = {"relatedProjects": [{"relationType": "ISSUE_TRACKER", "projectKey": {"id": _PROJECT}}]}
heatmap._observations(entry, {}, {entry.purl: data}, {})
assert entry.source_repo is None
- heatmap._repository_observations(entry, {"metadata": {"files": {}}}, {})
+ heatmap._repository_observations(entry, heatmap._repository({"metadata": {"files": {}}}), {})
assert entry.governance_files is None
- heatmap._repository_observations(entry, repository(), {"active_maintainers": []})
+ heatmap._repository_observations(
+ entry, heatmap._repository(repository()), heatmap._issues({"active_maintainers": []})
+ )
assert entry.governance_files == 1
assert entry.active_maintainers == 0
+async def test_package_metadata_retains_only_scoring_values(monkeypatch) -> None:
+ records = [
+ package(
+ rankings={"average": 12, "unused": ["bulk"] * 1000},
+ repo_metadata=repository(unused=["bulk"] * 1000),
+ issue_metadata={"active_maintainers": [{"unused": ["bulk"] * 1000}], "unused": ["bulk"] * 1000},
+ )
+ ]
+ fake = session(monkeypatch, packages=records)
+ compact = await heatmap._packages(fake, [_KEY])
+ assert compact[_KEY] == {
+ "latest_release_published_at": "2026-08-01T00:00:00Z",
+ "rankings_average": 12,
+ "repository_url": f"https://{_PROJECT}",
+ "repo_metadata": {"archived": False, "dds": 0.5, "governance_files": 1},
+ "issue_metadata": {"active_maintainers": 1},
+ }
+ assert records[0]["issue_metadata"]["active_maintainers"][0]["unused"] == ["bulk"] * 1000
+
+
@pytest.mark.parametrize(
("changes", "health", "inputs", "risk"),
[
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org