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/17 16:39:16 UTC
(tooling-trusted-releases) branch sbp updated: Update tasks when catalogue releases change
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 0d156e9a Update tasks when catalogue releases change
0d156e9a is described below
commit 0d156e9aabb3b66325c2feeb865407d71b7613fa
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Thu Sep 17 17:38:52 2026 +0100
Update tasks when catalogue releases change
---
atr/storage/writers/catalogue.py | 23 ++++++++
tests/unit/test_catalogue_import_apply.py | 96 +++++++++++++++++++++++++++++++
2 files changed, 119 insertions(+)
diff --git a/atr/storage/writers/catalogue.py b/atr/storage/writers/catalogue.py
index 50797584..30cbe97d 100644
--- a/atr/storage/writers/catalogue.py
+++ b/atr/storage/writers/catalogue.py
@@ -356,6 +356,11 @@ class FoundationAdmin:
if not release_keys:
return
via = sql.validate_instrumented_attribute
+ releases = sqlmodel.select(via(sql.Release.project_key), via(sql.Release.version)).where(
+ via(sql.Release.key).in_(release_keys)
+ )
+ identity = sqlalchemy.tuple_(via(sql.Task.project_key), via(sql.Task.version_key))
+ await self.__data.execute(sqlmodel.delete(sql.Task).where(identity.in_(releases)))
for model, attr in repoint.RELEASE_KEY_REFS:
if model is sql.Release:
continue
@@ -561,8 +566,26 @@ class FoundationAdmin:
)
async def _repoint_release_rows(self, old_key: str, from_project: str, version: str, to_project: str) -> None:
+ pending = await (
+ self.__data.task(
+ project_key=from_project,
+ version_key=version,
+ status_in=[sql.TaskStatus.QUEUED, sql.TaskStatus.ACTIVE],
+ )
+ .limit(1)
+ .get()
+ )
+ if pending is not None:
+ raise storage.AccessError(
+ f"Release '{old_key}' has queued or active tasks; wait for them before moving it.", status=409
+ )
via = sql.validate_instrumented_attribute
new_key = f"{to_project}-{version}"
+ await self.__data.execute(
+ sqlmodel.update(sql.Task)
+ .where(via(sql.Task.project_key) == from_project, via(sql.Task.version_key) == version)
+ .values(project_key=to_project)
+ )
await self.__data.execute(
sqlmodel.update(sql.Artifact)
.where(via(sql.Artifact.project_key) == from_project)
diff --git a/tests/unit/test_catalogue_import_apply.py b/tests/unit/test_catalogue_import_apply.py
index a6243b15..e9b73b1d 100644
--- a/tests/unit/test_catalogue_import_apply.py
+++ b/tests/unit/test_catalogue_import_apply.py
@@ -33,6 +33,7 @@ import sqlalchemy.pool
import sqlmodel
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.shared.catalogue_diff as catalogue_diff
@@ -114,6 +115,80 @@ async def test_an_artifact_repoint_survives_a_repoint_of_the_release_it_came_fro
assert await data.get(sql.Artifact, ("alpha-two", "1.0.0", "b.tgz")) is not None
+@pytest.mark.parametrize("operation", ["delete", "move", "import"])
+@pytest.mark.parametrize("status", [sql.TaskStatus.COMPLETED, sql.TaskStatus.FAILED, sql.TaskStatus.BROKEN])
+async def test_catalogue_release_changes_update_only_matching_tasks(sessionmaker, operation, status) -> None:
+ async with sessionmaker() as data:
+ await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0"})
+ data.add(sql.Project(key="alpha", name="Apache Alpha", committee_key="alpha"))
+ await data.commit()
+ tasks = [
+ _task("alpha-one", "1.0.0", status),
+ _task("alpha-one", "2.0.0", status),
+ _task("alpha-two", "1.0.0", status),
+ _task("alpha", "one-1.0.0", status),
+ _task("alpha-one", None, sql.TaskStatus.QUEUED),
+ ]
+ data.add_all(tasks)
+ await data.commit()
+ before = [task.model_dump() for task in tasks]
+
+ if operation == "delete":
+ await _writer(data).import_catalogue_csvs({"releases": []}, "alpha", catalogue_diff.Mode.REPLACE)
+ elif operation == "move":
+ await _writer(data).move_release(safe.ReleaseKey("alpha-one-1.0.0"), safe.ProjectKey("alpha-two"))
+ else:
+ row = _release_row("alpha-one", "1.0.0")
+ row["project_key"] = "alpha-two"
+ await _writer(data).import_catalogue_csvs({"releases": [row]}, "alpha", catalogue_diff.Mode.ADDITIVE)
+
+ data.expire_all()
+ changed = await data.get(sql.Task, before[0]["id"])
+ if operation == "delete":
+ assert changed is None
+ else:
+ assert changed is not None
+ assert changed.model_dump() == {**before[0], "project_key": "alpha-two"}
+ for expected in before[1:]:
+ unchanged = await data.get(sql.Task, expected["id"])
+ assert unchanged is not None
+ assert unchanged.model_dump() == expected
+ assert not (await data.execute(sqlalchemy.text("PRAGMA foreign_key_check"))).all()
+
+
+@pytest.mark.parametrize("operation", ["delete", "move", "import"])
+@pytest.mark.parametrize("status", [sql.TaskStatus.QUEUED, sql.TaskStatus.ACTIVE])
+async def test_catalogue_release_changes_with_unfinished_tasks(sessionmaker, operation, status) -> None:
+ async with sessionmaker() as data:
+ await _seed_two_projects_with_release(data, {"alpha-one": "1.0.0"})
+ task = _task("alpha-one", "1.0.0", status)
+ data.add(task)
+ await data.commit()
+ before = task.model_dump()
+
+ if operation == "delete":
+ await _writer(data).import_catalogue_csvs({"releases": []}, "alpha", catalogue_diff.Mode.REPLACE)
+ data.expire_all()
+ assert not await data.task(project_key="alpha-one", version_key="1.0.0").all()
+ assert await data.get(sql.Release, "alpha-one-1.0.0") is None
+ return
+
+ with pytest.raises(catalogue.storage.AccessError, match="queued or active tasks"):
+ if operation == "move":
+ await _writer(data).move_release(safe.ReleaseKey("alpha-one-1.0.0"), safe.ProjectKey("alpha-two"))
+ else:
+ row = _release_row("alpha-one", "1.0.0")
+ row["project_key"] = "alpha-two"
+ await _writer(data).import_catalogue_csvs({"releases": [row]}, "alpha", catalogue_diff.Mode.ADDITIVE)
+
+ data.expire_all()
+ unchanged = await data.get(sql.Task, before["id"])
+ assert unchanged is not None
+ assert unchanged.model_dump() == before
+ assert await data.get(sql.Release, "alpha-one-1.0.0") is not None
+ assert await data.get(sql.Release, "alpha-two-1.0.0") is None
+
+
@pytest.mark.asyncio
async def test_export_leaves_out_releases_with_live_workflow_data(sessionmaker) -> None:
# The import refuses those rows, so an export that carried them could not be uploaded again.
@@ -568,6 +643,27 @@ async def _seed_two_projects_with_release(data: db.Session, versioned_projects:
await data.commit()
+def _task(project: str, version: str | None, status: sql.TaskStatus) -> sql.Task:
+ created = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
+ return sql.Task(
+ task_type=sql.TaskType.MESSAGE_SEND,
+ task_args={"project_key": project, "version_key": version, "subject": "Original subject"},
+ project_key=project,
+ version_key=version,
+ asf_uid="tester",
+ status=status,
+ started=created if status == sql.TaskStatus.ACTIVE else None,
+ pid=1 if status == sql.TaskStatus.ACTIVE else None,
+ completed=created
+ if status in (sql.TaskStatus.COMPLETED, sql.TaskStatus.FAILED, sql.TaskStatus.BROKEN)
+ else None,
+ result=results.MessageSend(kind="message_send", mid="original@example.invalid", mail_send_warnings=[])
+ if status == sql.TaskStatus.COMPLETED
+ else None,
+ error="Failed" if status in (sql.TaskStatus.FAILED, sql.TaskStatus.BROKEN) else None,
+ )
+
+
def _writer(data: db.Session) -> catalogue.FoundationAdmin:
writer = object.__new__(catalogue.FoundationAdmin)
writer._FoundationAdmin__data = data
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org