From 3efe360a7262f803b9c8645cb000706cd20e4500 Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Mon, 24 Aug 2026 12:57:28 +0100 Subject: [PATCH 01/13] Make a worker-thread submission reach the event bus `asyncio.Queue` is not thread-safe, and a `SyncProcessor` runs in an executor thread. A `put_nowait` from there landing between `Queue.get`'s `empty()` check and its waiter being registered wakes nobody: the message stays in the deque, the loop awaits a waiter that will never resolve, and because the queue is not empty the bus never reaches its stop condition and spins until `max_timeout`. Submissions from off the loop thread now hop onto it with `call_soon_threadsafe`. A caller already on that thread, or with no bus running, still puts directly, so a message is queued by the time `submit_message` returns. The loop is captured in `initialize`, the first point at which one exists: processors are registered from `__init__`, on a thread that has none. Co-Authored-By: Claude Opus 5 (1M context) --- ddev/src/ddev/event_bus/orchestrator.py | 34 ++++++++++++++++++- ddev/tests/event_bus/test_event_bus.py | 43 +++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/ddev/src/ddev/event_bus/orchestrator.py b/ddev/src/ddev/event_bus/orchestrator.py index 67acf07206a42..e460863ef9075 100644 --- a/ddev/src/ddev/event_bus/orchestrator.py +++ b/ddev/src/ddev/event_bus/orchestrator.py @@ -48,10 +48,20 @@ class BaseMessage: id: str +def running_loop() -> asyncio.AbstractEventLoop | None: + """The loop running on this thread, or ``None`` outside one.""" + try: + return asyncio.get_running_loop() + except RuntimeError: + return None + + class BaseProcessor[T: BaseMessage]: def __init__(self, name: str): self.name = name self.queue: asyncio.Queue[BaseMessage] | None = None + # Both set by the bus: the queue at registration, the loop when the bus starts. + self.loop: asyncio.AbstractEventLoop | None = None async def on_success(self, message: T) -> None: pass @@ -76,9 +86,25 @@ async def on_error(self, error: MessageProcessingError | ProcessorHookError) -> raise error def submit_message(self, message: BaseMessage) -> None: + """Put *message* on the bus, from any thread. + + A ``SyncProcessor`` runs in an executor thread, and ``asyncio.Queue`` is not thread-safe: a + put landing between ``get``'s ``empty()`` check and its waiter being registered wakes nobody, + leaving the bus spinning on a message it never reads. Such a put is handed to the loop + thread instead. A caller already on that thread, or with no bus running, puts directly, so + the message is queued by the time this returns. + """ if self.queue is None: raise ProcessorQueueError("This processor has not been added to an active event bus") - self.queue.put_nowait(message) + + if self.loop is None or running_loop() is self.loop: + self.queue.put_nowait(message) + return + + try: + self.loop.call_soon_threadsafe(self.queue.put_nowait, message) + except RuntimeError as error: + raise ProcessorQueueError("The event bus is no longer running") from error def should_process_message(self, message: BaseMessage) -> bool: return True @@ -196,6 +222,12 @@ async def initialize(self): Initializes the orchestrator. """ self._running = True + # The first point a running loop exists: processors are registered from __init__, on a + # thread that has none. Each one needs it to submit safely from an executor thread. + loop = asyncio.get_running_loop() + for processors in self._subscribers.values(): + for processor in processors: + processor.loop = loop try: await self.on_initialize() except (FatalProcessingError, asyncio.CancelledError): diff --git a/ddev/tests/event_bus/test_event_bus.py b/ddev/tests/event_bus/test_event_bus.py index 529bf2cfc9c63..3139a8d31bca4 100644 --- a/ddev/tests/event_bus/test_event_bus.py +++ b/ddev/tests/event_bus/test_event_bus.py @@ -6,6 +6,7 @@ import asyncio import logging import math +import threading import time from collections.abc import Generator from contextlib import AbstractContextManager, contextmanager, suppress @@ -744,6 +745,48 @@ def process_message(self, message: Announcement): assert secretary.delivered_memos[0].id == "async_memo" +def test_sync_processor_submits_on_the_loop_thread(analyst: Analyst): + """A sync processor's submission must reach the queue on the loop thread. + + `asyncio.Queue` is not thread-safe. A put from an executor thread that lands between `get`'s + `empty()` check and its waiter being registered wakes nobody, so the message is never read and + the bus spins until `max_timeout` — with the results of whatever produced it lost. The window is + a few bytecodes wide, so this asserts the invariant that closes it rather than trying to lose the + race on demand. + """ + + class RecordingQueue(asyncio.Queue): + def __init__(self): + super().__init__() + self.put_threads: list[int] = [] + + def put_nowait(self, item): + self.put_threads.append(threading.get_ident()) + super().put_nowait(item) + + class Delegator(SyncProcessor[Memo]): + def process_message(self, message: Memo): + self.submit_message(Announcement(id="delegated", announcement_type="FromWorkerThread")) + + logger = logging.getLogger("test_thread_safe_submit") + orchestrator = MockOrchestrator(logger, max_timeout=10, grace_period=0.1) + queue = RecordingQueue() + # Replaced before registration, which is what hands the queue to each processor. + orchestrator._queue = queue + orchestrator.register_processor(Delegator("delegator"), [Memo]) + orchestrator.register_processor(analyst, [Announcement]) + + orchestrator.submit_message(Memo("delegate_me")) + + with assert_time(0, 5.0): + orchestrator.run() + + # The chain completed rather than stalling on a message nobody woke up for. + assert [message.id for message in analyst.completed_tasks] == ["delegated"] + # Every put landed on the loop thread, which `asyncio.run` runs on the calling thread. + assert set(queue.put_threads) == {threading.get_ident()} + + def test_processor_submit_without_bus(): processor = Secretary("orphan") with pytest.raises(ProcessorQueueError, match="This processor has not been added"): From ffc7765ccb131a75f245df78a4cc7ff30ae08942 Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Mon, 24 Aug 2026 12:58:26 +0100 Subject: [PATCH 02/13] Add changelog entry --- ddev/changelog.d/24962.fixed | 1 + 1 file changed, 1 insertion(+) create mode 100644 ddev/changelog.d/24962.fixed diff --git a/ddev/changelog.d/24962.fixed b/ddev/changelog.d/24962.fixed new file mode 100644 index 0000000000000..5abe1f88ad399 --- /dev/null +++ b/ddev/changelog.d/24962.fixed @@ -0,0 +1 @@ +Fix a message submitted from a worker thread being lost, leaving the event bus running until its timeout. From 53983b69fbc0d57418e95b29ce8a3a59f872677d Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Mon, 24 Aug 2026 14:26:28 +0100 Subject: [PATCH 03/13] Close the remaining loop-capture gaps and assert delivery `register_processor` handed a processor its queue but not the loop, and `initialize` captured the loop before `on_initialize` runs. A `SyncProcessor` registered from that hook kept `loop is None` and submitted directly from its executor thread, which is the case this fix exists to remove. It now takes the loop at registration when the bus is already running. `asyncio.run` closes the loop when the bus stops, so a submission afterwards tried to schedule onto a dead loop and raised. A closed loop is queued into directly instead, as it was before there was a loop to hop onto. The regression test asserted which thread each put came from, which pinned it to `call_soon_threadsafe` rather than to the contract. It now asserts only that the message is delivered, against a queue that drops every off-loop put. Co-Authored-By: Claude Opus 5 (1M context) --- ddev/src/ddev/event_bus/orchestrator.py | 12 +++++-- ddev/tests/event_bus/test_event_bus.py | 45 +++++++++++++------------ 2 files changed, 33 insertions(+), 24 deletions(-) diff --git a/ddev/src/ddev/event_bus/orchestrator.py b/ddev/src/ddev/event_bus/orchestrator.py index e460863ef9075..9a43bda8fb331 100644 --- a/ddev/src/ddev/event_bus/orchestrator.py +++ b/ddev/src/ddev/event_bus/orchestrator.py @@ -91,13 +91,15 @@ def submit_message(self, message: BaseMessage) -> None: A ``SyncProcessor`` runs in an executor thread, and ``asyncio.Queue`` is not thread-safe: a put landing between ``get``'s ``empty()`` check and its waiter being registered wakes nobody, leaving the bus spinning on a message it never reads. Such a put is handed to the loop - thread instead. A caller already on that thread, or with no bus running, puts directly, so - the message is queued by the time this returns. + thread instead. A caller already on that thread, or with no live loop to hand it to, puts + directly, so the message is queued by the time this returns. """ if self.queue is None: raise ProcessorQueueError("This processor has not been added to an active event bus") - if self.loop is None or running_loop() is self.loop: + # ``asyncio.run`` closes the loop when the bus stops, and a stopped bus is queued into + # directly, as it was before there was a loop to hop onto. + if self.loop is None or self.loop.is_closed() or running_loop() is self.loop: self.queue.put_nowait(message) return @@ -181,6 +183,10 @@ def __validate_parameters(self, max_timeout: float, grace_period: float): def register_processor[T: BaseMessage](self, processor: Processor[T], message_types: list[type[T]]): """Registers a processor to receive specific message types.""" processor.queue = self._queue + # Registering while the bus runs — from `on_initialize`, say — still needs the loop, or a + # `SyncProcessor` added that way submits unsafely from its executor thread. + if self._running: + processor.loop = asyncio.get_running_loop() for msg_type in message_types: self._subscribers.setdefault(msg_type, []).append(processor) diff --git a/ddev/tests/event_bus/test_event_bus.py b/ddev/tests/event_bus/test_event_bus.py index 3139a8d31bca4..05953b4774b47 100644 --- a/ddev/tests/event_bus/test_event_bus.py +++ b/ddev/tests/event_bus/test_event_bus.py @@ -745,23 +745,30 @@ def process_message(self, message: Announcement): assert secretary.delivered_memos[0].id == "async_memo" -def test_sync_processor_submits_on_the_loop_thread(analyst: Analyst): - """A sync processor's submission must reach the queue on the loop thread. - - `asyncio.Queue` is not thread-safe. A put from an executor thread that lands between `get`'s - `empty()` check and its waiter being registered wakes nobody, so the message is never read and - the bus spins until `max_timeout` — with the results of whatever produced it lost. The window is - a few bytecodes wide, so this asserts the invariant that closes it rather than trying to lose the - race on demand. +def test_a_worker_thread_submission_is_delivered(analyst: Analyst): + """A `SyncProcessor` runs in an executor thread, and what it submits must still be delivered. + + `asyncio.Queue` is not thread-safe: a put from another thread landing between `get`'s `empty()` + check and its waiter being registered wakes nobody, so the message is never read and the bus + spins until `max_timeout` with the results of whatever produced it lost. + + That window is a few bytecodes wide and cannot be forced without reimplementing `Queue.get`, so + the queue below stands in for it: it loses every off-loop put rather than the unlucky ones. That + is stricter than the real queue on purpose — it turns "delivery depends on winning a race" into a + deterministic failure, and leaves the assertion on the contract that matters, which is that the + message arrives at all. """ - class RecordingQueue(asyncio.Queue): - def __init__(self): + class LoseOffLoopPuts(asyncio.Queue): + """Drops a put made anywhere but the loop thread, the worst case of the real race.""" + + def __init__(self, loop_thread: int): super().__init__() - self.put_threads: list[int] = [] + self._loop_thread = loop_thread def put_nowait(self, item): - self.put_threads.append(threading.get_ident()) + if threading.get_ident() != self._loop_thread: + return super().put_nowait(item) class Delegator(SyncProcessor[Memo]): @@ -769,22 +776,18 @@ def process_message(self, message: Memo): self.submit_message(Announcement(id="delegated", announcement_type="FromWorkerThread")) logger = logging.getLogger("test_thread_safe_submit") - orchestrator = MockOrchestrator(logger, max_timeout=10, grace_period=0.1) - queue = RecordingQueue() - # Replaced before registration, which is what hands the queue to each processor. - orchestrator._queue = queue + orchestrator = MockOrchestrator(logger, max_timeout=2, grace_period=0.1) + # `asyncio.run` runs the loop on the calling thread. Replaced before registration, which is what + # hands the queue to each processor. + orchestrator._queue = LoseOffLoopPuts(threading.get_ident()) orchestrator.register_processor(Delegator("delegator"), [Memo]) orchestrator.register_processor(analyst, [Announcement]) orchestrator.submit_message(Memo("delegate_me")) - with assert_time(0, 5.0): - orchestrator.run() + orchestrator.run() - # The chain completed rather than stalling on a message nobody woke up for. assert [message.id for message in analyst.completed_tasks] == ["delegated"] - # Every put landed on the loop thread, which `asyncio.run` runs on the calling thread. - assert set(queue.put_threads) == {threading.get_ident()} def test_processor_submit_without_bus(): From 9abad916833f6c390190d7afee822b6d412f08e1 Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Mon, 24 Aug 2026 15:09:26 +0100 Subject: [PATCH 04/13] Trim the worker-thread delivery test's docstring Co-Authored-By: Claude Opus 5 (1M context) --- ddev/tests/event_bus/test_event_bus.py | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/ddev/tests/event_bus/test_event_bus.py b/ddev/tests/event_bus/test_event_bus.py index 05953b4774b47..77545e12b665d 100644 --- a/ddev/tests/event_bus/test_event_bus.py +++ b/ddev/tests/event_bus/test_event_bus.py @@ -746,17 +746,10 @@ def process_message(self, message: Announcement): def test_a_worker_thread_submission_is_delivered(analyst: Analyst): - """A `SyncProcessor` runs in an executor thread, and what it submits must still be delivered. + """A `SyncProcessor` submits from an executor thread, where `asyncio.Queue` can lose the put. - `asyncio.Queue` is not thread-safe: a put from another thread landing between `get`'s `empty()` - check and its waiter being registered wakes nobody, so the message is never read and the bus - spins until `max_timeout` with the results of whatever produced it lost. - - That window is a few bytecodes wide and cannot be forced without reimplementing `Queue.get`, so - the queue below stands in for it: it loses every off-loop put rather than the unlucky ones. That - is stricter than the real queue on purpose — it turns "delivery depends on winning a race" into a - deterministic failure, and leaves the assertion on the contract that matters, which is that the - message arrives at all. + The real queue loses only the unlucky ones; this one loses every off-loop put, so delivery fails + deterministically rather than by winning a race. """ class LoseOffLoopPuts(asyncio.Queue): @@ -777,8 +770,7 @@ def process_message(self, message: Memo): logger = logging.getLogger("test_thread_safe_submit") orchestrator = MockOrchestrator(logger, max_timeout=2, grace_period=0.1) - # `asyncio.run` runs the loop on the calling thread. Replaced before registration, which is what - # hands the queue to each processor. + # `asyncio.run` runs the loop here, and registration is what hands the queue to each processor. orchestrator._queue = LoseOffLoopPuts(threading.get_ident()) orchestrator.register_processor(Delegator("delegator"), [Memo]) orchestrator.register_processor(analyst, [Announcement]) From 496d3936bcbb4cc685ee3deeb3a09148ca5ce357 Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Tue, 25 Aug 2026 11:36:34 +0100 Subject: [PATCH 05/13] Give each processor a reference to its bus Restores the orchestrator's original design: a processor holds the bus it was registered in and delegates submit_message to it, so the thread-safe put lives in one place instead of being assembled from two injected variables. The loop is read at submit time rather than captured per processor, which drops running_loop(), processor.loop, processor.queue, the is_closed() check, the on-loop fast path, the RuntimeError guard, and the duplicate submit path. It also makes a mid-run registration correct without a branch. Every put now goes through the loop thread while the bus runs, so a worker's earlier submit can no longer land behind a later on-loop one. The bus captures the loop after on_initialize: that hook's submits have no task completion to be ordered behind, and a zero grace period stops the bus before a deferred put would run. Co-Authored-By: Claude Opus 5 (1M context) --- ddev/src/ddev/event_bus/exceptions.py | 2 +- ddev/src/ddev/event_bus/orchestrator.py | 66 ++++++------------ ddev/tests/cli/ci/tests/helpers.py | 10 +++ .../cli/ci/tests/test_task_test_gatherer.py | 63 +++++++++-------- .../cli/ci/tests/test_task_test_runner.py | 37 +++++----- ddev/tests/event_bus/test_event_bus.py | 68 +++++++++++++++++-- 6 files changed, 144 insertions(+), 102 deletions(-) diff --git a/ddev/src/ddev/event_bus/exceptions.py b/ddev/src/ddev/event_bus/exceptions.py index eea465d094ba8..d92a42aa3205e 100644 --- a/ddev/src/ddev/event_bus/exceptions.py +++ b/ddev/src/ddev/event_bus/exceptions.py @@ -21,7 +21,7 @@ class HookName(StrEnum): class ProcessorQueueError(Exception): """ - Exception raised when a processor queue is not initialized. + Exception raised when a processor has not been registered in an event bus. """ pass diff --git a/ddev/src/ddev/event_bus/orchestrator.py b/ddev/src/ddev/event_bus/orchestrator.py index 9a43bda8fb331..acac1d5a489d3 100644 --- a/ddev/src/ddev/event_bus/orchestrator.py +++ b/ddev/src/ddev/event_bus/orchestrator.py @@ -48,20 +48,11 @@ class BaseMessage: id: str -def running_loop() -> asyncio.AbstractEventLoop | None: - """The loop running on this thread, or ``None`` outside one.""" - try: - return asyncio.get_running_loop() - except RuntimeError: - return None - - class BaseProcessor[T: BaseMessage]: def __init__(self, name: str): self.name = name - self.queue: asyncio.Queue[BaseMessage] | None = None - # Both set by the bus: the queue at registration, the loop when the bus starts. - self.loop: asyncio.AbstractEventLoop | None = None + # Set by the bus at registration. + self.bus: EventBusOrchestrator | None = None async def on_success(self, message: T) -> None: pass @@ -86,27 +77,11 @@ async def on_error(self, error: MessageProcessingError | ProcessorHookError) -> raise error def submit_message(self, message: BaseMessage) -> None: - """Put *message* on the bus, from any thread. - - A ``SyncProcessor`` runs in an executor thread, and ``asyncio.Queue`` is not thread-safe: a - put landing between ``get``'s ``empty()`` check and its waiter being registered wakes nobody, - leaving the bus spinning on a message it never reads. Such a put is handed to the loop - thread instead. A caller already on that thread, or with no live loop to hand it to, puts - directly, so the message is queued by the time this returns. - """ - if self.queue is None: + """Put *message* on the bus this processor was registered in, from any thread.""" + if self.bus is None: raise ProcessorQueueError("This processor has not been added to an active event bus") - # ``asyncio.run`` closes the loop when the bus stops, and a stopped bus is queued into - # directly, as it was before there was a loop to hop onto. - if self.loop is None or self.loop.is_closed() or running_loop() is self.loop: - self.queue.put_nowait(message) - return - - try: - self.loop.call_soon_threadsafe(self.queue.put_nowait, message) - except RuntimeError as error: - raise ProcessorQueueError("The event bus is no longer running") from error + self.bus.submit_message(message) def should_process_message(self, message: BaseMessage) -> bool: return True @@ -168,6 +143,7 @@ def __init__( # These will be initialized in the running loop self._queue = asyncio.Queue[BaseMessage]() self._running = False + self._loop: asyncio.AbstractEventLoop | None = None def __validate_parameters(self, max_timeout: float, grace_period: float): """ @@ -182,19 +158,23 @@ def __validate_parameters(self, max_timeout: float, grace_period: float): def register_processor[T: BaseMessage](self, processor: Processor[T], message_types: list[type[T]]): """Registers a processor to receive specific message types.""" - processor.queue = self._queue - # Registering while the bus runs — from `on_initialize`, say — still needs the loop, or a - # `SyncProcessor` added that way submits unsafely from its executor thread. - if self._running: - processor.loop = asyncio.get_running_loop() + processor.bus = self for msg_type in message_types: self._subscribers.setdefault(msg_type, []).append(processor) def submit_message(self, message: BaseMessage): + """Adds a message to the queue, from any thread. + + A ``SyncProcessor`` runs in an executor thread, and ``asyncio.Queue`` is not thread-safe: a + put landing between ``get``'s ``empty()`` check and its waiter being registered wakes nobody, + leaving the bus spinning on a message it never reads. So while the bus runs, every put is + handed to the loop thread, whichever thread asked for it. Before it starts and after it + stops there is no loop to hand it to, and the queue takes the message directly. """ - Adds a message to the queue. - """ - self._queue.put_nowait(message) + if self._loop is not None and self._loop.is_running(): + self._loop.call_soon_threadsafe(self._queue.put_nowait, message) + else: + self._queue.put_nowait(message) def run(self): """ @@ -228,12 +208,6 @@ async def initialize(self): Initializes the orchestrator. """ self._running = True - # The first point a running loop exists: processors are registered from __init__, on a - # thread that has none. Each one needs it to submit safely from an executor thread. - loop = asyncio.get_running_loop() - for processors in self._subscribers.values(): - for processor in processors: - processor.loop = loop try: await self.on_initialize() except (FatalProcessingError, asyncio.CancelledError): @@ -243,6 +217,10 @@ async def initialize(self): OrchestratorHookError(HookName.ON_INITIALIZE, e), self.on_error, ) + # Only now, so what the hook submitted is already queued. A deferred put is read because the + # callback that queues it precedes the task completion that wakes the loop, and the hook has + # no task to be ordered behind: a zero grace period would stop the bus before it ran. + self._loop = asyncio.get_running_loop() @abstractmethod async def on_initialize(self): # pragma: no cover diff --git a/ddev/tests/cli/ci/tests/helpers.py b/ddev/tests/cli/ci/tests/helpers.py index 5ba76c503f3a3..f463273566f22 100644 --- a/ddev/tests/cli/ci/tests/helpers.py +++ b/ddev/tests/cli/ci/tests/helpers.py @@ -157,6 +157,16 @@ def copied(source: str, destination: str) -> ChangedFile: return ChangedFile(change_type=ChangeType.COPIED, path=destination, previous_path=source) +class RecordingBus: + """Stands in for the event bus in processor unit tests, recording what the processor submits.""" + + def __init__(self): + self.queue: asyncio.Queue[BaseMessage] = asyncio.Queue() + + def submit_message(self, message: BaseMessage) -> None: + self.queue.put_nowait(message) + + def drain_queue(queue: asyncio.Queue[BaseMessage]) -> list[BaseMessage]: messages: list[BaseMessage] = [] while not queue.empty(): diff --git a/ddev/tests/cli/ci/tests/test_task_test_gatherer.py b/ddev/tests/cli/ci/tests/test_task_test_gatherer.py index 293491196c379..b77e356092d4e 100644 --- a/ddev/tests/cli/ci/tests/test_task_test_gatherer.py +++ b/ddev/tests/cli/ci/tests/test_task_test_gatherer.py @@ -11,7 +11,6 @@ from __future__ import annotations -import asyncio import logging import shutil import threading @@ -37,7 +36,7 @@ from ddev.utils.github_async.models import JobStep, WorkflowJob from ddev.utils.junit import TestStatus from ddev.utils.platform import PlatformName -from tests.cli.ci.tests.helpers import drain_queue, jobs_reported, make_job +from tests.cli.ci.tests.helpers import RecordingBus, drain_queue, jobs_reported, make_job from tests.helpers.github_async import FakeAsyncGitHubClient # --------------------------------------------------------------------------- @@ -146,7 +145,7 @@ def _make_gatherer(tmp_path: Path, plan: dict[str, list[BatchJob]] | None = None output_base_path=tmp_path / "out", batches=[_test_batch(batch_id, jobs) for batch_id, jobs in plan.items()], ) - gatherer.queue = asyncio.Queue() + gatherer.bus = RecordingBus() # type: ignore[assignment] return gatherer @@ -198,7 +197,7 @@ def test_happy_path_organizes_artifacts_and_emits_update(tmp_path: Path): ) ) - messages = drain_queue(gatherer.queue) + messages = drain_queue(gatherer.bus.queue) assert len(messages) == 1 update = messages[0] assert isinstance(update, UpdatePRComment) @@ -232,7 +231,7 @@ def test_failure_path_records_failed_steps_and_reports(tmp_path: Path): ) ) - drain_queue(gatherer.queue) + drain_queue(gatherer.bus.queue) [status] = _registry(gatherer) assert status.failed_count == 1 @@ -274,7 +273,7 @@ def test_timed_out_batch_marks_all_jobs_failed(tmp_path: Path) -> None: batch_jobs = [_batch_job_result(job) for job in jobs] gatherer.process_message(_batch_finished("", status="failure", run_id=300, batch_jobs=batch_jobs, timed_out=True)) - drain_queue(gatherer.queue) + drain_queue(gatherer.bus.queue) [status] = _registry(gatherer) assert status.failed_count == 2 # The timeout is the batch's, not a step of any job: no step name is invented for it. @@ -295,7 +294,7 @@ def test_multiple_jobs_aggregate_into_one_workflow_status(tmp_path: Path): ] gatherer.process_message(_batch_finished(artifacts, status="failure", batch_jobs=batch_jobs)) - drain_queue(gatherer.queue) + drain_queue(gatherer.bus.queue) [status] = _registry(gatherer) assert status.success_count == 1 assert status.failed_count == 1 @@ -367,7 +366,7 @@ def test_emits_update_per_batch_done_on_last(tmp_path: Path) -> None: ) # First of two batches: an update is emitted immediately (live updates), but not yet done. - first = drain_queue(gatherer.queue) + first = drain_queue(gatherer.bus.queue) assert len(first) == 1 assert first[0].revision == 1 assert first[0].progress.done is False @@ -385,7 +384,7 @@ def test_emits_update_per_batch_done_on_last(tmp_path: Path) -> None: ) # Final batch: revision 2, done, aggregating both runs. - second = drain_queue(gatherer.queue) + second = drain_queue(gatherer.bus.queue) assert len(second) == 1 assert second[0].revision == 2 assert second[0].progress.done is True @@ -504,7 +503,7 @@ def test_empty_batch_jobs_still_terminates_the_batch(tmp_path: Path) -> None: gatherer = _make_gatherer(tmp_path) gatherer.process_message(_batch_finished("", status="failure", run_id=100, batch_jobs=[])) - update = drain_queue(gatherer.queue)[0] + update = drain_queue(gatherer.bus.queue)[0] assert update.revision == 1 batch = update.progress.batches[0] assert batch.state == ExecutionState.FINISHED @@ -522,7 +521,7 @@ def test_empty_batch_does_not_block_completion(tmp_path: Path) -> None: gatherer = _make_gatherer(tmp_path, _one_job_plan("b1", "b2")) gatherer.process_message(_batch_finished("", id="b1", run_id=100, batch_jobs=[])) - assert drain_queue(gatherer.queue)[0].progress.done is False + assert drain_queue(gatherer.bus.queue)[0].progress.done is False gatherer.process_message( _batch_finished( @@ -533,7 +532,7 @@ def test_empty_batch_does_not_block_completion(tmp_path: Path) -> None: ) ) - final = drain_queue(gatherer.queue)[0] + final = drain_queue(gatherer.bus.queue)[0] assert final.progress.done is True assert _batch_progress(final, "b1").error == ProgressError.NO_JOB_RESULTS @@ -552,7 +551,7 @@ def test_unplanned_batch_is_ignored(tmp_path: Path) -> None: ) ) - assert drain_queue(gatherer.queue) == [] + assert drain_queue(gatherer.bus.queue) == [] assert gatherer._revision == 0 assert [batch.batch_id for batch in gatherer.build_initial_update("initial").progress.batches] == ["batch-1"] # Nor may it write into the output tree the planned batches publish from. @@ -570,7 +569,7 @@ def test_duplicate_batch_finished_is_ignored(tmp_path: Path) -> None: gatherer = _make_gatherer(tmp_path) gatherer.process_message(batch) - first = drain_queue(gatherer.queue) + first = drain_queue(gatherer.bus.queue) assert len(first) == 1 assert first[0].revision == 1 @@ -578,7 +577,7 @@ def test_duplicate_batch_finished_is_ignored(tmp_path: Path) -> None: shutil.rmtree(tmp_path / "out") gatherer.process_message(batch) - assert drain_queue(gatherer.queue) == [] + assert drain_queue(gatherer.bus.queue) == [] assert gatherer._revision == 1 assert not (tmp_path / "out").exists() @@ -593,7 +592,7 @@ def test_duplicate_is_detected_by_batch_id_not_message_id(tmp_path: Path) -> Non gatherer.process_message(_batch_finished(artifacts, id="msg-a", batch_id="batch-1", batch_jobs=[job])) gatherer.process_message(_batch_finished(artifacts, id="msg-b", batch_id="batch-1", batch_jobs=[job])) - updates = drain_queue(gatherer.queue) + updates = drain_queue(gatherer.bus.queue) assert [update.revision for update in updates] == [1] assert gatherer._revision == 1 assert len(_registry(gatherer)) == 1 @@ -617,7 +616,7 @@ def test_correlates_on_batch_id_not_message_id(tmp_path: Path): ) assert set(gatherer._results_by_batch) == {"batch-09"} - drain_queue(gatherer.queue) + drain_queue(gatherer.bus.queue) [status] = _registry(gatherer) assert status.batch_id == "batch-09" assert status.id == 555 @@ -632,17 +631,17 @@ def test_duplicate_correlates_on_batch_id_across_reruns(tmp_path: Path): gatherer = _make_gatherer(tmp_path, {"batch-09": [make_job("j1")]}) gatherer.process_message(_batch_finished(artifacts, id="msg-a", batch_id="batch-09", run_id=100, batch_jobs=jobs)) - assert len(drain_queue(gatherer.queue)) == 1 + assert len(drain_queue(gatherer.bus.queue)) == 1 gatherer.process_message(_batch_finished(artifacts, id="msg-b", batch_id="batch-09", run_id=200, batch_jobs=jobs)) - assert drain_queue(gatherer.queue) == [] + assert drain_queue(gatherer.bus.queue) == [] assert gatherer._revision == 1 def test_no_emission_without_batch_finished(tmp_path: Path): # Invariant: the gatherer's state changes only when a BatchFinished is consumed. gatherer = _make_gatherer(tmp_path) - assert drain_queue(gatherer.queue) == [] + assert drain_queue(gatherer.bus.queue) == [] assert gatherer._results_by_batch == {} assert gatherer._revision == 0 @@ -696,7 +695,7 @@ def test_finished_batch_leaves_other_batches_planned(tmp_path: Path) -> None: ) ) - update = drain_queue(gatherer.queue)[0] + update = drain_queue(gatherer.bus.queue)[0] assert update.progress.done is False assert _batch_progress(update, "b1").state == ExecutionState.FINISHED assert _batch_progress(update, "b2").state == ExecutionState.PLANNED @@ -724,7 +723,7 @@ def test_progress_and_registry_agree(tmp_path: Path) -> None: ) ) - update = drain_queue(gatherer.queue)[0] + update = drain_queue(gatherer.bus.queue)[0] [workflow] = _registry(gatherer) assert (update.progress.passed, update.progress.failed, update.progress.skipped) == ( workflow.success_count, @@ -750,7 +749,7 @@ def test_batch_status_comes_from_the_workflow_not_from_its_jobs(tmp_path: Path) ) ) - update = drain_queue(gatherer.queue)[0] + update = drain_queue(gatherer.bus.queue)[0] batch = _batch_progress(update, "batch-1") assert batch.status == Status.FAILURE assert [job.latest.status for job in batch.jobs_progress] == [Status.SUCCESS] @@ -775,7 +774,7 @@ def test_unplanned_job_is_warned_about_but_left_out_of_the_totals(tmp_path: Path ) ) - update = drain_queue(gatherer.queue)[0] + update = drain_queue(gatherer.bus.queue)[0] batch = _batch_progress(update, "batch-1") assert [job.job.name for job in batch.jobs_progress] == ["j1"] assert (update.progress.total, update.progress.complete) == (1, 1) @@ -793,7 +792,7 @@ def test_timed_out_batch_is_recorded_on_the_batch(tmp_path: Path) -> None: ) ) - batch = _batch_progress(drain_queue(gatherer.queue)[0], "batch-1") + batch = _batch_progress(drain_queue(gatherer.bus.queue)[0], "batch-1") assert batch.status == Status.FAILURE assert batch.error == ProgressError.TIMED_OUT assert batch.jobs_progress[0].latest is not None @@ -823,7 +822,7 @@ def gather(message) -> None: for future in [pool.submit(gather, message) for message in messages]: future.result() - updates = drain_queue(gatherer.queue) + updates = drain_queue(gatherer.bus.queue) assert sorted(update.revision for update in updates) == [1, 2, 3, 4, 5] assert [update.progress.done for update in updates].count(True) == 1 @@ -838,7 +837,7 @@ def test_missing_artifact_dir_is_recorded_as_an_attempt_error(tmp_path: Path) -> _batch_finished("", batch_jobs=[_batch_job_result(_batch_job("j1"), _workflow_job("j1", "success"), None)]) ) - attempt = _batch_progress(drain_queue(gatherer.queue)[0], "batch-1").jobs_progress[0].latest + attempt = _batch_progress(drain_queue(gatherer.bus.queue)[0], "batch-1").jobs_progress[0].latest assert attempt is not None assert attempt.error == ProgressError.NO_ARTIFACTS assert attempt.reports == () @@ -863,7 +862,7 @@ def test_second_run_appends_an_attempt_and_keeps_untouched_jobs(tmp_path: Path) ], ) ) - drain_queue(gatherer.queue) + drain_queue(gatherer.bus.queue) rerun_dir = _make_job_tree(tmp_path / "artifacts" / "101", "j2") rerun = _batch_finished( @@ -952,7 +951,7 @@ def test_dispatcher_scenario_three_batches(tmp_path: Path) -> None: _scenario_job(a1, "kafka", "success", JUNIT_PASSING, run_id=1), ] gatherer.process_message(_batch_finished(a1, id="b1", run_id=1, batch_jobs=batch_01)) - rev1 = drain_queue(gatherer.queue) + rev1 = drain_queue(gatherer.bus.queue) assert len(rev1) == 1 assert (rev1[0].revision, rev1[0].progress.done) == (1, False) assert _totals(rev1[0]) == (4, 0, 0, 4) @@ -966,7 +965,7 @@ def test_dispatcher_scenario_three_batches(tmp_path: Path) -> None: _scenario_job(a2, "mysql", "failure", JUNIT_FAILING, run_id=2, failed_step="Run unit tests"), ] gatherer.process_message(_batch_finished(a2, id="b2", status="failure", run_id=2, batch_jobs=batch_02)) - rev2 = drain_queue(gatherer.queue) + rev2 = drain_queue(gatherer.bus.queue) assert len(rev2) == 1 assert (rev2[0].revision, rev2[0].progress.done) == (2, False) assert _totals(rev2[0]) == (7, 1, 0, 8) @@ -980,7 +979,7 @@ def test_dispatcher_scenario_three_batches(tmp_path: Path) -> None: _scenario_job(a3, "consul", "skipped", None, run_id=3), ] gatherer.process_message(_batch_finished(a3, id="b3", run_id=3, batch_jobs=batch_03)) - rev3 = drain_queue(gatherer.queue) + rev3 = drain_queue(gatherer.bus.queue) assert len(rev3) == 1 final = rev3[0] assert (final.revision, final.progress.done) == (3, True) @@ -1045,7 +1044,7 @@ def test_dispatcher_scenario_revisions_are_monotonic(tmp_path: Path): artifacts = tmp_path / "artifacts" / str(index) jobs = [_scenario_job(artifacts, f"int{index}", "success", JUNIT_PASSING, run_id=index)] gatherer.process_message(_batch_finished(artifacts, id=f"b{index}", run_id=index, batch_jobs=jobs)) - emitted = drain_queue(gatherer.queue) + emitted = drain_queue(gatherer.bus.queue) assert len(emitted) == 1 revisions.append(emitted[0].revision) diff --git a/ddev/tests/cli/ci/tests/test_task_test_runner.py b/ddev/tests/cli/ci/tests/test_task_test_runner.py index 003bf080a64b6..88b8f5a044fb3 100644 --- a/ddev/tests/cli/ci/tests/test_task_test_runner.py +++ b/ddev/tests/cli/ci/tests/test_task_test_runner.py @@ -5,7 +5,6 @@ from __future__ import annotations -import asyncio import json from pathlib import Path from typing import Any @@ -23,7 +22,7 @@ WorkflowJobsList, WorkflowRun, ) -from tests.cli.ci.tests.helpers import drain_queue, make_job +from tests.cli.ci.tests.helpers import RecordingBus, drain_queue, make_job from tests.helpers.github_async import FakeAsyncGitHubClient # --------------------------------------------------------------------------- @@ -98,7 +97,7 @@ def make_runner(client: FakeAsyncGitHubClient, tmp_path: Path) -> TaskTestRunner client=client, # type: ignore[arg-type] options=options, ) - runner.queue = asyncio.Queue() + runner.bus = RecordingBus() # type: ignore[assignment] return runner @@ -129,7 +128,7 @@ async def run_happy_path(tmp_path: Path) -> tuple[FakeAsyncGitHubClient, BatchFi ) await runner.process_message(batch) - submitted = drain_queue(runner.queue) + submitted = drain_queue(runner.bus.queue) assert len(submitted) == 1 finished = submitted[0] assert isinstance(finished, BatchFinished) @@ -289,7 +288,7 @@ async def test_uses_batch_id_not_message_id_for_correlation(tmp_path: Path): assert fake.calls_to("create_check_run")[0].kwargs["name"] == "test-batch/batch-07" assert fake.calls_to("create_workflow_dispatch")[0].kwargs["inputs"]["batch_id"] == "batch-07" - finished = drain_queue(runner.queue)[0] + finished = drain_queue(runner.bus.queue)[0] assert isinstance(finished, BatchFinished) assert finished.id == "msg-uuid-xyz" assert finished.batch_id == "batch-07" @@ -328,7 +327,7 @@ async def test_process_message_correlates_batch_jobs(tmp_path: Path): TestBatch(id="batch-c", batch_id="batch-c", job_list=[j1, j2], jobs_count=2, integrations=["ntp"]) ) - finished = drain_queue(runner.queue)[0] + finished = drain_queue(runner.bus.queue)[0] assert isinstance(finished, BatchFinished) assert finished.status == "failure" @@ -360,7 +359,7 @@ async def test_process_message_batch_job_without_workflow_match(tmp_path: Path): TestBatch(id="batch-d", batch_id="batch-d", job_list=[job], jobs_count=1, integrations=["ntp"]) ) - finished = drain_queue(runner.queue)[0] + finished = drain_queue(runner.bus.queue)[0] assert isinstance(finished, BatchFinished) [result] = finished.batch_jobs assert result.job == job @@ -382,7 +381,7 @@ async def test_process_message_batch_job_without_artifacts(tmp_path: Path): TestBatch(id="batch-e", batch_id="batch-e", job_list=[job], jobs_count=1, integrations=["ntp"]) ) - finished = drain_queue(runner.queue)[0] + finished = drain_queue(runner.bus.queue)[0] assert isinstance(finished, BatchFinished) [result] = finished.batch_jobs assert result.workflow_job is not None and result.workflow_job.conclusion == "success" @@ -406,7 +405,7 @@ async def test_process_message_emits_batch_finished_when_listing_jobs_fails(tmp_ # correlated job carrying no workflow job. await runner.process_message(make_batch()) - finished = drain_queue(runner.queue)[0] + finished = drain_queue(runner.bus.queue)[0] assert isinstance(finished, BatchFinished) assert finished.status == "success" assert all(result.workflow_job is None for result in finished.batch_jobs) @@ -423,7 +422,7 @@ async def test_process_message_failure_path(tmp_path: Path): TestBatch(id="batch-2", batch_id="batch-2", job_list=[make_job()], jobs_count=1, integrations=["ntp"]) ) - submitted = drain_queue(runner.queue) + submitted = drain_queue(runner.bus.queue) assert len(submitted) == 1 finished = submitted[0] assert isinstance(finished, BatchFinished) @@ -445,7 +444,7 @@ async def test_process_message_skipped_conclusion(tmp_path: Path): await runner.process_message(make_batch()) # A "skipped" GitHub conclusion maps to a "skipped" BatchFinished and a "skipped" check run. - submitted = drain_queue(runner.queue) + submitted = drain_queue(runner.bus.queue) assert len(submitted) == 1 finished = submitted[0] assert isinstance(finished, BatchFinished) @@ -470,7 +469,7 @@ async def test_process_message_polls_until_completed(tmp_path: Path): ) assert len(fake.calls_to("get_workflow_run")) == 4 - submitted = drain_queue(runner.queue) + submitted = drain_queue(runner.bus.queue) assert len(submitted) == 1 assert isinstance(submitted[0], BatchFinished) assert submitted[0].status == "success" @@ -510,7 +509,7 @@ async def test_process_message_null_conclusion(tmp_path: Path): await runner.process_message(make_batch()) # A null GitHub conclusion maps to a "failure" BatchFinished and a "neutral" check run. - submitted = drain_queue(runner.queue) + submitted = drain_queue(runner.bus.queue) assert len(submitted) == 1 finished = submitted[0] assert isinstance(finished, BatchFinished) @@ -536,7 +535,7 @@ async def test_process_message_emits_batch_finished_when_listing_artifacts_fails assert len(update_calls) == 1 assert update_calls[0].kwargs["conclusion"] == "success" - submitted = drain_queue(runner.queue) + submitted = drain_queue(runner.bus.queue) assert len(submitted) == 1 finished = submitted[0] assert isinstance(finished, BatchFinished) @@ -555,7 +554,7 @@ async def test_process_message_swallows_check_run_close_failure(tmp_path: Path): await runner.process_message(make_batch()) assert len(fake.calls_to("update_check_run")) == 1 - submitted = drain_queue(runner.queue) + submitted = drain_queue(runner.bus.queue) assert len(submitted) == 1 assert isinstance(submitted[0], BatchFinished) assert submitted[0].status == "success" @@ -582,7 +581,7 @@ async def test_download_failure_for_one_artifact_does_not_abort_others(tmp_path: "https://api.github.com/artifact/2/zip", "https://api.github.com/artifact/3/zip", ] - submitted = drain_queue(runner.queue) + submitted = drain_queue(runner.bus.queue) assert len(submitted) == 1 assert isinstance(submitted[0], BatchFinished) assert submitted[0].status == "success" @@ -657,11 +656,11 @@ async def test_failure_at_submit_message_closes_check_run_as_success(tmp_path: P mock_artifacts(fake, [make_artifact(1)]) runner = make_runner(fake, tmp_path) - class _BoomQueue: - def put_nowait(self, _: Any): + class _BoomBus: + def submit_message(self, _: Any): raise boom - runner.queue = _BoomQueue() # type: ignore[assignment] + runner.bus = _BoomBus() # type: ignore[assignment] with pytest.raises(RuntimeError, match="boom-submit-message"): await runner.process_message(make_batch()) diff --git a/ddev/tests/event_bus/test_event_bus.py b/ddev/tests/event_bus/test_event_bus.py index 77545e12b665d..e61e880a84fc4 100644 --- a/ddev/tests/event_bus/test_event_bus.py +++ b/ddev/tests/event_bus/test_event_bus.py @@ -745,6 +745,20 @@ def process_message(self, message: Announcement): assert secretary.delivered_memos[0].id == "async_memo" +class AsyncDelegator(AsyncProcessor[Memo]): + """Submits a follow-up from the loop thread.""" + + async def process_message(self, message: Memo): + self.submit_message(Announcement(id="delegated", announcement_type="Delegated")) + + +class SyncDelegator(SyncProcessor[Memo]): + """Submits a follow-up from an executor thread.""" + + def process_message(self, message: Memo): + self.submit_message(Announcement(id="delegated", announcement_type="Delegated")) + + def test_a_worker_thread_submission_is_delivered(analyst: Analyst): """A `SyncProcessor` submits from an executor thread, where `asyncio.Queue` can lose the put. @@ -764,15 +778,11 @@ def put_nowait(self, item): return super().put_nowait(item) - class Delegator(SyncProcessor[Memo]): - def process_message(self, message: Memo): - self.submit_message(Announcement(id="delegated", announcement_type="FromWorkerThread")) - logger = logging.getLogger("test_thread_safe_submit") orchestrator = MockOrchestrator(logger, max_timeout=2, grace_period=0.1) - # `asyncio.run` runs the loop here, and registration is what hands the queue to each processor. + # `asyncio.run` runs the loop on this thread, so this is the id the queue lets a put through on. orchestrator._queue = LoseOffLoopPuts(threading.get_ident()) - orchestrator.register_processor(Delegator("delegator"), [Memo]) + orchestrator.register_processor(SyncDelegator("delegator"), [Memo]) orchestrator.register_processor(analyst, [Announcement]) orchestrator.submit_message(Memo("delegate_me")) @@ -782,6 +792,52 @@ def process_message(self, message: Memo): assert [message.id for message in analyst.completed_tasks] == ["delegated"] +@pytest.mark.parametrize("delegator_class", [AsyncDelegator, SyncDelegator], ids=["async", "sync"]) +def test_a_processor_submission_survives_a_zero_grace_period( + analyst: Analyst, delegator_class: type[AsyncDelegator | SyncDelegator] +): + """A follow-up is queued before the bus can decide it has nothing left to do. + + Every put is now handed to the loop thread, so it happens after `submit_message` returns, and a + zero grace period stops the bus the moment the queue looks empty. Both kinds of processor are + covered because the callback that queues the message precedes the task completion that wakes the + bus by a different route for each. + """ + logger = logging.getLogger("test_zero_grace_period") + orchestrator = MockOrchestrator(logger, max_timeout=2, grace_period=0) + orchestrator.register_processor(delegator_class("delegator"), [Memo]) + orchestrator.register_processor(analyst, [Announcement]) + + orchestrator.submit_message(Memo("delegate_me")) + + orchestrator.run() + + assert [message.id for message in analyst.completed_tasks] == ["delegated"] + + +def test_an_on_initialize_submission_survives_a_zero_grace_period(secretary: Secretary): + """The bus reads what `on_initialize` submitted, with no grace period to fall back on. + + Nothing awaits between the hook and the bus's first look at the queue, so a put deferred to a loop + callback would not have happened yet, and a zero grace period stops instead of waiting for it. + """ + logger = logging.getLogger("test_initialize_submit") + orchestrator = MockOrchestrator(logger, max_timeout=2, grace_period=0) + orchestrator.register_processor(secretary, [Memo]) + + original_on_initialize = orchestrator.on_initialize + + async def on_initialize_submitting(): + await original_on_initialize() + orchestrator.submit_message(Memo("from_initialize")) + + orchestrator.on_initialize = on_initialize_submitting # type: ignore[method-assign] + + orchestrator.run() + + assert [message.id for message in secretary.delivered_memos] == ["from_initialize"] + + def test_processor_submit_without_bus(): processor = Secretary("orphan") with pytest.raises(ProcessorQueueError, match="This processor has not been added"): From 298f10807de54a40428008abd2a7e65e99bc293c Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Tue, 25 Aug 2026 12:07:11 +0100 Subject: [PATCH 06/13] Trim the submit_message docstring Co-Authored-By: Claude Opus 5 (1M context) --- ddev/src/ddev/event_bus/orchestrator.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/ddev/src/ddev/event_bus/orchestrator.py b/ddev/src/ddev/event_bus/orchestrator.py index acac1d5a489d3..e553a8687aaae 100644 --- a/ddev/src/ddev/event_bus/orchestrator.py +++ b/ddev/src/ddev/event_bus/orchestrator.py @@ -165,11 +165,8 @@ def register_processor[T: BaseMessage](self, processor: Processor[T], message_ty def submit_message(self, message: BaseMessage): """Adds a message to the queue, from any thread. - A ``SyncProcessor`` runs in an executor thread, and ``asyncio.Queue`` is not thread-safe: a - put landing between ``get``'s ``empty()`` check and its waiter being registered wakes nobody, - leaving the bus spinning on a message it never reads. So while the bus runs, every put is - handed to the loop thread, whichever thread asked for it. Before it starts and after it - stops there is no loop to hand it to, and the queue takes the message directly. + ``asyncio.Queue`` is not thread-safe, and a put it loses leaves the bus spinning on a message + it never reads, so while the bus runs the loop thread makes every put. """ if self._loop is not None and self._loop.is_running(): self._loop.call_soon_threadsafe(self._queue.put_nowait, message) From 6232d536e4a850e8171a1e1bb1fc99a90ead1988 Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Thu, 20 Aug 2026 16:40:35 +0100 Subject: [PATCH 07/13] Wire the Dispatcher together and give it an entry point Adds `Dispatcher`, an `EventBusOrchestrator` that owns a batching plan and registers the runner, gatherer and pull-request updater on one bus, and the `ddev ci dispatch-tests` command that builds and runs it. Every input can be passed explicitly, which is how a workflow calls it. Locally, `--pr` takes a number or a URL and reads the branch, commits and target branch from the GitHub API, and `--dry-run` shows the plan without touching GitHub. Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 3 + ddev/src/ddev/cli/ci/__init__.py | 2 + ddev/src/ddev/cli/ci/dispatch_tests.py | 275 ++++++++++++++++++ ddev/src/ddev/cli/ci/tests/batching/AGENTS.md | 2 +- .../src/ddev/cli/ci/tests/batching/targets.py | 17 ++ ddev/src/ddev/cli/ci/tests/dispatcher.py | 204 +++++++++++++ .../ddev/cli/ci/tests/dispatcher_config.py | 6 + .../ddev/cli/ci/tests/task_test_gatherer.py | 12 +- ddev/src/ddev/utils/github.py | 18 ++ ddev/tests/cli/ci/test_dispatch_tests.py | 85 ++++++ .../cli/ci/tests/batching/test_targets.py | 6 + ddev/tests/cli/ci/tests/test_dispatcher.py | 180 ++++++++++++ 12 files changed, 808 insertions(+), 2 deletions(-) create mode 100644 ddev/src/ddev/cli/ci/dispatch_tests.py create mode 100644 ddev/src/ddev/cli/ci/tests/dispatcher.py create mode 100644 ddev/tests/cli/ci/test_dispatch_tests.py create mode 100644 ddev/tests/cli/ci/tests/test_dispatcher.py diff --git a/.gitignore b/.gitignore index 807fbcbe3f380..6b8bf02cda95e 100644 --- a/.gitignore +++ b/.gitignore @@ -157,3 +157,6 @@ datadog_checks_downloader/datadog_checks/downloader/data/repo/targets/* datadog_checks_downloader/datadog_checks/downloader/data/repo/metadata/* !datadog_checks_downloader/datadog_checks/downloader/data/repo/metadata/.gitignore !datadog_checks_downloader/datadog_checks/downloader/data/repo/metadata/root.json + +# Artifacts and results a local `ddev ci dispatch-tests` run writes +/.dispatcher/ diff --git a/ddev/src/ddev/cli/ci/__init__.py b/ddev/src/ddev/cli/ci/__init__.py index 2870228466386..097cc53af3dc3 100644 --- a/ddev/src/ddev/cli/ci/__init__.py +++ b/ddev/src/ddev/cli/ci/__init__.py @@ -5,6 +5,7 @@ from datadog_checks.dev.tooling.commands.ci.setup import setup from ddev.cli.ci.codeowners import codeowners +from ddev.cli.ci.dispatch_tests import dispatch_tests @click.group(short_help='Collection of CI utilities') @@ -17,3 +18,4 @@ def ci(): ci.add_command(setup) ci.add_command(codeowners) +ci.add_command(dispatch_tests) diff --git a/ddev/src/ddev/cli/ci/dispatch_tests.py b/ddev/src/ddev/cli/ci/dispatch_tests.py new file mode 100644 index 0000000000000..3836a8678bdc7 --- /dev/null +++ b/ddev/src/ddev/cli/ci/dispatch_tests.py @@ -0,0 +1,275 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +"""The `ddev ci dispatch-tests` command: the Dispatcher's entry point.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import click + +if TYPE_CHECKING: + from ddev.cli.application import Application + from ddev.cli.ci.tests.batching.units import EnvironmentProvider + from ddev.cli.ci.tests.dispatcher import DispatcherContext, RunContext + from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig + from ddev.cli.ci.tests.messages import TestBatch + from ddev.utils.github_async.models import PullRequest + +DEFAULT_OUTPUT_DIRECTORY = ".dispatcher" +MAX_LISTED_INTEGRATIONS = 10 + + +@click.command(short_help='Run the Dispatcher to test a commit as parallel batches') +@click.pass_obj +@click.option( + '--pr', + 'pull_request', + metavar='PR_NUMBER_OR_URL', + default=None, + help='Pull request to test, as a number or a URL. Its branch, commits and target branch are read from GitHub.', +) +@click.option('--pr-number', type=int, default=None, help='Pull request number, when not using `--pr`.') +@click.option('--checkout-sha', default=None, help='Ref the test workflow checks out. Defaults to the base commit.') +@click.option('--base-sha', default=None, help='Commit the run reports against. Defaults to the local HEAD.') +@click.option('--branch', default=None, help='Branch being tested. Defaults to the current branch.') +@click.option('--target-branch', default=None, help='Target branch of the pull request, used as the comparison base.') +@click.option( + '--context', + 'run_context', + type=click.Choice(['pr', 'master', 'agent-test', 'release']), + default=None, + help='Kind of run. Defaults to `pr` when a pull request is known, `master` otherwise.', +) +@click.option('--repo', 'repository', default=None, metavar='OWNER/NAME', help='Repository to dispatch against.') +@click.option('--all', 'all_targets', is_flag=True, help='Test every eligible target instead of the affected ones.') +@click.option('--workflow', default=None, help='Workflow each batch is dispatched to.') +@click.option('--workflow-ref', default=None, help='Ref the workflow definition is loaded from.') +@click.option('--artifacts-dir', default=None, help='Where downloaded artifacts are written.') +@click.option('--output-dir', default=None, help='Where coverage and test results are organized.') +@click.option('--dry-run', is_flag=True, help='Show the plan and the resolved context without calling GitHub.') +def dispatch_tests( + app: Application, + pull_request: str | None, + pr_number: int | None, + checkout_sha: str | None, + base_sha: str | None, + branch: str | None, + target_branch: str | None, + run_context: str | None, + repository: str | None, + all_targets: bool, + workflow: str | None, + workflow_ref: str | None, + artifacts_dir: str | None, + output_dir: str | None, + dry_run: bool, +) -> None: + """Plan the tests a commit requires, run them as parallel batches of GitHub Actions jobs, and + report the result to the pull request and to the run summary. + + Every input can be passed explicitly, which is how a workflow calls it. Locally, `--pr` reads + the branch, commits and target branch from GitHub so only the pull request has to be named. + """ + import logging + from pathlib import Path + + from ddev.cli.ci.tests.batching.build import HatchEnvironmentProvider + from ddev.cli.ci.tests.dispatcher import DispatcherContext, RunContext, build_dispatcher + from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig + + # One INFO line per request would bury the Dispatcher's own progress. + logging.getLogger('httpx').setLevel(logging.WARNING) + + config = DispatcherConfig.from_repo_config(app.repo.config) + owner, repo = resolve_owner_repo(app, repository) + + resolved_number = resolved_branch = resolved_sha = resolved_target = None + if pull_request is not None: + resolved = fetch_pull_request(app, owner, repo, pull_request) + if resolved.head is None or resolved.base is None: + app.abort(f'Pull request {resolved.number} reports no branch references.') + resolved_number = resolved.number + resolved_branch, resolved_sha, resolved_target = resolved.head.ref, resolved.head.sha, resolved.base.ref + + pr_number = pr_number if pr_number is not None else resolved_number + branch = branch or resolved_branch or app.repo.git.current_branch() + base_sha = base_sha or resolved_sha or app.repo.git.latest_commit().sha + target_branch = target_branch or resolved_target + checkout_sha = checkout_sha or (f'refs/pull/{pr_number}/merge' if pr_number is not None else base_sha) + resolved_context = RunContext(run_context) if run_context else (RunContext.PR if pr_number else RunContext.MASTER) + + batches = build_plan( + app, + config=config, + base_sha=base_sha, + run_context=resolved_context, + target_branch=target_branch, + all_targets=all_targets, + environment_provider=HatchEnvironmentProvider(app.platform, config.default_python_version), + ) + if not batches: + app.display_info('No affected target to test.') + return + + context = DispatcherContext( + owner=owner, + repo=repo, + run_context=resolved_context, + checkout_sha=checkout_sha, + base_sha=base_sha, + branch=branch, + workflow=workflow or config.workflow, + workflow_ref=workflow_ref or config.workflow_ref, + target_branch=target_branch, + pr_number=pr_number, + ) + + display_plan(app, context, batches) + if dry_run: + app.display_info('Dry run: nothing was dispatched.') + return + + token = app.config.github.token + if not token: + app.abort('A GitHub token is required. Set `github.token` in your ddev config.') + + base_path = Path(output_dir) if output_dir else app.repo.path / DEFAULT_OUTPUT_DIRECTORY + dispatcher = build_dispatcher( + batches=batches, + context=context, + config=config, + token=token, + artifacts_path=Path(artifacts_dir) if artifacts_dir else base_path / 'artifacts', + output_path=base_path / 'results', + run_logger=app.logger, + ) + dispatcher.run() + + outcome = dispatcher.outcome + if outcome is None or not outcome.successful: + app.abort('Dispatcher tests failed.') + + app.display_success('Dispatcher tests passed.') + + +def resolve_owner_repo(app: Application, repository: str | None) -> tuple[str, str]: + """Split `owner/name`, defaulting to the active repository and the `DataDog` organization.""" + full_name = repository or app.repo.full_name + owner, separator, name = full_name.partition('/') + if not separator: + return 'DataDog', full_name + return owner, name + + +def fetch_pull_request(app: Application, owner: str, repo: str, reference: str) -> PullRequest: + """Read the pull request named by *reference* (a number or a URL) from the GitHub API.""" + import asyncio + + import httpx + from pydantic import ValidationError + + from ddev.utils.github import parse_pull_request_reference + from ddev.utils.github_async import async_github_client + from ddev.utils.github_errors import GitHubAuthenticationError + + number = parse_pull_request_reference(reference) + if number is None: + app.abort(f'`{reference}` is neither a pull request number nor a pull request URL.') + + token = app.config.github.token + if not token: + app.abort('A GitHub token is required to read a pull request. Set `github.token` in your ddev config.') + + async def fetch() -> PullRequest: + async with async_github_client(token=token) as client: + response = await client.get_pull_request(owner, repo, number) + return response.data + + try: + return asyncio.run(fetch()) + except GitHubAuthenticationError as error: + app.abort(str(error)) + except (httpx.HTTPError, ValidationError) as error: + app.abort(f'Could not read pull request {number}: {error}') + + +def build_plan( + app: Application, + *, + config: DispatcherConfig, + base_sha: str, + run_context: RunContext, + target_branch: str | None, + all_targets: bool, + environment_provider: EnvironmentProvider, +) -> list[TestBatch]: + """Build the batches this run must execute, aborting with a readable message on a bad plan. + + `--all` skips the comparison entirely: what changed is not what decides which targets run. + """ + from ddev.cli.ci.tests.batching.build import build_test_batches + from ddev.cli.ci.tests.batching.exceptions import PlanningError + from ddev.cli.ci.tests.batching.targets import all_target_rules + from ddev.cli.ci.tests.changes import CIContext, get_changed_files + from ddev.cli.ci.tests.dispatcher import RunContext + + changed_files = [] + rules = None + if all_targets: + rules = all_target_rules() + else: + ci_context = CIContext.PULL_REQUEST if run_context is RunContext.PR else CIContext.DEFAULT_BRANCH + try: + changed_files = get_changed_files(app.repo.git, base_sha, context=ci_context, target_branch=target_branch) + except ValueError as error: + app.abort(str(error)) + except OSError as error: + # `GitRepository` reports a failed git invocation as OSError, and the usual cause is a + # commit the local clone has never fetched. + app.abort( + f'Could not compare {base_sha} against {target_branch or "its parent"} locally: {error}\n' + 'Fetch the commit first, or pass `--all` to plan every target.' + ) + + try: + batches = build_test_batches( + app.repo, + changed_files, + environment_provider=environment_provider, + config=config.batching, + rules=rules, + ) + except PlanningError as error: + app.abort(f'Could not build a test plan: {error}') + + return batches + + +def display_plan(app: Application, context: DispatcherContext, batches: list[TestBatch]) -> None: + app.display_header('Dispatcher plan') + app.display_pair('Repository', f'{context.owner}/{context.repo}') + app.display_pair('Context', context.run_context.value) + app.display_pair('Branch', context.branch) + app.display_pair('Base commit', context.base_sha) + app.display_pair('Checkout ref', context.checkout_sha) + if context.pr_number is not None: + app.display_pair('Pull request', str(context.pr_number)) + if context.target_branch is not None: + app.display_pair('Target branch', context.target_branch) + app.display_pair('Workflow', f'{context.workflow} @ {context.workflow_ref}') + + total = sum(batch.jobs_count for batch in batches) + app.display_pair('Batches', f'{len(batches)} ({total} jobs)') + for batch in batches: + count = len(batch.integrations) + app.display(f' {batch.batch_id}: {batch.jobs_count} jobs, {count} integration{"" if count == 1 else "s"}') + # A repository-wide run names every integration, which is hundreds of lines of no use here. + app.display(f' {summarize(batch.integrations)}') + + +def summarize(names: list[str], limit: int = MAX_LISTED_INTEGRATIONS) -> str: + if len(names) <= limit: + return ', '.join(names) + return f'{", ".join(names[:limit])}, and {len(names) - limit} more' diff --git a/ddev/src/ddev/cli/ci/tests/batching/AGENTS.md b/ddev/src/ddev/cli/ci/tests/batching/AGENTS.md index 052b34cc86b63..60277f27f9776 100644 --- a/ddev/src/ddev/cli/ci/tests/batching/AGENTS.md +++ b/ddev/src/ddev/cli/ci/tests/batching/AGENTS.md @@ -22,7 +22,7 @@ them from git, and `../changes.py` decides which two commits a CI run compares. | Module | Role | | --- | --- | | `build.py` | Composes the stages and adapts concrete `Repository`/`Integration` objects to them. The package's public entry point. | -| `targets.py` | Maps changed files to affected target names through ordered, independent rules. | +| `targets.py` | Maps changed files to affected target names through ordered, independent rules. `AllTargetsRule` is the exception: it ignores the change set, for a run that tests everything. | | `units.py` | Expands targets into `TestUnit` values: one target, one platform, one environment. | | `jobs.py` | Turns each unit into the concrete `BatchJob` the workflow runs. | | `strategy/` | Packs jobs into capacity-bounded groups. `types.py` is the contract, `default.py` the implementation. | diff --git a/ddev/src/ddev/cli/ci/tests/batching/targets.py b/ddev/src/ddev/cli/ci/tests/batching/targets.py index 67761d8326de3..d89c50b6c52ce 100644 --- a/ddev/src/ddev/cli/ci/tests/batching/targets.py +++ b/ddev/src/ddev/cli/ci/tests/batching/targets.py @@ -151,6 +151,23 @@ def __call__(self, changed_files: Sequence[ChangedFile], facts: RepositoryFacts) yield from facts.eligible_targets() +@dataclass(frozen=True) +class AllTargetsRule: + """Select every eligible target, whatever changed. + + Used by a run that tests the whole repository on purpose — a push to the default branch, the + nightly schedule, the Agent test workflow — where the change set is not what decides. + """ + + def __call__(self, changed_files: Sequence[ChangedFile], facts: RepositoryFacts) -> Iterator[str]: + yield from facts.eligible_targets() + + +def all_target_rules() -> tuple[TargetRule, ...]: + """The rule set that tests every eligible target.""" + return (AllTargetsRule(),) + + def default_target_rules(*, is_core: bool) -> tuple[TargetRule, ...]: """Build the default ordered rule set for a repository.""" return (DirectTargetRule(), RepositoryWideRule(is_core=is_core)) diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher.py b/ddev/src/ddev/cli/ci/tests/dispatcher.py new file mode 100644 index 0000000000000..c01acc939a00e --- /dev/null +++ b/ddev/src/ddev/cli/ci/tests/dispatcher.py @@ -0,0 +1,204 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +"""The Dispatcher: the event bus that runs a batching plan and reports the result.""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from enum import StrEnum +from typing import TYPE_CHECKING + +from ddev.cli.ci.tests.messages import BatchFinished, TestBatch, UpdatePRComment +from ddev.cli.ci.tests.pr_comment import render_run_summary, summary_line +from ddev.cli.ci.tests.rate_limiting import RateLimiterFactory +from ddev.cli.ci.tests.status import Status +from ddev.cli.ci.tests.task_pull_request_updater import PullRequestUpdaterOptions, TaskPullRequestUpdater +from ddev.cli.ci.tests.task_test_gatherer import TaskTestGatherer +from ddev.cli.ci.tests.task_test_runner import TaskTestRunner, TestRunnerOptions +from ddev.event_bus.orchestrator import BaseMessage, EventBusOrchestrator +from ddev.utils.github_actions import write_step_summary + +if TYPE_CHECKING: + from pathlib import Path + + from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig + from ddev.cli.ci.tests.progress import DispatcherProgress + from ddev.utils.github_async import AsyncGitHubClient + +INITIAL_UPDATE_MESSAGE_ID = "dispatcher-initial" +# A batch takes minutes, so the bus is idle between results far longer than the bus default +# allows. It only has to outlast the gap between a task finishing and its message being read. +DEFAULT_GRACE_PERIOD = 30.0 + +logger = logging.getLogger(__name__) + + +class RunContext(StrEnum): + """What kind of run the Dispatcher is testing, reported as a monitoring tag.""" + + PR = "pr" + MASTER = "master" + AGENT_TEST = "agent-test" + RELEASE = "release" + + +@dataclass(frozen=True) +class DispatcherContext: + """Everything the Dispatcher needs to know about the run it is testing. + + `base_sha` and `checkout_sha` are deliberately separate: a pull request is tested at the merge + commit (`refs/pull//merge`) but its checks and metrics belong to the head commit. Outside a + pull request the two are the same. + """ + + owner: str + repo: str + run_context: RunContext + checkout_sha: str + base_sha: str + branch: str + workflow: str + workflow_ref: str + target_branch: str | None = None + pr_number: int | None = None + + +@dataclass(frozen=True) +class DispatcherOutcome: + """What a finished Dispatcher execution amounts to, for the caller to exit on.""" + + progress: DispatcherProgress + pr_comment_failed: bool + error: Exception | None = None + + @property + def successful(self) -> bool: + """Whether every batch reached a non-failing terminal state and the report was published. + + A batch that never finished counts as a failure: `progress.done` is false, and a run whose + results are unknown must not read as green. + """ + return ( + self.error is None + and not self.pr_comment_failed + and self.progress.done + and all(batch.status is not Status.FAILURE for batch in self.progress.batches) + ) + + +class Dispatcher(EventBusOrchestrator): + """Runs a batching plan to completion and publishes its result. + + The whole plan is known before the bus starts, so `on_initialize` primes the queue with the + initial pull-request update and every batch, and the tasks carry it from there: + `TestBatch` -> runner -> `BatchFinished` -> gatherer -> `UpdatePRComment` -> updater. The bus + stops when the queue drains and no task is left running. + """ + + def __init__( + self, + *, + batches: list[TestBatch], + client: AsyncGitHubClient, + runner: TaskTestRunner, + gatherer: TaskTestGatherer, + updater: TaskPullRequestUpdater, + max_timeout: float | None, + grace_period: float = DEFAULT_GRACE_PERIOD, + run_logger: logging.Logger | None = None, + ): + super().__init__(run_logger or logger, max_timeout=max_timeout, grace_period=grace_period) + self._batches = batches + self._client = client + self._gatherer = gatherer + self._updater = updater + self._outcome: DispatcherOutcome | None = None + + self.register_processor(runner, [TestBatch]) + self.register_processor(gatherer, [BatchFinished]) + self.register_processor(updater, [UpdatePRComment]) + + @property + def outcome(self) -> DispatcherOutcome | None: + """The result of the execution, or None before `run` has finished.""" + return self._outcome + + async def on_initialize(self): + self.submit_message(self._gatherer.build_initial_update(INITIAL_UPDATE_MESSAGE_ID)) + for batch in self._batches: + self.submit_message(batch) + self._logger.info("Dispatched %s batches", len(self._batches)) + + async def on_message_received(self, message: BaseMessage): + self._logger.debug("Message received: %s(%s)", type(message).__name__, message.id) + + async def on_finalize(self, exception: Exception | None): + try: + progress = self._gatherer.progress + self._outcome = DispatcherOutcome( + progress=progress, + pr_comment_failed=self._updater.pr_comment_failed, + error=exception, + ) + self._logger.info(summary_line(progress)) + if (body := self._updater.latest_body) is not None: + write_step_summary(render_run_summary(body, pr_comment_failed=self._updater.pr_comment_failed)) + finally: + await self._client.aclose() + + +def build_dispatcher( + *, + batches: list[TestBatch], + context: DispatcherContext, + config: DispatcherConfig, + token: str, + artifacts_path: Path, + output_path: Path, + run_logger: logging.Logger | None = None, +) -> Dispatcher: + """Assemble the client, the three tasks and the Dispatcher from a plan and its run context. + + One client and one rate limiter are shared by every task, so the run's request rate is bounded + as a whole rather than per task. The limiter tier is chosen from every integration in the plan, + which is the slowest thing the run will wait on. + """ + from ddev.utils.github_async import AsyncGitHubClient + + active_logger = run_logger or logger + integrations = frozenset(integration for batch in batches for integration in batch.integrations) + rate_limiter = RateLimiterFactory(config.github_rate_limits, active_logger).get_limiter(integrations) + client = AsyncGitHubClient(token, rate_limiter=rate_limiter) + + runner = TaskTestRunner( + "test-runner", + client, + TestRunnerOptions( + owner=context.owner, + repo=context.repo, + workflow_id=context.workflow, + ref=context.workflow_ref, + base_sha=context.base_sha, + checkout_sha=context.checkout_sha, + artifacts_base_path=artifacts_path, + poll_interval_seconds=config.poll_interval_seconds, + ), + ) + gatherer = TaskTestGatherer("test-gatherer", output_path, batches) + updater = TaskPullRequestUpdater( + "pull-request-updater", + client, + PullRequestUpdaterOptions(owner=context.owner, repo=context.repo, pr_number=context.pr_number), + ) + + return Dispatcher( + batches=batches, + client=client, + runner=runner, + gatherer=gatherer, + updater=updater, + max_timeout=config.global_timeout_seconds, + run_logger=active_logger, + ) diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher_config.py b/ddev/src/ddev/cli/ci/tests/dispatcher_config.py index fc9d45cf33f10..3eb69e5906627 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher_config.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher_config.py @@ -34,6 +34,12 @@ class DispatcherConfig(BaseModel): global_timeout_seconds: float = Field(default=10800.0, gt=0) # 3 hours # Used when Hatch does not declare a Python version. default_python_version: str = Field(default="3.13", pattern=r"^\d+\.\d+$") + # The workflow each batch is dispatched to, by file name or numeric id. + workflow: str = "test-batch.yml" + # The ref the workflow definition is loaded from. Never a pull-request ref: the definition must + # come from a reviewed branch even when the code under test does not. + workflow_ref: str = "master" + poll_interval_seconds: float = Field(default=30.0, gt=0) batching: BatchingConfig = BatchingConfig() github_rate_limits: RateLimiterFactoryConfig = RateLimiterFactoryConfig() diff --git a/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py b/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py index 35bfcc7044ad4..54e566ca8f0df 100644 --- a/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py +++ b/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py @@ -96,7 +96,7 @@ def process_message(self, message: BatchFinished) -> None: self._progress_by_batch[message.batch_id] = self._finished_batch_progress(planned, message, gathered) self._revision += 1 revision = self._revision - done = all(batch.state is ExecutionState.FINISHED for batch in self._progress_by_batch.values()) + done = self._done() self.submit_message(self.build_update_message(message.id, revision, done)) self._logger.info( @@ -117,6 +117,16 @@ def _accepts(self, batch_id: str, log_extra: dict[str, Any]) -> bool: return False return True + @property + def progress(self) -> DispatcherProgress: + """The current aggregate snapshot, for a caller outside the message flow.""" + with self._lock: + return DispatcherProgress(batches=tuple(self._progress_by_batch.values()), done=self._done()) + + def _done(self) -> bool: + """Whether every batch is terminal. Hold ``self._lock``.""" + return all(batch.state is ExecutionState.FINISHED for batch in self._progress_by_batch.values()) + def build_initial_update(self, message_id: str) -> UpdatePRComment: """Revision ``0``: the complete plan, before any batch has been dispatched. diff --git a/ddev/src/ddev/utils/github.py b/ddev/src/ddev/utils/github.py index 615b2265158e4..a3c4c8df24a05 100644 --- a/ddev/src/ddev/utils/github.py +++ b/ddev/src/ddev/utils/github.py @@ -4,6 +4,7 @@ from __future__ import annotations import json +import re from functools import cached_property from time import time from typing import TYPE_CHECKING, overload @@ -17,6 +18,9 @@ MAX_SECONDARY_RATE_LIMIT_RETRIES = 2 MAX_SECONDARY_RATE_LIMIT_WAIT_SECONDS = 3600 +PULL_REQUEST_NUMBER_PATTERN = re.compile(r'^\d+$') +PULL_REQUEST_URL_PATTERN = re.compile(r'^https?://github\.com/[^/]+/[^/]+/pull/(\d+)(?:[/?#].*)?$', re.IGNORECASE) + if TYPE_CHECKING: from typing import Any, Literal @@ -26,6 +30,20 @@ from ddev.repo.core import Repository +def parse_pull_request_reference(value: str) -> int | None: + """Return the pull-request number in *value*, or None when it is neither shape. + + Accepts a bare number or a GitHub pull-request URL, so a command can take whichever one the + user has at hand. + """ + reference = value.strip() + if PULL_REQUEST_NUMBER_PATTERN.match(reference): + return int(reference) + + match = PULL_REQUEST_URL_PATTERN.match(reference) + return int(match.group(1)) if match else None + + class PullRequest: def __init__(self, data: dict[str, Any]): self.__number = data['number'] diff --git a/ddev/tests/cli/ci/test_dispatch_tests.py b/ddev/tests/cli/ci/test_dispatch_tests.py new file mode 100644 index 0000000000000..fe514ca21c7c5 --- /dev/null +++ b/ddev/tests/cli/ci/test_dispatch_tests.py @@ -0,0 +1,85 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +"""Tests for `ddev ci dispatch-tests`: how it resolves the run it is asked to test.""" + +from __future__ import annotations + +import pytest + +from ddev.cli.ci.tests.messages import TestBatch +from ddev.utils.github_async.models import PullRequest +from tests.cli.ci.tests.helpers import make_job + +PR_NUMBER = 4242 +PR_PAYLOAD = PullRequest( + number=PR_NUMBER, + html_url=f'https://github.com/DataDog/integrations-core/pull/{PR_NUMBER}', + head={'ref': 'hs/a-branch', 'sha': 'head-sha-aaa'}, + base={'ref': 'a-target-branch', 'sha': 'base-sha-bbb'}, +) + + +@pytest.fixture +def planned(mocker): + """Stand in for the planning layer, which has its own tests and costs a Hatch call per target.""" + job = make_job(target='ntp') + batches = [TestBatch(id='batch-01', batch_id='batch-01', job_list=[job], jobs_count=1, integrations=['ntp'])] + return mocker.patch('ddev.cli.ci.dispatch_tests.build_plan', return_value=batches) + + +@pytest.fixture +def pull_request(fake_async_github): + fake_async_github.mock_response('get_pull_request', PR_PAYLOAD) + return fake_async_github + + +@pytest.mark.parametrize( + 'reference', + [PR_NUMBER, f'https://github.com/DataDog/integrations-core/pull/{PR_NUMBER}'], + ids=['number', 'url'], +) +def test_a_pull_request_supplies_the_run_context(ddev, pull_request, planned, reference): + """`--pr` is the only input a local run should need: everything else comes from the API.""" + result = ddev('ci', 'dispatch-tests', '--pr', str(reference), '--dry-run') + + assert result.exit_code == 0, result.output + assert 'hs/a-branch' in result.output + assert 'head-sha-aaa' in result.output + assert 'a-target-branch' in result.output + # A pull request is tested at its merge commit, not at its head. + assert f'refs/pull/{PR_NUMBER}/merge' in result.output + + +def test_an_explicit_value_wins_over_the_resolved_one(ddev, pull_request, planned): + result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--base-sha', 'my-own-sha', '--dry-run') + + assert result.exit_code == 0, result.output + assert 'my-own-sha' in result.output + assert 'head-sha-aaa' not in result.output + + +def test_a_dry_run_dispatches_nothing(ddev, pull_request, planned): + result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--dry-run') + + assert result.exit_code == 0, result.output + pull_request.assert_not_called('create_workflow_dispatch') + pull_request.assert_not_called('create_issue_comment') + + +def test_a_reference_that_is_neither_a_number_nor_a_url_is_refused(ddev, planned): + result = ddev('ci', 'dispatch-tests', '--pr', 'not-a-pull-request', '--dry-run') + + assert result.exit_code == 1 + assert 'neither a pull request number nor a pull request URL' in result.output + + +def test_an_empty_plan_is_not_dispatched(ddev, fake_async_github, mocker): + """Nothing to test is a clean outcome, not a failure and not an empty comment.""" + mocker.patch('ddev.cli.ci.dispatch_tests.build_plan', return_value=[]) + + result = ddev('ci', 'dispatch-tests', '--base-sha', 'a-sha') + + assert result.exit_code == 0, result.output + assert 'No affected target to test.' in result.output + fake_async_github.assert_not_called('create_workflow_dispatch') diff --git a/ddev/tests/cli/ci/tests/batching/test_targets.py b/ddev/tests/cli/ci/tests/batching/test_targets.py index 823d863df546a..4154a19ecab49 100644 --- a/ddev/tests/cli/ci/tests/batching/test_targets.py +++ b/ddev/tests/cli/ci/tests/batching/test_targets.py @@ -9,6 +9,7 @@ from ddev.cli.ci.tests.batching.targets import ( UNTESTABLE_TARGETS, + AllTargetsRule, DirectTargetRule, RegistryRepositoryFacts, RepositoryWideRule, @@ -264,3 +265,8 @@ def test_registry_repository_facts_eligible_targets_excludes_untestable_and_poli ) assert RegistryRepositoryFacts(registry).eligible_targets() == ["postgres"] + + +def test_all_targets_rule_selects_every_eligible_target_without_a_change(): + """`--all` must not depend on the change set: a run that tests everything has nothing to diff.""" + assert list(AllTargetsRule()([], facts("postgres", "mysql", "mesos_slave"))) == ["mysql", "postgres"] diff --git a/ddev/tests/cli/ci/tests/test_dispatcher.py b/ddev/tests/cli/ci/tests/test_dispatcher.py new file mode 100644 index 0000000000000..d93e5945f82db --- /dev/null +++ b/ddev/tests/cli/ci/tests/test_dispatcher.py @@ -0,0 +1,180 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +"""Tests for the Dispatcher: the bus that carries a plan from dispatch to published report.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from ddev.cli.ci.tests.dispatcher import Dispatcher, DispatcherContext, RunContext +from ddev.cli.ci.tests.messages import BatchJob, TestBatch +from ddev.cli.ci.tests.pr_comment import COMMENT_MARKER +from ddev.cli.ci.tests.task_pull_request_updater import PullRequestUpdaterOptions, TaskPullRequestUpdater +from ddev.cli.ci.tests.task_test_gatherer import TaskTestGatherer +from ddev.cli.ci.tests.task_test_runner import TaskTestRunner, TestRunnerOptions +from ddev.utils.github_async import GitHubResponse +from ddev.utils.github_async.models import ArtifactsList, WorkflowJob, WorkflowJobsList, WorkflowRun +from tests.cli.ci.tests.helpers import jobs_reported, make_job +from tests.helpers.github_async import FakeAsyncGitHubClient + +CONTEXT = DispatcherContext( + owner="DataDog", + repo="integrations-core", + run_context=RunContext.PR, + checkout_sha="refs/pull/42/merge", + base_sha="head-sha", + branch="a-branch", + workflow="test-batch.yml", + workflow_ref="master", + target_branch="master", + pr_number=42, +) + + +def wrap(data): + return GitHubResponse(data=data, headers={}) + + +def build_bus( + client: FakeAsyncGitHubClient, + tmp_path: Path, + batches: list[TestBatch], + *, + pr_number: int | None = 42, +) -> Dispatcher: + """A Dispatcher over the three real tasks, so the subscriptions under test are production's.""" + runner = TaskTestRunner( + "test-runner", + client, # type: ignore[arg-type] + TestRunnerOptions( + owner=CONTEXT.owner, + repo=CONTEXT.repo, + workflow_id=CONTEXT.workflow, + ref=CONTEXT.workflow_ref, + base_sha=CONTEXT.base_sha, + checkout_sha=CONTEXT.checkout_sha, + artifacts_base_path=tmp_path / "artifacts", + poll_interval_seconds=0.0, + ), + ) + gatherer = TaskTestGatherer("test-gatherer", tmp_path / "results", batches) + updater = TaskPullRequestUpdater( + "pull-request-updater", + client, # type: ignore[arg-type] + PullRequestUpdaterOptions(owner=CONTEXT.owner, repo=CONTEXT.repo, pr_number=pr_number), + ) + return Dispatcher( + batches=batches, + client=client, # type: ignore[arg-type] + runner=runner, + gatherer=gatherer, + updater=updater, + max_timeout=30, + grace_period=0.2, + ) + + +def batch(job: BatchJob, batch_id: str = "batch-01") -> TestBatch: + return TestBatch(id=batch_id, batch_id=batch_id, job_list=[job], jobs_count=1, integrations=[job.target]) + + +@pytest.fixture +def client(request) -> FakeAsyncGitHubClient: + """A fake GitHub that completes every dispatched run with *conclusion*.""" + conclusion = getattr(request, "param", "success") + fake = FakeAsyncGitHubClient() + fake.mock_response( + "get_workflow_run", + wrap( + WorkflowRun( + id=123, + name="test-batch", + status="completed", + conclusion=conclusion, + html_url="https://github.com/DataDog/integrations-core/actions/runs/123", + ) + ), + ) + fake.mock_response("list_workflow_run_artifacts", wrap(ArtifactsList(total_count=0, artifacts=[]))) + return fake + + +def mock_job_result(fake: FakeAsyncGitHubClient, job: BatchJob, conclusion: str) -> None: + fake.mock_response( + "list_workflow_jobs", + wrap( + WorkflowJobsList( + total_count=1, + jobs=[WorkflowJob(id=1, run_id=123, name=job.name, status="completed", conclusion=conclusion)], + ) + ), + ) + + +def test_a_batch_travels_from_dispatch_to_the_pull_request_comment(client, tmp_path): + """The wiring assertion: one batch in, and its result reaches the comment. + + It fails if any of the three subscriptions is wrong, because each message is only produced by + the task that consumes the one before it. + """ + job = make_job() + mock_job_result(client, job, "success") + dispatcher = build_bus(client, tmp_path, [batch(job)]) + + dispatcher.run() + + dispatches = client.calls_to("create_workflow_dispatch") + assert len(dispatches) == 1 + assert dispatches[0].kwargs["workflow_id"] == "test-batch.yml" + assert dispatches[0].kwargs["ref"] == "master" + assert dispatches[0].kwargs["inputs"]["batch_id"] == "batch-01" + assert dispatches[0].kwargs["inputs"]["checkout_sha"] == "refs/pull/42/merge" + + # The plan is published before anything runs, then edited once the batch has been gathered. + created = client.calls_to("create_issue_comment") + edited = client.calls_to("update_issue_comment") + assert len(created) == 1 + assert created[0].kwargs["issue_number"] == 42 + assert len(edited) == 1 + assert jobs_reported(created[0].kwargs["body"]) == 0 + assert jobs_reported(edited[0].kwargs["body"]) == 1 + + outcome = dispatcher.outcome + assert outcome is not None + assert outcome.successful + assert outcome.progress.done + assert outcome.progress.passed == 1 + + +@pytest.mark.parametrize("client", ["failure"], indirect=True) +def test_a_failed_batch_makes_the_run_unsuccessful(client, tmp_path): + job = make_job() + mock_job_result(client, job, "failure") + dispatcher = build_bus(client, tmp_path, [batch(job)]) + + dispatcher.run() + + outcome = dispatcher.outcome + assert outcome is not None + assert not outcome.successful + assert outcome.progress.failed == 1 + + +def test_the_report_is_written_to_the_run_summary(client, tmp_path, monkeypatch): + """A run with no pull request has the run summary as its only report.""" + summary = tmp_path / "summary.md" + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary)) + job = make_job() + mock_job_result(client, job, "success") + dispatcher = build_bus(client, tmp_path, [batch(job)], pr_number=None) + + dispatcher.run() + + client.assert_not_called("create_issue_comment") + report = summary.read_text() + assert "Dispatcher tests" in report + # The marker exists to find a comment; nothing looks a run summary up. + assert COMMENT_MARKER not in report From 33fd8097a0b5c7ae78841ea070aec7c07d34d4df Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Thu, 20 Aug 2026 16:42:14 +0100 Subject: [PATCH 08/13] Add changelog entry Co-Authored-By: Claude Opus 5 (1M context) --- ddev/changelog.d/24935.added | 1 + 1 file changed, 1 insertion(+) create mode 100644 ddev/changelog.d/24935.added diff --git a/ddev/changelog.d/24935.added b/ddev/changelog.d/24935.added new file mode 100644 index 0000000000000..6c802710dd506 --- /dev/null +++ b/ddev/changelog.d/24935.added @@ -0,0 +1 @@ +Add the `ddev ci dispatch-tests` command and the Dispatcher that runs a batching plan. From 4efad9f0a3bc28182d32342dafb43639cca86d9a Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Thu, 20 Aug 2026 17:25:25 +0100 Subject: [PATCH 09/13] Make a worker-thread submission reach the event bus MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `asyncio.Queue` is not thread-safe, and a `SyncProcessor` runs in an executor thread. A `put_nowait` from there landing between `Queue.get`'s `empty()` check and its waiter being registered wakes nobody: the message stays in the deque, the loop awaits a waiter that will never resolve, and because the queue is not empty the bus never reaches its stop condition and spins until `max_timeout`. The gatherer is the only cross-thread emitter, so this was reachable the moment the Dispatcher chained it to the pull-request updater. A single-batch run — a pull request touching one integration — is the widest window, and would have hung for the full three-hour timeout while still reporting success. Submissions from off the loop thread now hop onto it. A caller already there, or with no bus running, still puts directly, so a message is queued by the time `submit_message` returns. Also: a run only counts as successful once its final report has been published, so a stall after the last batch can no longer exit 0 with a stale comment; and the run-summary test reads its file as UTF-8, which Windows does not default to. Co-Authored-By: Claude Opus 5 (1M context) --- ddev/src/ddev/cli/ci/tests/dispatcher.py | 8 ++++++-- .../cli/ci/tests/task_pull_request_updater.py | 18 +++++++++++++++++- ddev/tests/cli/ci/tests/test_dispatcher.py | 2 +- 3 files changed, 24 insertions(+), 4 deletions(-) diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher.py b/ddev/src/ddev/cli/ci/tests/dispatcher.py index c01acc939a00e..49c50f7f14d7b 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher.py @@ -71,6 +71,7 @@ class DispatcherOutcome: progress: DispatcherProgress pr_comment_failed: bool + final_report_published: bool error: Exception | None = None @property @@ -78,11 +79,13 @@ def successful(self) -> bool: """Whether every batch reached a non-failing terminal state and the report was published. A batch that never finished counts as a failure: `progress.done` is false, and a run whose - results are unknown must not read as green. + results are unknown must not read as green. Publishing the final report is part of that, + because a run that stalls after the last batch leaves the reader looking at stale progress. + An intermediate comment failure is not, since the next snapshot supersedes it. """ return ( self.error is None - and not self.pr_comment_failed + and self.final_report_published and self.progress.done and all(batch.status is not Status.FAILURE for batch in self.progress.batches) ) @@ -140,6 +143,7 @@ async def on_finalize(self, exception: Exception | None): self._outcome = DispatcherOutcome( progress=progress, pr_comment_failed=self._updater.pr_comment_failed, + final_report_published=self._updater.final_report_published, error=exception, ) self._logger.info(summary_line(progress)) diff --git a/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py b/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py index 2726f2fb740d9..56b79f3c0724c 100644 --- a/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py +++ b/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py @@ -77,6 +77,7 @@ def __init__(self, name: str, client: AsyncGitHubClient, options: PullRequestUpd self._latest_revision = -1 self._latest_body: str | None = None self._pr_comment_failed = False + self._final_report_published = False self._lock = asyncio.Lock() self._logger = logging.getLogger(f"{__name__}.{name}") @@ -93,6 +94,16 @@ def pr_comment_failed(self) -> bool: """Whether the newest report failed to reach its pull-request comment.""" return self._pr_comment_failed + @property + def final_report_published(self) -> bool: + """Whether the report for a completed run has been published. + + A run whose final snapshot never reached its destination has results nobody can see, so the + caller must not treat it as a success. An intermediate failure is different: the next + snapshot supersedes it. + """ + return self._final_report_published + async def process_message(self, message: UpdatePRComment): # Rendering is pure, so it happens outside the lock. body = render_comment(message.progress) @@ -111,8 +122,13 @@ async def process_message(self, message: UpdatePRComment): pr_number = self._options.pr_number if pr_number is None: self._logger.info("No pull request to update: %s", summary_line(message.progress), extra=log_extra) + published = True else: - self._pr_comment_failed = not await self._write(pr_number, message, body, log_extra) + published = await self._write(pr_number, message, body, log_extra) + self._pr_comment_failed = not published + + if message.progress.done and published: + self._final_report_published = True # Retained even when the write failed: this is the newest report we have, and losing the # comment must not lose the results. The revision advances with it, so a later snapshot diff --git a/ddev/tests/cli/ci/tests/test_dispatcher.py b/ddev/tests/cli/ci/tests/test_dispatcher.py index d93e5945f82db..66567f4d6d4ea 100644 --- a/ddev/tests/cli/ci/tests/test_dispatcher.py +++ b/ddev/tests/cli/ci/tests/test_dispatcher.py @@ -174,7 +174,7 @@ def test_the_report_is_written_to_the_run_summary(client, tmp_path, monkeypatch) dispatcher.run() client.assert_not_called("create_issue_comment") - report = summary.read_text() + report = summary.read_text(encoding="utf-8") assert "Dispatcher tests" in report # The marker exists to find a comment; nothing looks a run summary up. assert COMMENT_MARKER not in report From 029845a1dcf79a727cdc05e3cd0dc43003e455f7 Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Fri, 21 Aug 2026 16:29:52 +0100 Subject: [PATCH 10/13] Apply review findings to the Dispatcher entry point Reuse the PR-reference helpers `port-commit` already had rather than carrying a second copy: `resolve_owner_repo` and the pull-request URL pattern move to `ddev.utils.github`, and both commands read them there. Report a fatal bus failure as a message instead of a traceback, drop the two `DispatcherOutcome` fields nothing reads, and move the grace period into `DispatcherConfig` beside the other bus timings. The initial update's message id moves to the gatherer, which is the only thing that builds it. On the tests: use the fake client's own response wrapping, share one `TestBatch` builder, and assert the run summary at the layer that owns it rather than restating the renderer's contract. Co-Authored-By: Claude Opus 5 (1M context) --- ddev/src/ddev/cli/ci/dispatch_tests.py | 23 ++++----- ddev/src/ddev/cli/ci/tests/dispatcher.py | 39 ++++++--------- .../ddev/cli/ci/tests/dispatcher_config.py | 3 ++ .../cli/ci/tests/task_pull_request_updater.py | 7 +-- .../ddev/cli/ci/tests/task_test_gatherer.py | 7 ++- .../ddev/cli/release/port_commit_workflow.py | 16 +------ ddev/src/ddev/utils/github.py | 10 ++++ ddev/tests/cli/ci/test_dispatch_tests.py | 31 ++++++++---- ddev/tests/cli/ci/tests/helpers.py | 13 ++++- ddev/tests/cli/ci/tests/test_dispatcher.py | 47 ++++++------------- .../cli/ci/tests/test_task_test_gatherer.py | 10 ++-- 11 files changed, 98 insertions(+), 108 deletions(-) diff --git a/ddev/src/ddev/cli/ci/dispatch_tests.py b/ddev/src/ddev/cli/ci/dispatch_tests.py index 3836a8678bdc7..3f465fe1b7e34 100644 --- a/ddev/src/ddev/cli/ci/dispatch_tests.py +++ b/ddev/src/ddev/cli/ci/dispatch_tests.py @@ -18,7 +18,6 @@ from ddev.utils.github_async.models import PullRequest DEFAULT_OUTPUT_DIRECTORY = ".dispatcher" -MAX_LISTED_INTEGRATIONS = 10 @click.command(short_help='Run the Dispatcher to test a commit as parallel batches') @@ -78,6 +77,7 @@ def dispatch_tests( from ddev.cli.ci.tests.batching.build import HatchEnvironmentProvider from ddev.cli.ci.tests.dispatcher import DispatcherContext, RunContext, build_dispatcher from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig + from ddev.utils.github import resolve_owner_repo # One INFO line per request would bury the Dispatcher's own progress. logging.getLogger('httpx').setLevel(logging.WARNING) @@ -145,7 +145,12 @@ def dispatch_tests( output_path=base_path / 'results', run_logger=app.logger, ) - dispatcher.run() + # A fatal processor or hook failure leaves the bus by raising out of `run`. `on_finalize` has + # already published whatever it knew by then, so a message is more use here than a traceback. + try: + dispatcher.run() + except Exception as error: + app.abort(f'Dispatcher execution failed: {error}') outcome = dispatcher.outcome if outcome is None or not outcome.successful: @@ -154,15 +159,6 @@ def dispatch_tests( app.display_success('Dispatcher tests passed.') -def resolve_owner_repo(app: Application, repository: str | None) -> tuple[str, str]: - """Split `owner/name`, defaulting to the active repository and the `DataDog` organization.""" - full_name = repository or app.repo.full_name - owner, separator, name = full_name.partition('/') - if not separator: - return 'DataDog', full_name - return owner, name - - def fetch_pull_request(app: Application, owner: str, repo: str, reference: str) -> PullRequest: """Read the pull request named by *reference* (a number or a URL) from the GitHub API.""" import asyncio @@ -265,11 +261,12 @@ def display_plan(app: Application, context: DispatcherContext, batches: list[Tes for batch in batches: count = len(batch.integrations) app.display(f' {batch.batch_id}: {batch.jobs_count} jobs, {count} integration{"" if count == 1 else "s"}') - # A repository-wide run names every integration, which is hundreds of lines of no use here. app.display(f' {summarize(batch.integrations)}') -def summarize(names: list[str], limit: int = MAX_LISTED_INTEGRATIONS) -> str: +def summarize(names: list[str]) -> str: + """The first few names and a count of the rest: a repository-wide run has hundreds.""" + limit = 10 if len(names) <= limit: return ', '.join(names) return f'{", ".join(names[:limit])}, and {len(names) - limit} more' diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher.py b/ddev/src/ddev/cli/ci/tests/dispatcher.py index 49c50f7f14d7b..0b276367dd80f 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher.py @@ -27,11 +27,6 @@ from ddev.cli.ci.tests.progress import DispatcherProgress from ddev.utils.github_async import AsyncGitHubClient -INITIAL_UPDATE_MESSAGE_ID = "dispatcher-initial" -# A batch takes minutes, so the bus is idle between results far longer than the bus default -# allows. It only has to outlast the gap between a task finishing and its message being read. -DEFAULT_GRACE_PERIOD = 30.0 - logger = logging.getLogger(__name__) @@ -46,11 +41,12 @@ class RunContext(StrEnum): @dataclass(frozen=True) class DispatcherContext: - """Everything the Dispatcher needs to know about the run it is testing. + """The run being tested. `build_dispatcher` consumes part of it; the rest describes the run + for the plan header and, once metrics land, for their tags. `base_sha` and `checkout_sha` are deliberately separate: a pull request is tested at the merge - commit (`refs/pull//merge`) but its checks and metrics belong to the head commit. Outside a - pull request the two are the same. + commit (`refs/pull//merge`) but its checks belong to the head commit. Outside a pull request + the two are the same. """ owner: str @@ -70,22 +66,18 @@ class DispatcherOutcome: """What a finished Dispatcher execution amounts to, for the caller to exit on.""" progress: DispatcherProgress - pr_comment_failed: bool final_report_published: bool - error: Exception | None = None @property def successful(self) -> bool: - """Whether every batch reached a non-failing terminal state and the report was published. + """Whether every batch finished without failing and the final report reached its reader. - A batch that never finished counts as a failure: `progress.done` is false, and a run whose - results are unknown must not read as green. Publishing the final report is part of that, - because a run that stalls after the last batch leaves the reader looking at stale progress. - An intermediate comment failure is not, since the next snapshot supersedes it. + A run whose results nobody can see is not green, so an unfinished plan and an unpublished + final report both count as failures. An intermediate comment failure does not: the next + snapshot supersedes it. """ return ( - self.error is None - and self.final_report_published + self.final_report_published and self.progress.done and all(batch.status is not Status.FAILURE for batch in self.progress.batches) ) @@ -94,10 +86,8 @@ def successful(self) -> bool: class Dispatcher(EventBusOrchestrator): """Runs a batching plan to completion and publishes its result. - The whole plan is known before the bus starts, so `on_initialize` primes the queue with the - initial pull-request update and every batch, and the tasks carry it from there: - `TestBatch` -> runner -> `BatchFinished` -> gatherer -> `UpdatePRComment` -> updater. The bus - stops when the queue drains and no task is left running. + The whole plan is known before the bus starts, so `on_initialize` queues the initial update and + every batch: `TestBatch` -> runner -> `BatchFinished` -> gatherer -> `UpdatePRComment` -> updater. """ def __init__( @@ -109,7 +99,7 @@ def __init__( gatherer: TaskTestGatherer, updater: TaskPullRequestUpdater, max_timeout: float | None, - grace_period: float = DEFAULT_GRACE_PERIOD, + grace_period: float, run_logger: logging.Logger | None = None, ): super().__init__(run_logger or logger, max_timeout=max_timeout, grace_period=grace_period) @@ -129,7 +119,7 @@ def outcome(self) -> DispatcherOutcome | None: return self._outcome async def on_initialize(self): - self.submit_message(self._gatherer.build_initial_update(INITIAL_UPDATE_MESSAGE_ID)) + self.submit_message(self._gatherer.build_initial_update()) for batch in self._batches: self.submit_message(batch) self._logger.info("Dispatched %s batches", len(self._batches)) @@ -142,9 +132,7 @@ async def on_finalize(self, exception: Exception | None): progress = self._gatherer.progress self._outcome = DispatcherOutcome( progress=progress, - pr_comment_failed=self._updater.pr_comment_failed, final_report_published=self._updater.final_report_published, - error=exception, ) self._logger.info(summary_line(progress)) if (body := self._updater.latest_body) is not None: @@ -204,5 +192,6 @@ def build_dispatcher( gatherer=gatherer, updater=updater, max_timeout=config.global_timeout_seconds, + grace_period=config.grace_period_seconds, run_logger=active_logger, ) diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher_config.py b/ddev/src/ddev/cli/ci/tests/dispatcher_config.py index 3eb69e5906627..6b8d59728c37b 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher_config.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher_config.py @@ -32,6 +32,9 @@ class DispatcherConfig(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") global_timeout_seconds: float = Field(default=10800.0, gt=0) # 3 hours + # How long the bus waits for a new message once nothing is running. A batch runs inside a + # task, so this only spans the gap between a task finishing and its message being read. + grace_period_seconds: float = Field(default=30.0, gt=0) # Used when Hatch does not declare a Python version. default_python_version: str = Field(default="3.13", pattern=r"^\d+\.\d+$") # The workflow each batch is dispatched to, by file name or numeric id. diff --git a/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py b/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py index 56b79f3c0724c..48dedf5134b62 100644 --- a/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py +++ b/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py @@ -96,12 +96,7 @@ def pr_comment_failed(self) -> bool: @property def final_report_published(self) -> bool: - """Whether the report for a completed run has been published. - - A run whose final snapshot never reached its destination has results nobody can see, so the - caller must not treat it as a success. An intermediate failure is different: the next - snapshot supersedes it. - """ + """Whether the report for a completed run reached its destination.""" return self._final_report_published async def process_message(self, message: UpdatePRComment): diff --git a/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py b/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py index 54e566ca8f0df..63d5d4e8e859c 100644 --- a/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py +++ b/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py @@ -45,6 +45,9 @@ # workflow-job conclusion, and a job with no correlated workflow job is a runner bug and raises. COVERAGE_GLOB = "coverage*.xml" JUNIT_GLOB = "test-*.xml" +# Every later update borrows the id of the ``BatchFinished`` that caused it. Revision ``0`` has no +# cause, so it carries its own. +INITIAL_UPDATE_MESSAGE_ID = "dispatcher-initial" class TaskTestGatherer(SyncProcessor[BatchFinished]): @@ -127,14 +130,14 @@ def _done(self) -> bool: """Whether every batch is terminal. Hold ``self._lock``.""" return all(batch.state is ExecutionState.FINISHED for batch in self._progress_by_batch.values()) - def build_initial_update(self, message_id: str) -> UpdatePRComment: + def build_initial_update(self) -> UpdatePRComment: """Revision ``0``: the complete plan, before any batch has been dispatched. Returned rather than submitted: a processor can only submit once the bus has attached its queue, so the dispatcher entry point publishes this when it starts the bus. """ with self._lock: - return self.build_update_message(message_id, revision=0, done=False) + return self.build_update_message(INITIAL_UPDATE_MESSAGE_ID, revision=0, done=False) def build_update_message(self, message_id: str, revision: int, done: bool) -> UpdatePRComment: """Build an ``UpdatePRComment`` for *revision*. Hold ``self._lock`` when state is live.""" diff --git a/ddev/src/ddev/cli/release/port_commit_workflow.py b/ddev/src/ddev/cli/release/port_commit_workflow.py index 2a3cfbcafea11..85f70aabc6ae4 100644 --- a/ddev/src/ddev/cli/release/port_commit_workflow.py +++ b/ddev/src/ddev/cli/release/port_commit_workflow.py @@ -26,6 +26,7 @@ from ddev.utils.fs import Path from ddev.utils.git import GitRepository +from ddev.utils.github import PULL_REQUEST_URL_PATTERN, resolve_owner_repo if TYPE_CHECKING: from ddev.cli.application import Application @@ -40,7 +41,6 @@ HEX_PATTERN = re.compile(r'^[0-9a-fA-F]+$') DIGITS_PATTERN = re.compile(r'^\d+$') PR_PREFIX_PATTERN = re.compile(r'^PR-(\d+)$', re.IGNORECASE) -PR_URL_PATTERN = re.compile(r'^https?://github\.com/[^/]+/[^/]+/pull/(\d+)(?:[/?#].*)?$', re.IGNORECASE) # Paths whose content is regenerated per-branch, so a backport must reset them to the target # branch's version instead of carrying over the source commit's. A trailing slash matches as a @@ -417,7 +417,7 @@ def _resolve_input(app: Application, raw: str, *, dry_run: bool) -> str: def _extract_explicit_pr_number(raw: str) -> int | None: """Return the PR number when `raw` is a `PR-12345` token or a GitHub PR URL, else None.""" - for pattern in (PR_PREFIX_PATTERN, PR_URL_PATTERN): + for pattern in (PR_PREFIX_PATTERN, PULL_REQUEST_URL_PATTERN): match = pattern.fullmatch(raw) if match: return int(match.group(1)) @@ -627,18 +627,6 @@ def derive_backport_bases(pr: PullRequest) -> list[str]: return bases -def resolve_owner_repo(app: Application) -> tuple[str, str]: - """Resolve (owner, repo) for the active repository. - - Falls back to `DataDog/` when `full_name` is unqualified. - """ - full = app.repo.full_name - if '/' in full: - owner, repo = full.split('/', 1) - return owner, repo - return 'DataDog', full - - def _sanitize_branch_for_path(branch: str) -> str: return branch.replace('/', '-') diff --git a/ddev/src/ddev/utils/github.py b/ddev/src/ddev/utils/github.py index a3c4c8df24a05..2d38584db6455 100644 --- a/ddev/src/ddev/utils/github.py +++ b/ddev/src/ddev/utils/github.py @@ -26,10 +26,20 @@ from httpx import Client + from ddev.cli.application import Application from ddev.cli.terminal import BorrowedStatus from ddev.repo.core import Repository +def resolve_owner_repo(app: Application, repository: str | None = None) -> tuple[str, str]: + """Split `owner/name`, defaulting to the active repository and the `DataDog` organization.""" + full_name = repository or app.repo.full_name + owner, separator, name = full_name.partition('/') + if not separator: + return 'DataDog', full_name + return owner, name + + def parse_pull_request_reference(value: str) -> int | None: """Return the pull-request number in *value*, or None when it is neither shape. diff --git a/ddev/tests/cli/ci/test_dispatch_tests.py b/ddev/tests/cli/ci/test_dispatch_tests.py index fe514ca21c7c5..34f80d396f199 100644 --- a/ddev/tests/cli/ci/test_dispatch_tests.py +++ b/ddev/tests/cli/ci/test_dispatch_tests.py @@ -7,9 +7,8 @@ import pytest -from ddev.cli.ci.tests.messages import TestBatch from ddev.utils.github_async.models import PullRequest -from tests.cli.ci.tests.helpers import make_job +from tests.cli.ci.tests.helpers import make_batch, make_job PR_NUMBER = 4242 PR_PAYLOAD = PullRequest( @@ -23,13 +22,13 @@ @pytest.fixture def planned(mocker): """Stand in for the planning layer, which has its own tests and costs a Hatch call per target.""" - job = make_job(target='ntp') - batches = [TestBatch(id='batch-01', batch_id='batch-01', job_list=[job], jobs_count=1, integrations=['ntp'])] + batches = [make_batch(make_job(target='ntp'))] return mocker.patch('ddev.cli.ci.dispatch_tests.build_plan', return_value=batches) @pytest.fixture -def pull_request(fake_async_github): +def github(fake_async_github): + """A GitHub that answers `get_pull_request` with `PR_PAYLOAD`.""" fake_async_github.mock_response('get_pull_request', PR_PAYLOAD) return fake_async_github @@ -39,7 +38,7 @@ def pull_request(fake_async_github): [PR_NUMBER, f'https://github.com/DataDog/integrations-core/pull/{PR_NUMBER}'], ids=['number', 'url'], ) -def test_a_pull_request_supplies_the_run_context(ddev, pull_request, planned, reference): +def test_a_pull_request_supplies_the_run_context(ddev, github, planned, reference): """`--pr` is the only input a local run should need: everything else comes from the API.""" result = ddev('ci', 'dispatch-tests', '--pr', str(reference), '--dry-run') @@ -51,7 +50,7 @@ def test_a_pull_request_supplies_the_run_context(ddev, pull_request, planned, re assert f'refs/pull/{PR_NUMBER}/merge' in result.output -def test_an_explicit_value_wins_over_the_resolved_one(ddev, pull_request, planned): +def test_an_explicit_value_wins_over_the_resolved_one(ddev, github, planned): result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--base-sha', 'my-own-sha', '--dry-run') assert result.exit_code == 0, result.output @@ -59,12 +58,12 @@ def test_an_explicit_value_wins_over_the_resolved_one(ddev, pull_request, planne assert 'head-sha-aaa' not in result.output -def test_a_dry_run_dispatches_nothing(ddev, pull_request, planned): +def test_a_dry_run_dispatches_nothing(ddev, github, planned): result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--dry-run') assert result.exit_code == 0, result.output - pull_request.assert_not_called('create_workflow_dispatch') - pull_request.assert_not_called('create_issue_comment') + github.assert_not_called('create_workflow_dispatch') + github.assert_not_called('create_issue_comment') def test_a_reference_that_is_neither_a_number_nor_a_url_is_refused(ddev, planned): @@ -83,3 +82,15 @@ def test_an_empty_plan_is_not_dispatched(ddev, fake_async_github, mocker): assert result.exit_code == 0, result.output assert 'No affected target to test.' in result.output fake_async_github.assert_not_called('create_workflow_dispatch') + + +def test_the_context_option_offers_every_run_context(): + """`--context` restates its choices as literals: `RunContext` lives in a module too heavy to + import while building the decorator, so a member added there must not be left unreachable. + """ + from ddev.cli.ci.dispatch_tests import dispatch_tests + from ddev.cli.ci.tests.dispatcher import RunContext + + option = next(param for param in dispatch_tests.params if param.name == 'run_context') + + assert set(option.type.choices) == {member.value for member in RunContext} diff --git a/ddev/tests/cli/ci/tests/helpers.py b/ddev/tests/cli/ci/tests/helpers.py index f463273566f22..8cabff64deeb4 100644 --- a/ddev/tests/cli/ci/tests/helpers.py +++ b/ddev/tests/cli/ci/tests/helpers.py @@ -10,7 +10,7 @@ from collections.abc import Iterable, Sequence from ddev.cli.ci.tests.batching.units import ResolvedEnvironment, TestUnit -from ddev.cli.ci.tests.messages import BatchJob +from ddev.cli.ci.tests.messages import BatchJob, TestBatch from ddev.cli.ci.tests.progress import ( BatchProgress, DispatcherProgress, @@ -91,6 +91,17 @@ def make_job( ) +def make_batch(*batch_jobs: BatchJob, batch_id: str = "batch-01") -> TestBatch: + job_list = list(batch_jobs) or [make_job()] + return TestBatch( + id=batch_id, + batch_id=batch_id, + job_list=job_list, + jobs_count=len(job_list), + integrations=sorted({job.target for job in job_list}), + ) + + def jobs(target: str, count: int) -> list[BatchJob]: # Each job carries a distinct environment, as production jobs within an integration do, so # names and artifact identities are unique within the target. diff --git a/ddev/tests/cli/ci/tests/test_dispatcher.py b/ddev/tests/cli/ci/tests/test_dispatcher.py index 66567f4d6d4ea..3ca09b00c612c 100644 --- a/ddev/tests/cli/ci/tests/test_dispatcher.py +++ b/ddev/tests/cli/ci/tests/test_dispatcher.py @@ -11,13 +11,11 @@ from ddev.cli.ci.tests.dispatcher import Dispatcher, DispatcherContext, RunContext from ddev.cli.ci.tests.messages import BatchJob, TestBatch -from ddev.cli.ci.tests.pr_comment import COMMENT_MARKER from ddev.cli.ci.tests.task_pull_request_updater import PullRequestUpdaterOptions, TaskPullRequestUpdater from ddev.cli.ci.tests.task_test_gatherer import TaskTestGatherer from ddev.cli.ci.tests.task_test_runner import TaskTestRunner, TestRunnerOptions -from ddev.utils.github_async import GitHubResponse from ddev.utils.github_async.models import ArtifactsList, WorkflowJob, WorkflowJobsList, WorkflowRun -from tests.cli.ci.tests.helpers import jobs_reported, make_job +from tests.cli.ci.tests.helpers import jobs_reported, make_batch, make_job from tests.helpers.github_async import FakeAsyncGitHubClient CONTEXT = DispatcherContext( @@ -34,10 +32,6 @@ ) -def wrap(data): - return GitHubResponse(data=data, headers={}) - - def build_bus( client: FakeAsyncGitHubClient, tmp_path: Path, @@ -77,10 +71,6 @@ def build_bus( ) -def batch(job: BatchJob, batch_id: str = "batch-01") -> TestBatch: - return TestBatch(id=batch_id, batch_id=batch_id, job_list=[job], jobs_count=1, integrations=[job.target]) - - @pytest.fixture def client(request) -> FakeAsyncGitHubClient: """A fake GitHub that completes every dispatched run with *conclusion*.""" @@ -88,28 +78,24 @@ def client(request) -> FakeAsyncGitHubClient: fake = FakeAsyncGitHubClient() fake.mock_response( "get_workflow_run", - wrap( - WorkflowRun( - id=123, - name="test-batch", - status="completed", - conclusion=conclusion, - html_url="https://github.com/DataDog/integrations-core/actions/runs/123", - ) + WorkflowRun( + id=123, + name="test-batch", + status="completed", + conclusion=conclusion, + html_url="https://github.com/DataDog/integrations-core/actions/runs/123", ), ) - fake.mock_response("list_workflow_run_artifacts", wrap(ArtifactsList(total_count=0, artifacts=[]))) + fake.mock_response("list_workflow_run_artifacts", ArtifactsList(total_count=0, artifacts=[])) return fake def mock_job_result(fake: FakeAsyncGitHubClient, job: BatchJob, conclusion: str) -> None: fake.mock_response( "list_workflow_jobs", - wrap( - WorkflowJobsList( - total_count=1, - jobs=[WorkflowJob(id=1, run_id=123, name=job.name, status="completed", conclusion=conclusion)], - ) + WorkflowJobsList( + total_count=1, + jobs=[WorkflowJob(id=1, run_id=123, name=job.name, status="completed", conclusion=conclusion)], ), ) @@ -122,7 +108,7 @@ def test_a_batch_travels_from_dispatch_to_the_pull_request_comment(client, tmp_p """ job = make_job() mock_job_result(client, job, "success") - dispatcher = build_bus(client, tmp_path, [batch(job)]) + dispatcher = build_bus(client, tmp_path, [make_batch(job)]) dispatcher.run() @@ -153,7 +139,7 @@ def test_a_batch_travels_from_dispatch_to_the_pull_request_comment(client, tmp_p def test_a_failed_batch_makes_the_run_unsuccessful(client, tmp_path): job = make_job() mock_job_result(client, job, "failure") - dispatcher = build_bus(client, tmp_path, [batch(job)]) + dispatcher = build_bus(client, tmp_path, [make_batch(job)]) dispatcher.run() @@ -169,12 +155,9 @@ def test_the_report_is_written_to_the_run_summary(client, tmp_path, monkeypatch) monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary)) job = make_job() mock_job_result(client, job, "success") - dispatcher = build_bus(client, tmp_path, [batch(job)], pr_number=None) + dispatcher = build_bus(client, tmp_path, [make_batch(job)], pr_number=None) dispatcher.run() client.assert_not_called("create_issue_comment") - report = summary.read_text(encoding="utf-8") - assert "Dispatcher tests" in report - # The marker exists to find a comment; nothing looks a run summary up. - assert COMMENT_MARKER not in report + assert jobs_reported(summary.read_text(encoding="utf-8")) == 1 diff --git a/ddev/tests/cli/ci/tests/test_task_test_gatherer.py b/ddev/tests/cli/ci/tests/test_task_test_gatherer.py index b77e356092d4e..1c6209c5fda02 100644 --- a/ddev/tests/cli/ci/tests/test_task_test_gatherer.py +++ b/ddev/tests/cli/ci/tests/test_task_test_gatherer.py @@ -31,7 +31,7 @@ from ddev.cli.ci.tests.progress import ExecutionState, ProgressError from ddev.cli.ci.tests.status import Status from ddev.cli.ci.tests.task_pull_request_updater import PullRequestUpdaterOptions, TaskPullRequestUpdater -from ddev.cli.ci.tests.task_test_gatherer import TaskTestGatherer +from ddev.cli.ci.tests.task_test_gatherer import INITIAL_UPDATE_MESSAGE_ID, TaskTestGatherer from ddev.event_bus.orchestrator import BaseMessage, EventBusOrchestrator from ddev.utils.github_async.models import JobStep, WorkflowJob from ddev.utils.junit import TestStatus @@ -553,7 +553,7 @@ def test_unplanned_batch_is_ignored(tmp_path: Path) -> None: assert drain_queue(gatherer.bus.queue) == [] assert gatherer._revision == 0 - assert [batch.batch_id for batch in gatherer.build_initial_update("initial").progress.batches] == ["batch-1"] + assert [batch.batch_id for batch in gatherer.build_initial_update().progress.batches] == ["batch-1"] # Nor may it write into the output tree the planned batches publish from. assert not (tmp_path / "out").exists() @@ -665,8 +665,8 @@ def test_initial_update_is_revision_zero_over_the_whole_plan(tmp_path: Path) -> plan = {"b1": [_batch_job("j1"), _batch_job("j2", target="kafka")], "b2": [_batch_job("j3", target="redis")]} gatherer = _make_gatherer(tmp_path, plan) - update = gatherer.build_initial_update("initial") - assert (update.id, update.revision) == ("initial", 0) + update = gatherer.build_initial_update() + assert (update.id, update.revision) == (INITIAL_UPDATE_MESSAGE_ID, 0) assert _registry(gatherer) == [] progress = update.progress @@ -1088,7 +1088,7 @@ def test_gatherer_updates_the_pr_comment_through_the_event_bus(tmp_path: Path): bus.register_processor(gatherer, [BatchFinished]) bus.register_processor(updater, [UpdatePRComment]) - bus.submit_message(gatherer.build_initial_update("initial")) + bus.submit_message(gatherer.build_initial_update()) for index, (batch_id, jobs) in enumerate(plan.items(), start=1): artifacts = tmp_path / "artifacts" / batch_id results = [_scenario_job(artifacts, job.target, "success", JUNIT_PASSING, run_id=index) for job in jobs] From fc08a88158e6b7816da5c7a793b58c5a3f7a2cdb Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Mon, 24 Aug 2026 13:07:07 +0100 Subject: [PATCH 11/13] Document DispatcherConfig fields as attribute docstrings Half the fields carried a comment and half carried nothing, which reads as an oversight rather than a choice. Every field is now documented the same way, below the field, where the documentation is attached to what it describes. Co-Authored-By: Claude Opus 5 (1M context) --- .../ddev/cli/ci/tests/dispatcher_config.py | 35 +++++++++++++++---- 1 file changed, 28 insertions(+), 7 deletions(-) diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher_config.py b/ddev/src/ddev/cli/ci/tests/dispatcher_config.py index 6b8d59728c37b..51842ee01b024 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher_config.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher_config.py @@ -31,20 +31,41 @@ class DispatcherConfig(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") - global_timeout_seconds: float = Field(default=10800.0, gt=0) # 3 hours - # How long the bus waits for a new message once nothing is running. A batch runs inside a - # task, so this only spans the gap between a task finishing and its message being read. + global_timeout_seconds: float = Field(default=10800.0, gt=0) + """Wall-clock ceiling on a whole run, in seconds. Three hours by default.""" + grace_period_seconds: float = Field(default=30.0, gt=0) - # Used when Hatch does not declare a Python version. + """How long the bus waits for a new message once nothing is running. + + A batch runs inside a task, so this only spans the gap between a task finishing and its message + being read, not the batch itself. + """ + default_python_version: str = Field(default="3.13", pattern=r"^\d+\.\d+$") - # The workflow each batch is dispatched to, by file name or numeric id. + """Python version a job runs on when Hatch declares none for its environment.""" + workflow: str = "test-batch.yml" - # The ref the workflow definition is loaded from. Never a pull-request ref: the definition must - # come from a reviewed branch even when the code under test does not. + """The workflow each batch is dispatched to, by file name or numeric id.""" + workflow_ref: str = "master" + """The ref the workflow definition is loaded from. + + Never a pull-request ref: the definition must come from a reviewed branch even when the code + under test does not. + """ + poll_interval_seconds: float = Field(default=30.0, gt=0) + """How long the runner waits between polls of a batch's workflow run. + + Every batch polls at this rate for as long as it runs, so it dominates the run's GitHub API + usage against the shared octo-sts budget. + """ + batching: BatchingConfig = BatchingConfig() + """Policy for turning discovered test units into batches, from `[dispatcher.batching]`.""" + github_rate_limits: RateLimiterFactoryConfig = RateLimiterFactoryConfig() + """Rate limiter tiers shared by every task, from `[dispatcher.github_rate_limits]`.""" @classmethod def from_repo_config(cls, repo_config: RepositoryConfig) -> DispatcherConfig: From 138b6ed2b84ad1513c93273dc5972ebb1b07d560 Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Mon, 24 Aug 2026 14:36:05 +0100 Subject: [PATCH 12/13] Say what the final report guarantees, and test --context through the CLI `final_report_published` and `DispatcherOutcome.successful` claimed the final report reached a reader. On the no-pull-request path the flag only means there was no comment to lose it to, and `write_step_summary` is a documented no-op outside GitHub Actions. Both docstrings now say what holds. Not failing a run that has nowhere to report to is deliberate, so the behaviour is unchanged. The `--context` test read Click's `params` and compared the choices with the enum, which passes even if the option is wired to the wrong destination. It now parameterizes over `RunContext` and runs the command, so it still catches a member that was never wired up while asserting what a caller can observe. Co-Authored-By: Claude Opus 5 (1M context) --- ddev/src/ddev/cli/ci/tests/dispatcher.py | 12 ++++++++---- .../cli/ci/tests/task_pull_request_updater.py | 2 +- ddev/tests/cli/ci/test_dispatch_tests.py | 17 +++++++++-------- 3 files changed, 18 insertions(+), 13 deletions(-) diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher.py b/ddev/src/ddev/cli/ci/tests/dispatcher.py index 0b276367dd80f..5027718ce4cc9 100644 --- a/ddev/src/ddev/cli/ci/tests/dispatcher.py +++ b/ddev/src/ddev/cli/ci/tests/dispatcher.py @@ -70,11 +70,15 @@ class DispatcherOutcome: @property def successful(self) -> bool: - """Whether every batch finished without failing and the final report reached its reader. + """Whether every batch finished without failing and the final report was not lost. - A run whose results nobody can see is not green, so an unfinished plan and an unpublished - final report both count as failures. An intermediate comment failure does not: the next - snapshot supersedes it. + An unfinished plan is a failure: results nobody can see must not read as green. So is losing + the final snapshot to a pull-request comment that would not take it. An intermediate comment + failure is not, since the next snapshot supersedes it. + + A run with no pull request has nothing to lose the report to, so it passes that condition on + arrival. `on_finalize` logs `summary_line` whatever happens, and the run summary is written + when GitHub Actions offers one, so such a run still reports somewhere. """ return ( self.final_report_published diff --git a/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py b/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py index 48dedf5134b62..8d12aa8cd9dbd 100644 --- a/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py +++ b/ddev/src/ddev/cli/ci/tests/task_pull_request_updater.py @@ -96,7 +96,7 @@ def pr_comment_failed(self) -> bool: @property def final_report_published(self) -> bool: - """Whether the report for a completed run reached its destination.""" + """Whether a completed run's report was not lost: it reached the comment, or had none to reach.""" return self._final_report_published async def process_message(self, message: UpdatePRComment): diff --git a/ddev/tests/cli/ci/test_dispatch_tests.py b/ddev/tests/cli/ci/test_dispatch_tests.py index 34f80d396f199..8278a74c0c130 100644 --- a/ddev/tests/cli/ci/test_dispatch_tests.py +++ b/ddev/tests/cli/ci/test_dispatch_tests.py @@ -7,6 +7,7 @@ import pytest +from ddev.cli.ci.tests.dispatcher import RunContext from ddev.utils.github_async.models import PullRequest from tests.cli.ci.tests.helpers import make_batch, make_job @@ -84,13 +85,13 @@ def test_an_empty_plan_is_not_dispatched(ddev, fake_async_github, mocker): fake_async_github.assert_not_called('create_workflow_dispatch') -def test_the_context_option_offers_every_run_context(): - """`--context` restates its choices as literals: `RunContext` lives in a module too heavy to - import while building the decorator, so a member added there must not be left unreachable. +@pytest.mark.parametrize('run_context', list(RunContext), ids=lambda member: member.value) +def test_every_run_context_is_accepted(ddev, github, planned, run_context): + """`--context` restates its choices as literals, because `RunContext` lives in a module too + heavy to import while building the decorator. Parameterizing over the enum catches a member + added there and never wired up. """ - from ddev.cli.ci.dispatch_tests import dispatch_tests - from ddev.cli.ci.tests.dispatcher import RunContext + result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--context', run_context.value, '--dry-run') - option = next(param for param in dispatch_tests.params if param.name == 'run_context') - - assert set(option.type.choices) == {member.value for member in RunContext} + assert result.exit_code == 0, result.output + assert f'Context -> {run_context.value}' in result.output From 52dc245bc761f7b72313966ee19d40d2f6888e12 Mon Sep 17 00:00:00 2001 From: HadhemiDD Date: Tue, 25 Aug 2026 14:19:47 +0100 Subject: [PATCH 13/13] Validate every option before the run does any work Adds validate_options as the command's first call. It refuses the four options --pr reads from GitHub (--pr-number, --branch, --base-sha, --target-branch), parses the --pr reference, and requires a token for any run that reads a pull request or dispatches anything, leaving a dry run planning from local git as the one case that needs none. It returns the parsed number and the token, so fetch_pull_request takes both as arguments and holds no validation of its own: the token was previously looked up and checked in two places with two messages. Also drops --artifacts-dir. Artifacts already default to a subfolder of the output directory and nothing passed the option, so all it offered was a way to scatter one run's outputs across unrelated places. Co-Authored-By: Claude Opus 5 (1M context) --- ddev/src/ddev/cli/ci/dispatch_tests.py | 85 ++++++++++++++++++------ ddev/tests/cli/ci/test_dispatch_tests.py | 35 ++++++++-- 2 files changed, 95 insertions(+), 25 deletions(-) diff --git a/ddev/src/ddev/cli/ci/dispatch_tests.py b/ddev/src/ddev/cli/ci/dispatch_tests.py index 3f465fe1b7e34..1683520efcf33 100644 --- a/ddev/src/ddev/cli/ci/dispatch_tests.py +++ b/ddev/src/ddev/cli/ci/dispatch_tests.py @@ -45,8 +45,11 @@ @click.option('--all', 'all_targets', is_flag=True, help='Test every eligible target instead of the affected ones.') @click.option('--workflow', default=None, help='Workflow each batch is dispatched to.') @click.option('--workflow-ref', default=None, help='Ref the workflow definition is loaded from.') -@click.option('--artifacts-dir', default=None, help='Where downloaded artifacts are written.') -@click.option('--output-dir', default=None, help='Where coverage and test results are organized.') +@click.option( + '--output-dir', + default=None, + help='Where the run writes what it produces: artifacts, coverage and test results.', +) @click.option('--dry-run', is_flag=True, help='Show the plan and the resolved context without calling GitHub.') def dispatch_tests( app: Application, @@ -61,7 +64,6 @@ def dispatch_tests( all_targets: bool, workflow: str | None, workflow_ref: str | None, - artifacts_dir: str | None, output_dir: str | None, dry_run: bool, ) -> None: @@ -79,6 +81,16 @@ def dispatch_tests( from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig from ddev.utils.github import resolve_owner_repo + requested_pr, token = validate_options( + app, + pull_request=pull_request, + pr_number=pr_number, + branch=branch, + base_sha=base_sha, + target_branch=target_branch, + dry_run=dry_run, + ) + # One INFO line per request would bury the Dispatcher's own progress. logging.getLogger('httpx').setLevel(logging.WARNING) @@ -86,8 +98,8 @@ def dispatch_tests( owner, repo = resolve_owner_repo(app, repository) resolved_number = resolved_branch = resolved_sha = resolved_target = None - if pull_request is not None: - resolved = fetch_pull_request(app, owner, repo, pull_request) + if requested_pr is not None: + resolved = fetch_pull_request(app, owner, repo, requested_pr, token) if resolved.head is None or resolved.base is None: app.abort(f'Pull request {resolved.number} reports no branch references.') resolved_number = resolved.number @@ -131,17 +143,13 @@ def dispatch_tests( app.display_info('Dry run: nothing was dispatched.') return - token = app.config.github.token - if not token: - app.abort('A GitHub token is required. Set `github.token` in your ddev config.') - base_path = Path(output_dir) if output_dir else app.repo.path / DEFAULT_OUTPUT_DIRECTORY dispatcher = build_dispatcher( batches=batches, context=context, config=config, token=token, - artifacts_path=Path(artifacts_dir) if artifacts_dir else base_path / 'artifacts', + artifacts_path=base_path / 'artifacts', output_path=base_path / 'results', run_logger=app.logger, ) @@ -159,25 +167,60 @@ def dispatch_tests( app.display_success('Dispatcher tests passed.') -def fetch_pull_request(app: Application, owner: str, repo: str, reference: str) -> PullRequest: - """Read the pull request named by *reference* (a number or a URL) from the GitHub API.""" +def validate_options( + app: Application, + *, + pull_request: str | None, + pr_number: int | None, + branch: str | None, + base_sha: str | None, + target_branch: str | None, + dry_run: bool, +) -> tuple[int | None, str]: + """Check every input before the run does any work, and return what checking them resolved. + + That is the pull request ``--pr`` names, if any, and the GitHub token, empty when the run needs + none: a dry run planning from local git talks to nobody. Reading a pull request needs a token + even for a dry run, because the API client refuses to be built without one. + """ + from ddev.utils.github import parse_pull_request_reference + + requested_pr = None + if pull_request is not None: + resolved_by_pr = [ + name + for name, value in ( + ('`--pr-number`', pr_number), + ('`--branch`', branch), + ('`--base-sha`', base_sha), + ('`--target-branch`', target_branch), + ) + if value is not None + ] + if resolved_by_pr: + app.abort(f'{", ".join(resolved_by_pr)} cannot be passed with `--pr`, which reads them from GitHub.') + + requested_pr = parse_pull_request_reference(pull_request) + if requested_pr is None: + app.abort(f'`{pull_request}` is neither a pull request number nor a pull request URL.') + + token = app.config.github.token + if not token and (pull_request is not None or not dry_run): + app.abort('A GitHub token is required. Set `github.token` in your ddev config.') + + return requested_pr, token + + +def fetch_pull_request(app: Application, owner: str, repo: str, number: int, token: str) -> PullRequest: + """Read pull request *number* from the GitHub API.""" import asyncio import httpx from pydantic import ValidationError - from ddev.utils.github import parse_pull_request_reference from ddev.utils.github_async import async_github_client from ddev.utils.github_errors import GitHubAuthenticationError - number = parse_pull_request_reference(reference) - if number is None: - app.abort(f'`{reference}` is neither a pull request number nor a pull request URL.') - - token = app.config.github.token - if not token: - app.abort('A GitHub token is required to read a pull request. Set `github.token` in your ddev config.') - async def fetch() -> PullRequest: async with async_github_client(token=token) as client: response = await client.get_pull_request(owner, repo, number) diff --git a/ddev/tests/cli/ci/test_dispatch_tests.py b/ddev/tests/cli/ci/test_dispatch_tests.py index 8278a74c0c130..ca71964631870 100644 --- a/ddev/tests/cli/ci/test_dispatch_tests.py +++ b/ddev/tests/cli/ci/test_dispatch_tests.py @@ -51,12 +51,38 @@ def test_a_pull_request_supplies_the_run_context(ddev, github, planned, referenc assert f'refs/pull/{PR_NUMBER}/merge' in result.output -def test_an_explicit_value_wins_over_the_resolved_one(ddev, github, planned): - result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--base-sha', 'my-own-sha', '--dry-run') +@pytest.mark.parametrize( + 'option, value', + [('--pr-number', '77'), ('--branch', 'a-branch'), ('--base-sha', 'a-sha'), ('--target-branch', 'a-target')], +) +def test_what_a_pull_request_resolves_cannot_also_be_passed(ddev, github, planned, option, value): + """Taking one and ignoring the other would run one pull request's branch against another's diff.""" + result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), option, value, '--dry-run') + + assert result.exit_code == 1 + assert f'`{option}` cannot be passed with `--pr`' in result.output + planned.assert_not_called() + + +def test_a_run_that_dispatches_needs_a_token_before_it_plans(ddev, planned, mocker): + """Planning shells out to git and Hatch for every target, so a missing token must stop it first.""" + mocker.patch.dict('os.environ', {'DD_GITHUB_TOKEN': '', 'GH_TOKEN': '', 'GITHUB_TOKEN': ''}) + + result = ddev('ci', 'dispatch-tests', '--base-sha', 'a-sha') + + assert result.exit_code == 1 + assert 'A GitHub token is required' in result.output + planned.assert_not_called() + + +def test_a_dry_run_reading_no_pull_request_needs_no_token(ddev, planned, mocker): + """The only run that talks to nobody, so it is the one exception to needing a token.""" + mocker.patch.dict('os.environ', {'DD_GITHUB_TOKEN': '', 'GH_TOKEN': '', 'GITHUB_TOKEN': ''}) + + result = ddev('ci', 'dispatch-tests', '--base-sha', 'a-sha', '--dry-run') assert result.exit_code == 0, result.output - assert 'my-own-sha' in result.output - assert 'head-sha-aaa' not in result.output + assert 'Dry run: nothing was dispatched.' in result.output def test_a_dry_run_dispatches_nothing(ddev, github, planned): @@ -72,6 +98,7 @@ def test_a_reference_that_is_neither_a_number_nor_a_url_is_refused(ddev, planned assert result.exit_code == 1 assert 'neither a pull request number nor a pull request URL' in result.output + planned.assert_not_called() def test_an_empty_plan_is_not_dispatched(ddev, fake_async_github, mocker):