From 128b5ced74a49969c35958c386559841c9a3f8f1 Mon Sep 17 00:00:00 2001 From: Tyler Kennedy Date: Sun, 1 Feb 2026 18:02:30 -0500 Subject: [PATCH 1/7] UUID-Everywhere We had partially transitioned to UUIDs in v0.25, this completes the work and adds a regression test and closes #84. Also adds some Executor utilities to the public API for getting and iterating the current job pool. --- chancy/executors/asyncex.py | 3 ++ chancy/executors/base.py | 32 ++++++++++++++ chancy/executors/process.py | 3 +- chancy/job.py | 5 ++- chancy/utils.py | 6 +-- tests/test_issue_84_cancel_type_mismatch.py | 47 +++++++++++++++++++++ 6 files changed, 90 insertions(+), 6 deletions(-) create mode 100644 tests/test_issue_84_cancel_type_mismatch.py diff --git a/chancy/executors/asyncex.py b/chancy/executors/asyncex.py index ec4c33b..72abda9 100644 --- a/chancy/executors/asyncex.py +++ b/chancy/executors/asyncex.py @@ -75,6 +75,9 @@ async def cancel(self, ref: Reference): task.cancel() return + def get_running_jobs(self) -> list["QueuedJob"]: + return list(self.jobs.values()) + async def stop(self): """ Stop the executor, giving it a chance to clean up any resources it diff --git a/chancy/executors/base.py b/chancy/executors/base.py index 325ed89..2eefe80 100644 --- a/chancy/executors/base.py +++ b/chancy/executors/base.py @@ -188,6 +188,35 @@ async def cancel(self, ref: Reference): :param ref: The reference to the job to cancel. """ + def get_running_job(self, ref: Reference) -> QueuedJob | None: + """ + Get a running job by its reference. + + :param ref: The reference to the job to find. + :return: The job if found, None otherwise. + """ + for job in self.get_running_jobs(): + if job.id == ref.identifier: + return job + return None + + @abc.abstractmethod + def get_running_jobs(self) -> list[QueuedJob]: + """ + Get all jobs currently running in this executor. + + :return: A list of running jobs. + """ + + def is_job_running(self, ref: Reference) -> bool: + """ + Check if a job is currently running in this executor. + + :param ref: The reference to the job to check. + :return: True if the job is running, False otherwise. + """ + return self.get_running_job(ref) is not None + @abc.abstractmethod def get_default_concurrency(self) -> int: """ @@ -244,5 +273,8 @@ async def cancel(self, ref: Reference): future.cancel() return + def get_running_jobs(self) -> list[QueuedJob]: + return list(self.jobs.values()) + def __len__(self): return len(self.jobs) diff --git a/chancy/executors/process.py b/chancy/executors/process.py index 6fe8718..89cd327 100644 --- a/chancy/executors/process.py +++ b/chancy/executors/process.py @@ -15,6 +15,7 @@ from asyncio import Future, CancelledError from concurrent.futures import ProcessPoolExecutor from typing import Callable, Any +from uuid import UUID from chancy import Reference from chancy.executors.base import ConcurrentExecutor @@ -138,7 +139,7 @@ async def push(self, job: QueuedJob) -> Future: return future - async def _handle_timeout(self, job_id: str, time_limit: int): + async def _handle_timeout(self, job_id: UUID, time_limit: int): try: await asyncio.sleep(time_limit) pid = self.pids_for_job.get(job_id) diff --git a/chancy/job.py b/chancy/job.py index 8d34819..ea76b65 100644 --- a/chancy/job.py +++ b/chancy/job.py @@ -221,7 +221,7 @@ class State(enum.Enum): SUCCEEDED = "succeeded" #: The unique identifier for this job instance. - id: str + id: UUID #: The time at which this job was created. created_at: datetime #: The time at which this job was started, if it has been started. @@ -237,8 +237,9 @@ class State(enum.Enum): @classmethod def unpack(cls, data: dict) -> "QueuedJob": + id_ = data["id"] return cls( - id=str(data["id"]), + id=id_ if isinstance(id_, UUID) else UUID(id_), func=data["func"], kwargs=data["kwargs"], priority=data["priority"], diff --git a/chancy/utils.py b/chancy/utils.py index 7d236f3..65d63c0 100644 --- a/chancy/utils.py +++ b/chancy/utils.py @@ -121,7 +121,7 @@ def import_string(name): return getattr(module, func_name) -def chancy_uuid() -> str: +def chancy_uuid() -> uuid.UUID: """ Generate a UUID suitable for use as a job ID. @@ -129,12 +129,12 @@ def chancy_uuid() -> str: It's UUID7, kinda, since the draft keeps changing. - :return: str + :return: UUID """ t = (time.time_ns() // 100) & 0xFFFFFFFFFFFFFF rand = secrets.randbits(62) uuid7 = (t << 68) | (7 << 64) | (2 << 62) | rand - return f"{uuid7:032x}" + return uuid.UUID(f"{uuid7:032x}") def json_dumps(obj, **kwargs): diff --git a/tests/test_issue_84_cancel_type_mismatch.py b/tests/test_issue_84_cancel_type_mismatch.py new file mode 100644 index 0000000..901117e --- /dev/null +++ b/tests/test_issue_84_cancel_type_mismatch.py @@ -0,0 +1,47 @@ +""" +Regression test for issue #84 + +In ProcessExecutor, pids_for_job is keyed by job.id (string), but the cancel() +method looked up by ref.identifier (UUID). Since string keys don't match UUID +keys in a dict, the lookup returned None and the SIGUSR1 signal was never sent. + +https://github.com/TkTech/chancy/issues/84 +""" + +import time +import asyncio + +import pytest + +from chancy import Chancy, Worker, Queue, job, QueuedJob + + +@job() +def long_running_job(): + time.sleep(120) + + +@pytest.mark.asyncio +async def test_process_executor_job_cancellation( + chancy: Chancy, worker: Worker +): + """ + Push a job, verify it's running, cancel it, verify it's cancelled. + """ + await chancy.declare(Queue("cancel_test", executor=Chancy.Executor.Process)) + + ref = await chancy.push(long_running_job.job.with_queue("cancel_test")) + j = await chancy.wait_for_job( + ref, timeout=10, states={QueuedJob.State.RUNNING} + ) + assert j.state == j.State.RUNNING + + await chancy.cancel_job(ref) + + executor = worker._executors.get("cancel_test") + async with asyncio.timeout(10): + while executor.is_job_running(ref): + await asyncio.sleep(0.1) + + j = await chancy.wait_for_job(ref, timeout=10) + assert j.state == j.State.FAILED From dba83b0be9af1edb8b9e90ee436c9136ac537ff8 Mon Sep 17 00:00:00 2001 From: Tyler Kennedy Date: Sun, 1 Feb 2026 19:34:49 -0500 Subject: [PATCH 2/7] Prepare for an upcoming 3.16 deprecation in the asyncio module. --- chancy/contrib/django/backend.py | 4 ++-- chancy/executors/asyncex.py | 3 ++- chancy/executors/process.py | 3 ++- chancy/executors/sub.py | 5 +++-- chancy/executors/thread.py | 3 ++- 5 files changed, 11 insertions(+), 7 deletions(-) diff --git a/chancy/contrib/django/backend.py b/chancy/contrib/django/backend.py index 1b14a65..fd80cd8 100644 --- a/chancy/contrib/django/backend.py +++ b/chancy/contrib/django/backend.py @@ -31,7 +31,7 @@ Chancy documentation for details. """ -import asyncio +import inspect from datetime import datetime, timezone from functools import cached_property from typing import TYPE_CHECKING, Any @@ -192,7 +192,7 @@ async def aenqueue( wrapper_func = ( ASYNC_WRAPPER_FUNC - if asyncio.iscoroutinefunction(task.func) + if inspect.iscoroutinefunction(task.func) else WRAPPER_FUNC ) diff --git a/chancy/executors/asyncex.py b/chancy/executors/asyncex.py index 72abda9..7d188c8 100644 --- a/chancy/executors/asyncex.py +++ b/chancy/executors/asyncex.py @@ -1,4 +1,5 @@ import asyncio +import inspect from asyncio import CancelledError from chancy import Reference @@ -47,7 +48,7 @@ def __len__(self): async def _job_wrapper(self, job: QueuedJob): try: func, kwargs = Executor.get_function_and_kwargs(job) - if not asyncio.iscoroutinefunction(func): + if not inspect.iscoroutinefunction(func): raise ValueError( f"Function {job.func!r} is not an async function, which is" f" required for the AsyncExecutor. Please use the" diff --git a/chancy/executors/process.py b/chancy/executors/process.py index 89cd327..c75c3d8 100644 --- a/chancy/executors/process.py +++ b/chancy/executors/process.py @@ -1,5 +1,6 @@ import asyncio import functools +import inspect import multiprocessing import os import warnings @@ -191,7 +192,7 @@ def job_wrapper(cls, job: QueuedJob, pids_for_job) -> tuple[QueuedJob, Any]: ) ) - if asyncio.iscoroutinefunction(func): + if inspect.iscoroutinefunction(func): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: diff --git a/chancy/executors/sub.py b/chancy/executors/sub.py index 6bb91f3..7f91df7 100644 --- a/chancy/executors/sub.py +++ b/chancy/executors/sub.py @@ -1,5 +1,6 @@ -import os import asyncio +import inspect +import os import threading import functools from concurrent.futures import Future @@ -122,7 +123,7 @@ def job_wrapper(cls, job: QueuedJob) -> tuple[QueuedJob, Any]: timer.start() try: - if asyncio.iscoroutinefunction(func): + if inspect.iscoroutinefunction(func): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: diff --git a/chancy/executors/thread.py b/chancy/executors/thread.py index df5ee74..e7575e2 100644 --- a/chancy/executors/thread.py +++ b/chancy/executors/thread.py @@ -1,3 +1,4 @@ +import inspect import os import threading from concurrent.futures import ThreadPoolExecutor, Future @@ -77,7 +78,7 @@ def job_wrapper(self, job: QueuedJob) -> tuple[QueuedJob, Any]: timer.start() try: - if asyncio.iscoroutinefunction(func): + if inspect.iscoroutinefunction(func): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: From 8c49f347134ff2f9b55576068eb9a1e7de277218 Mon Sep 17 00:00:00 2001 From: Tyler Kennedy Date: Sun, 1 Feb 2026 19:35:17 -0500 Subject: [PATCH 3/7] The Mac OS CI/CD runners are very unreliable, 5x the very tight timing test in the test_scale regression catcher. --- tests/test_scale.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/test_scale.py b/tests/test_scale.py index 378002f..5d28d69 100644 --- a/tests/test_scale.py +++ b/tests/test_scale.py @@ -83,10 +83,14 @@ async def test_busy_chancy(chancy: Chancy, worker_no_start: Worker): await conn.commit() + # In reality this is typically sub-0.1. However, we get occasional + # spikes in CI, especially on OS X, that make this very noisy for test + # failures, so we set a more generous limit to just catch egregious + # performance regressions. with timed_block() as timer: await worker_no_start.fetch_jobs(queue, conn, up_to=1) - assert timer.elapsed < 0.1 + assert timer.elapsed < 0.5 with timed_block() as timer: await worker_no_start.fetch_jobs(queue, conn, up_to=100) - assert timer.elapsed < 0.1 + assert timer.elapsed < 0.5 From aa52319cb6888845f46ef42a47b339fe91da399e Mon Sep 17 00:00:00 2001 From: Tyler Kennedy Date: Sun, 1 Feb 2026 19:59:25 -0500 Subject: [PATCH 4/7] Cover async executor in the regression test as well, since it also supports cancellation. --- tests/test_issue_84_cancel_type_mismatch.py | 33 +++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/tests/test_issue_84_cancel_type_mismatch.py b/tests/test_issue_84_cancel_type_mismatch.py index 901117e..5deb473 100644 --- a/tests/test_issue_84_cancel_type_mismatch.py +++ b/tests/test_issue_84_cancel_type_mismatch.py @@ -21,6 +21,11 @@ def long_running_job(): time.sleep(120) +@job() +async def async_long_running_job(): + await asyncio.sleep(120) + + @pytest.mark.asyncio async def test_process_executor_job_cancellation( chancy: Chancy, worker: Worker @@ -45,3 +50,31 @@ async def test_process_executor_job_cancellation( j = await chancy.wait_for_job(ref, timeout=10) assert j.state == j.State.FAILED + + +@pytest.mark.asyncio +async def test_async_executor_job_cancellation(chancy: Chancy, worker: Worker): + """ + Push an async job, verify it's running, cancel it, verify it's cancelled. + """ + await chancy.declare( + Queue("async_cancel_test", executor=Chancy.Executor.Async) + ) + + ref = await chancy.push( + async_long_running_job.job.with_queue("async_cancel_test") + ) + j = await chancy.wait_for_job( + ref, timeout=10, states={QueuedJob.State.RUNNING} + ) + assert j.state == j.State.RUNNING + + await chancy.cancel_job(ref) + + executor = worker._executors.get("async_cancel_test") + async with asyncio.timeout(10): + while executor.is_job_running(ref): + await asyncio.sleep(0.1) + + j = await chancy.wait_for_job(ref, timeout=10) + assert j.state == j.State.FAILED From 4ac836fba37c84b2773c11cd7691f7092bae1c6e Mon Sep 17 00:00:00 2001 From: Tyler Kennedy Date: Sun, 1 Feb 2026 22:37:51 -0500 Subject: [PATCH 5/7] Collapse regression test into old, flawed test. Try to improve performance of two slow tests. --- tests/test_issue_84_cancel_type_mismatch.py | 80 --------------------- tests/test_jobs.py | 78 +++++++++++++------- tests/test_worker.py | 7 +- 3 files changed, 57 insertions(+), 108 deletions(-) delete mode 100644 tests/test_issue_84_cancel_type_mismatch.py diff --git a/tests/test_issue_84_cancel_type_mismatch.py b/tests/test_issue_84_cancel_type_mismatch.py deleted file mode 100644 index 5deb473..0000000 --- a/tests/test_issue_84_cancel_type_mismatch.py +++ /dev/null @@ -1,80 +0,0 @@ -""" -Regression test for issue #84 - -In ProcessExecutor, pids_for_job is keyed by job.id (string), but the cancel() -method looked up by ref.identifier (UUID). Since string keys don't match UUID -keys in a dict, the lookup returned None and the SIGUSR1 signal was never sent. - -https://github.com/TkTech/chancy/issues/84 -""" - -import time -import asyncio - -import pytest - -from chancy import Chancy, Worker, Queue, job, QueuedJob - - -@job() -def long_running_job(): - time.sleep(120) - - -@job() -async def async_long_running_job(): - await asyncio.sleep(120) - - -@pytest.mark.asyncio -async def test_process_executor_job_cancellation( - chancy: Chancy, worker: Worker -): - """ - Push a job, verify it's running, cancel it, verify it's cancelled. - """ - await chancy.declare(Queue("cancel_test", executor=Chancy.Executor.Process)) - - ref = await chancy.push(long_running_job.job.with_queue("cancel_test")) - j = await chancy.wait_for_job( - ref, timeout=10, states={QueuedJob.State.RUNNING} - ) - assert j.state == j.State.RUNNING - - await chancy.cancel_job(ref) - - executor = worker._executors.get("cancel_test") - async with asyncio.timeout(10): - while executor.is_job_running(ref): - await asyncio.sleep(0.1) - - j = await chancy.wait_for_job(ref, timeout=10) - assert j.state == j.State.FAILED - - -@pytest.mark.asyncio -async def test_async_executor_job_cancellation(chancy: Chancy, worker: Worker): - """ - Push an async job, verify it's running, cancel it, verify it's cancelled. - """ - await chancy.declare( - Queue("async_cancel_test", executor=Chancy.Executor.Async) - ) - - ref = await chancy.push( - async_long_running_job.job.with_queue("async_cancel_test") - ) - j = await chancy.wait_for_job( - ref, timeout=10, states={QueuedJob.State.RUNNING} - ) - assert j.state == j.State.RUNNING - - await chancy.cancel_job(ref) - - executor = worker._executors.get("async_cancel_test") - async with asyncio.timeout(10): - while executor.is_job_running(ref): - await asyncio.sleep(0.1) - - j = await chancy.wait_for_job(ref, timeout=10) - assert j.state == j.State.FAILED diff --git a/tests/test_jobs.py b/tests/test_jobs.py index fa53760..10b9838 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -3,7 +3,7 @@ import pytest -from chancy import Chancy, Worker, Queue, QueuedJob, Reference, job, Job +from chancy import Chancy, Worker, Queue, QueuedJob, job, Job from test_worker import job_that_fails @@ -171,30 +171,6 @@ async def test_async_job_instance_kwarg_on_sync_executor( assert j.meta.get("received_instance") is True -@pytest.mark.asyncio -async def test_job_cancellation(chancy: Chancy, worker: Worker): - """ - Test that jobs can be cancelled on supporting executors. - """ - - async def cancel_in_a_bit(to_cancel: Reference): - await asyncio.sleep(10) - await chancy.cancel_job(to_cancel) - - await chancy.declare(Queue("async", executor=Chancy.Executor.Async)) - await chancy.declare(Queue("sync", executor=Chancy.Executor.Process)) - - ref = await chancy.push(very_long_job.job.with_queue("async")) - asyncio.create_task(cancel_in_a_bit(ref)) - j = await chancy.wait_for_job(ref, timeout=30) - assert j.state == j.State.FAILED - - ref = await chancy.push(sync_very_long_job.job.with_queue("sync")) - asyncio.create_task(cancel_in_a_bit(ref)) - j = await chancy.wait_for_job(ref, timeout=30) - assert j.state == j.State.FAILED - - @pytest.mark.asyncio async def test_job_signature_with_kwarg_marker(chancy, worker): """ @@ -315,3 +291,55 @@ async def test_purge_jobs(chancy: Chancy, worker: Worker, sync_executor: str): await chancy.purge_jobs([ref_ok, ref_fail]) assert await chancy.get_job(ref_ok) is None assert await chancy.get_job(ref_fail) is None + + +@pytest.mark.asyncio +async def test_process_executor_job_cancellation( + chancy: Chancy, worker: Worker +): + """ + Test that jobs can be cancelled on the ProcessExecutor. + """ + await chancy.declare(Queue("cancel_test", executor=Chancy.Executor.Process)) + + ref = await chancy.push(sync_very_long_job.job.with_queue("cancel_test")) + j = await chancy.wait_for_job( + ref, timeout=10, states={QueuedJob.State.RUNNING} + ) + assert j.state == j.State.RUNNING + + await chancy.cancel_job(ref) + + executor = worker._executors.get("cancel_test") + async with asyncio.timeout(10): + while executor.is_job_running(ref): + await asyncio.sleep(0.1) + + j = await chancy.wait_for_job(ref, timeout=10) + assert j.state == j.State.FAILED + + +@pytest.mark.asyncio +async def test_async_executor_job_cancellation(chancy: Chancy, worker: Worker): + """ + Test that jobs can be cancelled on the AsyncExecutor. + """ + await chancy.declare( + Queue("async_cancel_test", executor=Chancy.Executor.Async) + ) + + ref = await chancy.push(very_long_job.job.with_queue("async_cancel_test")) + j = await chancy.wait_for_job( + ref, timeout=10, states={QueuedJob.State.RUNNING} + ) + assert j.state == j.State.RUNNING + + await chancy.cancel_job(ref) + + executor = worker._executors.get("async_cancel_test") + async with asyncio.timeout(10): + while executor.is_job_running(ref): + await asyncio.sleep(0.1) + + j = await chancy.wait_for_job(ref, timeout=10) + assert j.state == j.State.FAILED diff --git a/tests/test_worker.py b/tests/test_worker.py index 484c562..b6f093e 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -93,8 +93,10 @@ async def test_queue_removal(chancy: Chancy, worker: Worker): await chancy.delete_queue("test_removal", purge_jobs=True) await worker.hub.wait_for("worker.queue.removed", timeout=30) - # Give the executor time to clean up - await asyncio.sleep(5) + # Wait for the executor to clean up + async with asyncio.timeout(10): + while "test_removal" in worker.executors: + await asyncio.sleep(0.1) assert "test_removal" not in worker.executors @@ -119,7 +121,6 @@ async def test_immediate_processing(chancy: Chancy, worker: Worker): """ await chancy.declare(Queue("test_immediate", polling_interval=60)) await worker.hub.wait_for("worker.queue.started") - await asyncio.sleep(5) j = await chancy.push(job_to_run.job.with_queue("test_immediate")) From 49fb81cc0943feed257e99232e0adaa3078de7c1 Mon Sep 17 00:00:00 2001 From: Tyler Kennedy Date: Mon, 2 Feb 2026 07:16:00 -0500 Subject: [PATCH 6/7] Parallelize the test runners, and fix a constraint missing a table prefix we found along the way. --- .github/workflows/release.yml | 8 ++- chancy/migrations/v1.py | 9 +++- chancy/migrations/v7.py | 62 +++++++++++++++++++++++ pyproject.toml | 1 + tests/conftest.py | 48 ++++++++++++++++-- tests/contrib/django/test_connection.py | 4 +- tests/contrib/django/test_django_tasks.py | 4 +- tests/contrib/django/test_models.py | 12 +++++ tests/plugins/test_trigger.py | 30 ++++++----- 9 files changed, 154 insertions(+), 24 deletions(-) create mode 100644 chancy/migrations/v7.py diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 523bc4c..3942934 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -96,9 +96,13 @@ jobs: - name: Installing dependencies (Python) run: uv sync --all-extras - - name: Running tests + - name: Running tests (parallel) run: | - uv run pytest -s -v --cov=chancy --cov-report=xml + uv run pytest -n auto -v --cov=chancy --cov-report=xml + + - name: Running Django ORM tests (sequential) + run: | + uv run pytest tests/contrib/django/test_models.py -v --cov=chancy --cov-report=xml --cov-append - name: Uploading coverage uses: codecov/codecov-action@v4 diff --git a/chancy/migrations/v1.py b/chancy/migrations/v1.py index ab0c3f9..76048e1 100644 --- a/chancy/migrations/v1.py +++ b/chancy/migrations/v1.py @@ -60,10 +60,15 @@ async def up(self, migrator: Migrator, cursor: AsyncCursor[DictRow]): expires_at TIMESTAMPTZ NOT NULL ); ALTER TABLE {leader} - ADD CONSTRAINT leader_worker_id_unique UNIQUE + ADD CONSTRAINT {leader_worker_id_unique} UNIQUE (worker_id); """ - ).format(leader=sql.Identifier(f"{migrator.prefix}leader")) + ).format( + leader=sql.Identifier(f"{migrator.prefix}leader"), + leader_worker_id_unique=sql.Identifier( + f"{migrator.prefix}leader_worker_id_unique" + ), + ) ) await cursor.execute( diff --git a/chancy/migrations/v7.py b/chancy/migrations/v7.py new file mode 100644 index 0000000..e0fc6f2 --- /dev/null +++ b/chancy/migrations/v7.py @@ -0,0 +1,62 @@ +from psycopg import AsyncCursor, sql +from psycopg.rows import DictRow + +from chancy.migrate import Migration, Migrator + + +class V7Migration(Migration): + async def up(self, migrator: Migrator, cursor: AsyncCursor[DictRow]): + """ + Rename the leader_worker_id_unique constraint to use the prefix. + + This fixes an oversight where the constraint name was hardcoded + instead of using the configurable prefix, which prevented running + multiple Chancy instances with different prefixes in the same database. + """ + old_name = "leader_worker_id_unique" + new_name = f"{migrator.prefix}leader_worker_id_unique" + + # Check if the old unprefixed constraint exists + await cursor.execute( + """ + SELECT 1 FROM pg_constraint + WHERE conname = %s + """, + (old_name,), + ) + if await cursor.fetchone(): + await cursor.execute( + sql.SQL( + "ALTER TABLE {table} RENAME CONSTRAINT {old} TO {new}" + ).format( + table=sql.Identifier(f"{migrator.prefix}leader"), + old=sql.Identifier(old_name), + new=sql.Identifier(new_name), + ) + ) + + async def down(self, migrator: Migrator, cursor: AsyncCursor[DictRow]): + """ + Rename the constraint back to the unprefixed name. + """ + old_name = f"{migrator.prefix}leader_worker_id_unique" + new_name = "leader_worker_id_unique" + + # Check if the prefixed constraint exists + await cursor.execute( + """ + SELECT 1 FROM pg_constraint + WHERE conname = %s + """, + (old_name,), + ) + if await cursor.fetchone(): + await cursor.execute( + sql.SQL( + "ALTER TABLE {table} RENAME CONSTRAINT {old} TO {new}" + ).format( + table=sql.Identifier(f"{migrator.prefix}leader"), + old=sql.Identifier(old_name), + new=sql.Identifier(new_name), + ) + ) diff --git a/pyproject.toml b/pyproject.toml index 739d9ad..0c29bf5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,6 +57,7 @@ dev = [ "pytest-django>=4.8.0", "pre-commit>=4.3.0", "ruff>=0.14.2", + "pytest-xdist>=3.8.0", ] [project.scripts] diff --git a/tests/conftest.py b/tests/conftest.py index 52b1828..c2cc62b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,5 @@ import asyncio +import os from typing import AsyncIterator import pytest @@ -8,6 +9,36 @@ from chancy import Chancy, Worker +@pytest.fixture(scope="session") +def xdist_worker() -> str | None: + """Returns the xdist worker ID (e.g., 'gw0') or None if not running parallel.""" + return os.environ.get("PYTEST_XDIST_WORKER") + + +@pytest.fixture(scope="session") +def test_prefix(xdist_worker) -> str: + """ + Unique table prefix for parallel test isolation. + + Returns 'chancy_' normally, or 'chancy_gw0_' etc. when running with xdist. + """ + if xdist_worker is None: + return "chancy_" + return f"chancy_{xdist_worker}_" + + +@pytest.fixture(scope="session") +def test_suffix(xdist_worker) -> str: + """ + Unique suffix for test resources (tables, schemas) that need isolation. + + Returns '' normally, or '_gw0' etc. when running with xdist. + """ + if xdist_worker is None: + return "" + return f"_{xdist_worker}" + + @pytest.fixture(scope="session") def event_loop_policy(): # Since psycopg's asyncio implementation cannot use the default @@ -18,14 +49,22 @@ def event_loop_policy(): @pytest_asyncio.fixture() -async def chancy(request): +async def chancy(request, test_prefix): """ Provides a Chancy application instance with an open connection pool to the test database. + + When running with pytest-xdist, each worker gets a unique table prefix + to enable parallel test execution without conflicts. """ + params = getattr(request, "param", {}) + # Allow tests to override prefix, but default to worker-specific prefix + if "prefix" not in params: + params = {**params, "prefix": test_prefix} + async with Chancy( "postgresql://postgres:localtest@localhost:8190/postgres", - **getattr(request, "param", {}), + **params, ) as chancy: await chancy.migrate() yield chancy @@ -33,13 +72,16 @@ async def chancy(request): @pytest.fixture -def chancy_just_app(): +def chancy_just_app(test_prefix): """ Provides just a configured chancy instance with no open connection pool or migrations. + + When running with pytest-xdist, uses a unique table prefix for isolation. """ return Chancy( "postgresql://postgres:localtest@localhost:8190/postgres", + prefix=test_prefix, ) diff --git a/tests/contrib/django/test_connection.py b/tests/contrib/django/test_connection.py index 7f6f484..6110e2b 100644 --- a/tests/contrib/django/test_connection.py +++ b/tests/contrib/django/test_connection.py @@ -5,7 +5,7 @@ @pytest.mark.django_db @pytest.mark.asyncio -async def test_django_style_connection(settings): - async with Chancy(settings.DATABASES["default"]) as app: +async def test_django_style_connection(settings, test_prefix): + async with Chancy(settings.DATABASES["default"], prefix=test_prefix) as app: await app.migrate() await app.migrate(to_version=0) diff --git a/tests/contrib/django/test_django_tasks.py b/tests/contrib/django/test_django_tasks.py index 812353f..4d179b7 100644 --- a/tests/contrib/django/test_django_tasks.py +++ b/tests/contrib/django/test_django_tasks.py @@ -55,7 +55,7 @@ async def async_add(a: int, b: int) -> int: @pytest.fixture -def django_tasks_settings(settings): +def django_tasks_settings(settings, chancy): """Configure Django Tasks to use the Chancy backend.""" settings.TASKS = { "default": { @@ -63,7 +63,7 @@ def django_tasks_settings(settings): "QUEUES": ["default", "async_queue"], "OPTIONS": { "dsn": "postgresql://postgres:localtest@localhost:8190/postgres", - "prefix": "chancy_", + "prefix": chancy.prefix, }, }, } diff --git a/tests/contrib/django/test_models.py b/tests/contrib/django/test_models.py index c721f5b..2a8cb6d 100644 --- a/tests/contrib/django/test_models.py +++ b/tests/contrib/django/test_models.py @@ -1,12 +1,24 @@ """ Tests for the Django models integration. + +Note: These tests cannot run in parallel (pytest-xdist) because Django ORM +models have table names set at import time, not runtime. """ +import os import pytest from chancy import job +# Skip these tests when running with pytest-xdist since Django model +# table names are set at import time and can't be made worker-specific +pytestmark = pytest.mark.skipif( + os.environ.get("PYTEST_XDIST_WORKER") is not None, + reason="Django ORM models have static table names incompatible with parallel tests", +) + + @job() def test_job(): pass diff --git a/tests/plugins/test_trigger.py b/tests/plugins/test_trigger.py index b767727..f1b4c64 100644 --- a/tests/plugins/test_trigger.py +++ b/tests/plugins/test_trigger.py @@ -12,7 +12,7 @@ def handle_insert(*, j: QueuedJob): """Test job for INSERT operations""" assert j.meta["trigger"]["operation"] == "INSERT" - assert j.meta["trigger"]["table_name"] == "test_users" + assert j.meta["trigger"]["table_name"].startswith("test_users") assert j.meta["trigger"]["schema_name"] == "public" @@ -37,7 +37,7 @@ def handle_any_change(*, j: QueuedJob): @job(queue="trigger_events", priority=10) def handle_with_priority(*, j: QueuedJob): """Test job with priority""" - assert j.meta["trigger"]["table_name"] == "test_users" + assert j.meta["trigger"]["table_name"].startswith("test_users") # Keep old name for backward compatibility with existing test @@ -45,8 +45,9 @@ def handle_with_priority(*, j: QueuedJob): @pytest_asyncio.fixture -async def test_table(chancy): - table_name = "test_users" +async def test_table(chancy, test_suffix): + # Use unique table name per xdist worker to avoid conflicts + table_name = f"test_users{test_suffix}" async with chancy.pool.connection() as conn: async with conn.cursor() as cursor: @@ -65,7 +66,7 @@ async def test_table(chancy): async with chancy.pool.connection() as conn: async with conn.cursor() as cursor: await cursor.execute( - sql.SQL("DROP TABLE IF EXISTS {table}").format( + sql.SQL("DROP TABLE IF EXISTS {table} CASCADE").format( table=sql.Identifier(table_name) ) ) @@ -710,9 +711,11 @@ async def test_trigger_disabled_at_registration( @pytest_asyncio.fixture -async def test_schema(chancy): +async def test_schema(chancy, test_suffix): """Create a test schema for schema-specific tests""" - schema_name = "test_schema" + # Use unique schema/table names per xdist worker + schema_name = f"test_schema{test_suffix}" + table_name = f"test_users{test_suffix}" async with chancy.pool.connection() as conn: async with conn.cursor() as cursor: @@ -730,11 +733,11 @@ async def test_schema(chancy): ) """).format( schema=sql.Identifier(schema_name), - table=sql.Identifier("test_users"), + table=sql.Identifier(table_name), ) ) - yield schema_name + yield (schema_name, table_name) async with chancy.pool.connection() as conn: async with conn.cursor() as cursor: @@ -753,14 +756,15 @@ async def test_schema(chancy): @pytest.mark.asyncio async def test_trigger_different_schema(chancy: Chancy, worker, test_schema): """Test trigger on a table in a non-public schema""" + schema_name, table_name = test_schema await chancy.declare(Queue("trigger_events")) await Trigger.register_trigger( chancy, - table_name="test_users", + table_name=table_name, operations=["INSERT"], job_template=handle_insert, - schema_name=test_schema, + schema_name=schema_name, ) async with chancy.pool.connection() as conn: @@ -771,8 +775,8 @@ async def test_trigger_different_schema(chancy: Chancy, worker, test_schema): INSERT INTO {schema}.{table} (name, email) VALUES ('Schema User', 'schema@example.com') """).format( - schema=sql.Identifier(test_schema), - table=sql.Identifier("test_users"), + schema=sql.Identifier(schema_name), + table=sql.Identifier(table_name), ) ) From 91c03e0ff3d0a9fe3ba25a7d1ed205f4d747b839 Mon Sep 17 00:00:00 2001 From: Tyler Kennedy Date: Mon, 2 Feb 2026 21:14:46 -0500 Subject: [PATCH 7/7] Skip process cancellation on Windows (no signals) --- tests/test_jobs.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/test_jobs.py b/tests/test_jobs.py index 10b9838..a1869b3 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -1,3 +1,4 @@ +import sys import time import asyncio @@ -294,6 +295,9 @@ async def test_purge_jobs(chancy: Chancy, worker: Worker, sync_executor: str): @pytest.mark.asyncio +@pytest.mark.skipif( + sys.platform == "win32", reason="SIGUSR1 not available on Windows" +) async def test_process_executor_job_cancellation( chancy: Chancy, worker: Worker ):