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/10 16:27:39 UTC
(tooling-trusted-releases) branch sbp updated (020c5678 -> 1609a433)
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a change to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
from 020c5678 Make SBOM file discovery more lenient
add ffca3bf6 Don't show podlings on attic card, fix CPU time limit for long tasks
new 1609a433 Schedule the generation of SBOM heatmap tables
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
atr/catalog_site.py | 22 ++--
atr/get/release.py | 4 +-
atr/tasks/heatmap.py | 130 +++++++++++++++++++-
atr/tasks/maintenance.py | 2 +
atr/worker.py | 30 +++--
tests/unit/test_heatmap_tasks.py | 252 ++++++++++++++++++++++++++++++++++++++-
6 files changed, 418 insertions(+), 22 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org
(tooling-trusted-releases) 01/01: Schedule the generation of SBOM heatmap tables
Posted by sb...@apache.org.
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
commit 1609a43396a9f2df94e7962cc97fd484600b107a
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Thu Sep 10 17:27:05 2026 +0100
Schedule the generation of SBOM heatmap tables
---
atr/catalog_site.py | 22 ++--
atr/tasks/heatmap.py | 130 +++++++++++++++++++-
atr/tasks/maintenance.py | 2 +
atr/worker.py | 21 +++-
tests/unit/test_heatmap_tasks.py | 252 ++++++++++++++++++++++++++++++++++++++-
5 files changed, 410 insertions(+), 17 deletions(-)
diff --git a/atr/catalog_site.py b/atr/catalog_site.py
index 13835d55..0dc23393 100644
--- a/atr/catalog_site.py
+++ b/atr/catalog_site.py
@@ -184,6 +184,15 @@ class _ReleaseSummary:
date: datetime.datetime | None
+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
+
+
async def generate_all(data: db.Session) -> None:
"""Rebuild every page, overwriting in place so the site stays servable throughout."""
site_dir = paths.get_catalog_site_dir()
@@ -285,17 +294,6 @@ async def regenerate_project(data: db.Session, project_key: str) -> None:
await _guarded_write("release summary feeds", _write_project_releases(data, committees, site_dir))
-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]]:
@@ -948,7 +946,7 @@ async def _write_project(
heatmaps = await _heatmaps(data, project.key, releases)
for version in assembled.versions:
document = release_documents.get(str(version.version))
- snapshot = _applicable_heatmap(heatmaps.get(str(version.version)), version)
+ 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/tasks/heatmap.py b/atr/tasks/heatmap.py
index c7043a91..1b9ccc57 100644
--- a/atr/tasks/heatmap.py
+++ b/atr/tasks/heatmap.py
@@ -15,22 +15,28 @@
# specific language governing permissions and limitations
# under the License.
+import asyncio
import datetime
-from typing import Any
+from typing import Any, Final
import sqlmodel
import atr.catalog_site as catalog_site
import atr.constants as constants
import atr.db as db
+import atr.log as log
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.paths as paths
import atr.sbom.heatmap
import atr.shared.catalog as catalog
import atr.tasks.checks as checks
+_REFRESH_INTERVAL: Final = datetime.timedelta(days=7)
+_RETRY_INTERVAL: Final = datetime.timedelta(days=1)
+
@checks.with_model(args.SBOMHeatmap)
async def analyse(task_args: args.SBOMHeatmap) -> results.SBOMHeatmap:
@@ -83,3 +89,125 @@ async def queue(data: db.Session, project_key: safe.ProjectKey, version_key: saf
)
data.add(task)
return task
+
+
+async def sweep() -> None:
+ now = datetime.datetime.now(datetime.UTC)
+ via = sql.validate_instrumented_attribute
+ with_sboms = (
+ sqlmodel.select(sql.Release.key)
+ .join(
+ sql.Artifact,
+ sqlmodel.and_(
+ sql.Artifact.project_key == sql.Release.project_key, sql.Artifact.version == sql.Release.version
+ ),
+ )
+ .where(sql.Release.phase == sql.ReleasePhase.RELEASE, via(sql.Artifact.sbom_path).is_not(None))
+ .where(via(sql.Release.is_archived).is_(False), sql.Artifact.sbom_path != "")
+ )
+ with_reports = (
+ sqlmodel.select(sql.Release.key)
+ .join(
+ sql.Task,
+ sqlmodel.and_(sql.Task.project_key == sql.Release.project_key, sql.Task.version_key == sql.Release.version),
+ )
+ .where(sql.Task.task_type == sql.TaskType.SBOM_HEATMAP, sql.Task.status == sql.TaskStatus.COMPLETED)
+ )
+ async with db.session() as data:
+ release_keys = (await data.execute(with_sboms.union(with_reports))).scalars().all()
+ projects = set()
+ for release_key in release_keys:
+ try:
+ project_key = await _sweep_release(release_key, now)
+ if project_key is not None:
+ projects.add(project_key)
+ except Exception:
+ log.exception(f"Heatmap sweep failed for {release_key}")
+ async with db.session() as data:
+ await data.begin_immediate()
+ for project_key in sorted(projects):
+ await catalog_site.queue_regeneration(data, constants.SYSTEM_SERVICE_UID, project_key)
+ await data.commit()
+ log.info(f"Heatmap sweep checked {len(release_keys)} releases and requested {len(projects)} project regenerations")
+
+
+def _analysis_due(
+ latest: sql.Task | None, completed: sql.Task | None, applicable: bool, now: datetime.datetime
+) -> bool:
+ if latest is not None:
+ if latest.status in {sql.TaskStatus.QUEUED, sql.TaskStatus.ACTIVE}:
+ return False
+ if (
+ (latest.status in {sql.TaskStatus.FAILED, sql.TaskStatus.BROKEN})
+ and (latest.completed is not None)
+ and (latest.completed > (now - _RETRY_INTERVAL))
+ ):
+ return False
+ return (
+ (not applicable)
+ or (completed is None)
+ or (completed.completed is None)
+ or (completed.completed <= (now - _REFRESH_INTERVAL))
+ )
+
+
+def _publication_needed(project_key: str, version_key: str, completed: datetime.datetime | None) -> bool:
+ directory = paths.get_catalog_site_dir() / project_key / version_key
+ if completed is None:
+ return any((directory / name).path.exists() for name in ("heatmap.json", "heatmap.html"))
+ for name in ("heatmap.json", "heatmap.html", "index.html"):
+ try:
+ if (directory / name).path.stat().st_mtime < completed.timestamp():
+ return True
+ except FileNotFoundError:
+ return True
+ return False
+
+
+async def _sweep_release(release_key: str, now: datetime.datetime) -> str | None:
+ via = sql.validate_instrumented_attribute
+ async with db.session() as data:
+ await data.begin_immediate()
+ release = await data.release(key=release_key).get()
+ if release is None:
+ return None
+ artifacts = await data.artifact(project_key=release.project_key, version=release.version, _release=True).all()
+ versions = catalog.assemble(release.project.version_method, artifacts, [], now).versions
+ version = next(iter(versions), None)
+ history = data.task(
+ task_type=sql.TaskType.SBOM_HEATMAP, project_key=release.project_key, version_key=release.version
+ )
+ latest = await history.order_by(via(sql.Task.added).desc(), via(sql.Task.id).desc()).limit(1).get()
+ completed = latest
+ if (completed is not None) and (completed.status != sql.TaskStatus.COMPLETED):
+ completed = (
+ await data.task(
+ task_type=sql.TaskType.SBOM_HEATMAP,
+ project_key=release.project_key,
+ version_key=release.version,
+ status=sql.TaskStatus.COMPLETED,
+ )
+ .order_by(via(sql.Task.completed).desc(), via(sql.Task.id).desc())
+ .limit(1)
+ .get()
+ )
+ snapshot = completed.result if completed and isinstance(completed.result, results.SBOMHeatmap) else None
+ applicable = (
+ (release.phase == sql.ReleasePhase.RELEASE)
+ and (version is not None)
+ and (catalog_site.applicable_heatmap(snapshot, version) is not None)
+ )
+ if (
+ (release.phase == sql.ReleasePhase.RELEASE)
+ and (not release.is_archived)
+ and (version is not None)
+ and catalog.sbom_urls(version)
+ and _analysis_due(latest, completed, applicable, now)
+ ):
+ await queue(data, release.safe_project_key, release.safe_version_key)
+ project_key, version_key = release.project_key, release.version
+ published_at = completed.completed if applicable and completed else None
+ await data.commit()
+ if await asyncio.to_thread(_publication_needed, project_key, version_key, published_at):
+ return project_key
+ return None
diff --git a/atr/tasks/maintenance.py b/atr/tasks/maintenance.py
index 2be8d10c..5bfcbcf8 100644
--- a/atr/tasks/maintenance.py
+++ b/atr/tasks/maintenance.py
@@ -28,6 +28,7 @@ import atr.models.results as results
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.inactivity as inactivity
_EXPIRED_TOKEN_RETENTION_DAYS: Final[int] = 30
@@ -53,6 +54,7 @@ async def run(task_args: args.MaintenanceArgs) -> results.Results | None:
await _storage_maintenance()
await _workflow_auth_maintenance()
await _inactivity_maintenance()
+ await heatmap.sweep()
log.info(
"Storage maintenance completed successfully",
diff --git a/atr/worker.py b/atr/worker.py
index 65821ac5..d6dcae95 100644
--- a/atr/worker.py
+++ b/atr/worker.py
@@ -50,6 +50,7 @@ import atr.util as util
_CPU_LIMIT_SECONDS: Final = 300
_DEFER_SECONDS: Final = 120
+_HEATMAP_START_INTERVAL: Final = datetime.timedelta(minutes=5)
_MEMORY_WATCHDOG_POLL_SECONDS: Final = 0.5
_SHUTDOWN_GRACE_SECONDS: Final = 15.0
_TASK_LOG_LOGGER: Final = "atr.tasks.log"
@@ -277,21 +278,31 @@ 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
+ now = datetime.datetime.now(datetime.UTC)
heatmap_active = (
sqlmodel.select(sql.Task.id)
.where(sql.Task.task_type == sql.TaskType.SBOM_HEATMAP, sql.Task.status == task.ACTIVE)
.correlate(None)
.exists()
)
+ heatmap_cooldown = (
+ sqlmodel.select(sql.TextValue.key)
+ .where(sql.TextValue.ns == "heatmap", sql.TextValue.key == "next_start")
+ .where(via(sql.TextValue.value) > now.isoformat(timespec="microseconds"))
+ .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_(
+ sql.Task.task_type != sql.TaskType.SBOM_HEATMAP,
+ sqlmodel.and_(~heatmap_active, ~heatmap_cooldown),
+ ),
sqlmodel.or_(
via(sql.Task.scheduled).is_(None),
- via(sql.Task.scheduled) <= datetime.datetime.now(datetime.UTC),
+ via(sql.Task.scheduled) <= now,
),
)
)
@@ -302,7 +313,6 @@ async def _task_next_claim() -> tuple[int, str, list[str] | dict[str, Any], str]
# Use an UPDATE with a WHERE clause to atomically claim the task
# This ensures that only one worker can claim a specific task
- now = datetime.datetime.now(datetime.UTC)
created = psutil.Process().create_time()
update_stmt = (
sqlmodel.update(sql.Task)
@@ -321,6 +331,11 @@ async def _task_next_claim() -> tuple[int, str, list[str] | dict[str, Any], str]
if claimed_task:
task_id, task_type, task_args, asf_uid = claimed_task
+ if task_type == sql.TaskType.SBOM_HEATMAP:
+ next_start = datetime.datetime.now(datetime.UTC) + _HEATMAP_START_INTERVAL
+ await data.ns_text_set(
+ "heatmap", "next_start", next_start.isoformat(timespec="microseconds"), commit=False
+ )
log.info(f"Claimed task {task_id} ({task_type}) with args {_task_args_for_log(task_type, task_args)}")
return task_id, task_type, task_args, asf_uid
diff --git a/tests/unit/test_heatmap_tasks.py b/tests/unit/test_heatmap_tasks.py
index e9321842..49a52bef 100644
--- a/tests/unit/test_heatmap_tasks.py
+++ b/tests/unit/test_heatmap_tasks.py
@@ -28,6 +28,7 @@ import sqlmodel
import atr.catalog_site as catalog_site
import atr.constants as constants
import atr.db as db
+import atr.manager as manager
import atr.models.results as results
import atr.models.safe as safe
import atr.models.sql as sql
@@ -35,6 +36,7 @@ import atr.paths as paths
import atr.sbom.heatmap
import atr.tasks as tasks
import atr.tasks.heatmap as heatmap
+import atr.tasks.maintenance as maintenance
import atr.worker as worker
_PROJECT = safe.ProjectKey("example")
@@ -43,6 +45,15 @@ _SBOM_URL = "https://downloads.apache.org/example/1.0/source.zip.cdx.json"
_NOW = datetime.datetime(2026, 9, 9, tzinfo=datetime.UTC)
+@pytest.fixture(autouse=True)
+def clock(monkeypatch):
+ clock = mock.Mock(wraps=datetime)
+ clock.datetime.now.return_value = _NOW
+ monkeypatch.setattr(heatmap, "datetime", clock)
+ monkeypatch.setattr(worker, "datetime", clock)
+ return clock
+
+
@pytest.fixture
async def database(monkeypatch, tmp_path):
engine = sqlalchemy.ext.asyncio.create_async_engine("sqlite+aiosqlite:///" + str(tmp_path / "tasks.sqlite"))
@@ -152,6 +163,47 @@ async def test_failed_refresh_and_late_completion_preserve_the_previous_result(d
assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is None
+async def test_heatmap_pacing_survives_recovery_and_does_not_block_other_tasks(database, clock) -> None:
+ async with database() as data:
+ queued = await heatmap.queue(data, _PROJECT, _VERSION)
+ await data.commit()
+ task_id = queued.id
+ assert (await worker._task_next_claim())[0] == task_id
+ async with database() as data:
+ active = await data.task(id=task_id).get()
+ assert await manager._requeue_claim(active) == 1
+ await tasks.clear_scheduled()
+ await heatmap.sweep()
+ async with database() as data:
+ 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 len(await data.task(task_type=sql.TaskType.SBOM_HEATMAP).all()) == 1
+ clock.datetime.now.return_value = _NOW + datetime.timedelta(seconds=30)
+ assert (await worker._task_next_claim())[0] == ordinary_id
+ async with database() as data:
+ assert await data.ns_text_get("heatmap", "next_start") == "2026-09-09T00:05:00.000000+00:00"
+ clock.datetime.now.return_value = _NOW + datetime.timedelta(seconds=299)
+ assert await worker._task_next_claim() is None
+ clock.datetime.now.return_value = _NOW + datetime.timedelta(minutes=5)
+ assert (await worker._task_next_claim())[0] == task_id
+
+
+async def test_heatmap_pacing_update_is_atomic_with_claim(database, monkeypatch) -> None:
+ async with database() as data:
+ queued = await heatmap.queue(data, _PROJECT, _VERSION)
+ await data.commit()
+ task_id = queued.id
+ with monkeypatch.context() as patch:
+ patch.setattr(db.Session, "ns_text_set", mock.AsyncMock(side_effect=RuntimeError("unavailable")))
+ with pytest.raises(RuntimeError, match="unavailable"):
+ await worker._task_next_claim()
+ async with database() as data:
+ assert (await data.task(id=task_id).get()).status == sql.TaskStatus.QUEUED
+ assert (await worker._task_next_claim())[0] == task_id
+
+
async def test_heatmap_waiters_do_not_block_other_tasks(database) -> None:
async with database() as data:
older = await heatmap.queue(data, _PROJECT, _VERSION)
@@ -174,7 +226,7 @@ async def test_heatmap_waiters_do_not_block_other_tasks(database) -> None:
@pytest.mark.parametrize("status", [sql.TaskStatus.COMPLETED, sql.TaskStatus.FAILED])
-async def test_heatmaps_are_claimed_exclusively_across_projects(database, status) -> None:
+async def test_heatmaps_are_claimed_exclusively_across_projects(database, clock, status) -> None:
async with database() as data:
data.add(sql.Project(key="other", name="Other", committee_key="example"))
await data.flush()
@@ -189,9 +241,29 @@ async def test_heatmaps_are_claimed_exclusively_across_projects(database, status
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)
+ if status == sql.TaskStatus.COMPLETED:
+ assert (await worker._task_next_claim())[1] == sql.TaskType.CATALOG_SITE_GENERATE
+ assert await worker._task_next_claim() is None
+ clock.datetime.now.return_value = _NOW + datetime.timedelta(minutes=5)
assert (await worker._task_next_claim())[0] == (ids - {claimed[0]}).pop()
+async def test_maintenance_runs_heatmap_sweep(monkeypatch) -> None:
+ for name in (
+ "_expired_pats_maintenance",
+ "_session_data_maintenance",
+ "_storage_maintenance",
+ "_workflow_auth_maintenance",
+ "_inactivity_maintenance",
+ ):
+ monkeypatch.setattr(maintenance, name, mock.AsyncMock())
+ monkeypatch.setattr(tasks, "schedule_next", mock.AsyncMock())
+ sweep = mock.AsyncMock()
+ monkeypatch.setattr(heatmap, "sweep", sweep)
+ await maintenance.run({"asf_uid": constants.SYSTEM_SERVICE_UID, "next_schedule_seconds": 86400})
+ sweep.assert_awaited_once_with()
+
+
@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
@@ -227,6 +299,184 @@ async def test_queued_analysis_is_stored_and_published(database, monkeypatch, tm
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"))
+ await heatmap.sweep()
+ async with database() as data:
+ assert await data.task(task_type=sql.TaskType.SBOM_HEATMAP, status=sql.TaskStatus.QUEUED).get() is None
+
+
+async def test_sweep_continues_after_a_release_error(database, monkeypatch) -> None:
+ failed = mock.Mock()
+ monkeypatch.setattr(heatmap.log, "exception", failed)
+ async with database() as data:
+ saved = await _saved(data)
+ await data.execute(
+ sqlalchemy.text("UPDATE task SET result = :result WHERE id = :id"),
+ {"result": "{", "id": saved.id},
+ )
+ data.add(sql.Release(project_key="example", version="2.0", phase=sql.ReleasePhase.RELEASE, created=_NOW))
+ await data.flush()
+ data.add(
+ sql.Artifact(
+ project_key="example",
+ version="2.0",
+ release_key="example-2.0",
+ artifact_path="source.zip",
+ sbom_path="source.zip.cdx.json",
+ download_path_suffix="example/2.0",
+ )
+ )
+ await data.commit()
+ await heatmap.sweep()
+ failed.assert_called_once_with("Heatmap sweep failed for example-1.0")
+ async with database() as data:
+ queued = await data.task(task_type=sql.TaskType.SBOM_HEATMAP, status=sql.TaskStatus.QUEUED).all()
+ assert [task.version_key for task in queued] == ["2.0"]
+
+
+@pytest.mark.parametrize("archived", [False, True])
+async def test_sweep_deduplicates_concurrent_runs_and_restarts(database, archived) -> None:
+ async with database() as data:
+ release = await data.release(key="example-1.0").get()
+ release.is_archived = archived
+ await data.commit()
+ await asyncio.gather(heatmap.sweep(), heatmap.sweep())
+ await tasks.clear_scheduled()
+ await heatmap.sweep()
+ async with database() as data:
+ queued = await data.task(task_type=sql.TaskType.SBOM_HEATMAP).all()
+ if archived:
+ assert not queued
+ return
+ assert len(queued) == 1
+ assert queued[0].status == sql.TaskStatus.QUEUED
+ assert queued[0].version_key == "1.0"
+ assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is None
+ await worker._task_next_claim()
+ await heatmap.sweep()
+ async with database() as data:
+ assert len(await data.task(task_type=sql.TaskType.SBOM_HEATMAP).all()) == 1
+
+
+@pytest.mark.parametrize("change", ["missing_sbom", "empty_sbom", "draft", "archived"])
+async def test_sweep_ignores_ineligible_releases(database, change) -> None:
+ async with database() as data:
+ artifact = await data.artifact(project_key="example", version="1.0").get()
+ release = await data.release(key="example-1.0").get()
+ if change == "draft":
+ release.phase = sql.ReleasePhase.RELEASE_CANDIDATE_DRAFT
+ elif change == "archived":
+ release.is_archived = True
+ else:
+ artifact.sbom_path = None if change == "missing_sbom" else ""
+ await data.commit()
+ await heatmap.sweep()
+ async with database() as data:
+ assert await data.task(task_type=sql.TaskType.SBOM_HEATMAP).get() is None
+
+
+@pytest.mark.parametrize("change", ["none", "fresh", "week", "replacement", "analysis_version"])
+async def test_sweep_refreshes_only_missing_changed_or_week_old_reports(database, clock, change) -> None:
+ async with database() as data:
+ if change != "none":
+ saved = await _saved(data)
+ await catalog_site.regenerate_project(data, "example")
+ if change == "fresh":
+ clock.datetime.now.return_value = _NOW + datetime.timedelta(days=7, microseconds=-1)
+ elif change == "week":
+ clock.datetime.now.return_value = _NOW + datetime.timedelta(days=7)
+ elif change == "replacement":
+ artifact = await data.artifact(project_key="example", version="1.0").get()
+ artifact.sbom_path = "replacement.cdx.json"
+ elif change == "analysis_version":
+ saved.result = saved.result.model_copy(update={"analysis_version": 0})
+ await data.commit()
+ await heatmap.sweep()
+ async with database() as data:
+ queued = await data.task(task_type=sql.TaskType.SBOM_HEATMAP, status=sql.TaskStatus.QUEUED).all()
+ assert len(queued) == (0 if change == "fresh" else 1)
+ if change == "fresh":
+ assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is None
+
+
+@pytest.mark.parametrize("change", ["removed", "withdrawn", "archived"])
+async def test_sweep_removes_obsolete_publication_without_reanalysis(database, tmp_path, change) -> None:
+ async with database() as data:
+ await _saved(data)
+ await catalog_site.regenerate_project(data, "example")
+ if change == "removed":
+ artifact = await data.artifact(project_key="example", version="1.0").get()
+ artifact.sbom_path = None
+ else:
+ release = await data.release(key="example-1.0").get()
+ if change == "archived":
+ release.is_archived = True
+ else:
+ release.phase = sql.ReleasePhase.RELEASE_CANDIDATE_DRAFT
+ await data.commit()
+ await heatmap.sweep()
+ async with database() as data:
+ assert await data.task(task_type=sql.TaskType.SBOM_HEATMAP, status=sql.TaskStatus.QUEUED).get() is None
+ regeneration = await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get()
+ await data.delete(regeneration)
+ await data.commit()
+ await catalog_site.regenerate_project(data, "example")
+ assert not (tmp_path / "example/1.0/heatmap.json").exists()
+ assert "heatmap.html" not in (tmp_path / "example/1.0/index.html").read_text()
+ await heatmap.sweep()
+ async with database() as data:
+ assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is None
+
+
+@pytest.mark.parametrize("filename", ["heatmap.json", "heatmap.html", "index.html"])
+@pytest.mark.parametrize("damage", ["missing", "stale"])
+async def test_sweep_repairs_publication_without_reanalysis(database, tmp_path, filename, damage) -> None:
+ async with database() as data:
+ await _saved(data)
+ await catalog_site.regenerate_project(data, "example")
+ path = tmp_path / "example/1.0" / filename
+ if damage == "missing":
+ path.unlink()
+ else:
+ timestamp = (_NOW - datetime.timedelta(seconds=1)).timestamp()
+ os.utime(path, (timestamp, timestamp))
+ await heatmap.sweep()
+ await heatmap.sweep()
+ async with database() as data:
+ assert len(await data.task(task_type=sql.TaskType.SBOM_HEATMAP).all()) == 1
+ regeneration = await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get()
+ assert regeneration.project_key == "example"
+ await data.delete(regeneration)
+ await data.commit()
+ await catalog_site.regenerate_project(data, "example")
+ await heatmap.sweep()
+ async with database() as data:
+ assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is None
+
+
+@pytest.mark.parametrize("status", [sql.TaskStatus.FAILED, sql.TaskStatus.BROKEN])
+async def test_sweep_waits_a_day_after_failure_and_keeps_previous_report(database, clock, status) -> None:
+ async with database() as data:
+ saved = await _saved(data)
+ saved.added = _NOW - datetime.timedelta(days=9)
+ saved.completed = _NOW - datetime.timedelta(days=8)
+ await data.commit()
+ await catalog_site.regenerate_project(data, "example")
+ failed = await heatmap.queue(data, _PROJECT, _VERSION)
+ failed.added = _NOW - datetime.timedelta(days=2)
+ failed.status = status
+ failed.completed = _NOW
+ failed.error = "unavailable"
+ await data.commit()
+ clock.datetime.now.return_value = _NOW + datetime.timedelta(hours=23)
+ await heatmap.sweep()
+ async with database() as data:
+ assert await data.task(task_type=sql.TaskType.SBOM_HEATMAP, status=sql.TaskStatus.QUEUED).get() is None
+ assert (await data.task(id=saved.id).get()).result == _snapshot()
+ assert await data.task(task_type=sql.TaskType.CATALOG_SITE_GENERATE).get() is None
+ clock.datetime.now.return_value = _NOW + datetime.timedelta(days=1)
+ await heatmap.sweep()
+ async with database() as data:
+ assert await data.task(task_type=sql.TaskType.SBOM_HEATMAP, status=sql.TaskStatus.QUEUED).get() is not None
@pytest.mark.parametrize("published", [False, True])
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org