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:40 UTC
(tooling-trusted-releases) 01/01: Schedule the generation of SBOM heatmap tables
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