Uses testcontainers to provide a redis instance for the unit tests

.github/workflows/ci.yml

@@ -67,16 +67,10 @@ jobs:
  env:
  PYTHON: ${{ matrix.python }}
  OS: ${{ matrix.os }}
+ ARQ_TEST_REDIS_VERSION: ${{ matrix.redis }}
 
  runs-on: ${{ matrix.os }}-latest
 
- services:
- redis:
- image: redis:${{ matrix.redis }}
- ports:
- - 6379:6379
- options: --entrypoint redis-server
-
  steps:
  - uses: actions/checkout@v4
 

.gitignore

@@ -18,3 +18,4 @@ __pycache__/
 .venv/
 /.auto-format
 /scratch/
+.python-version

requirements/docs.txt

@@ -8,13 +8,13 @@ alabaster==0.7.16
  # via sphinx
 babel==2.14.0
  # via sphinx
-certifi==2024.2.2
+certifi==2024.7.4
  # via requests
 charset-normalizer==3.3.2
  # via requests
 docutils==0.19
  # via sphinx
-idna==3.6
+idna==3.7
  # via requests
 imagesize==1.4.1
  # via sphinx
@@ -26,7 +26,7 @@ packaging==24.0
  # via sphinx
 pygments==2.17.2
  # via sphinx
-requests==2.31.0
+requests==2.32.3
  # via sphinx
 snowballstemmer==2.2.0
  # via sphinx
@@ -44,5 +44,5 @@ sphinxcontrib-qthelp==1.0.7
  # via sphinx
 sphinxcontrib-serializinghtml==1.1.10
  # via sphinx
-urllib3==2.2.1
+urllib3==2.2.2
  # via requests

requirements/pyproject.txt

@@ -10,7 +10,7 @@ click==8.1.7
  # via arq (pyproject.toml)
 hiredis==2.3.2
  # via redis
-idna==3.6
+idna==3.7
  # via anyio
 redis==4.6.0
  # via arq (pyproject.toml)

requirements/testing.in

@@ -8,3 +8,4 @@ pytest-mock
 pytest-pretty
 pytest-timeout
 pytz
+testcontainers<4 # until we remove 3.8 support

requirements/testing.txt

@@ -1,15 +1,27 @@
 #
-# This file is autogenerated by pip-compile with Python 3.12
+# This file is autogenerated by pip-compile with Python 3.9
 # by the following command:
 #
 # pip-compile --output-file=requirements/testing.txt --strip-extras requirements/testing.in
 #
 annotated-types==0.6.0
  # via pydantic
+certifi==2024.7.4
+ # via requests
+charset-normalizer==3.3.2
+ # via requests
 coverage==7.4.4
  # via -r requirements/testing.in
+deprecation==2.1.0
+ # via testcontainers
 dirty-equals==0.7.1.post0
  # via -r requirements/testing.in
+docker==7.1.0
+ # via testcontainers
+exceptiongroup==1.2.2
+ # via pytest
+idna==3.7
+ # via requests
 iniconfig==2.0.0
  # via pytest
 markdown-it-py==3.0.0
@@ -19,7 +31,9 @@ mdurl==0.1.2
 msgpack==1.0.8
  # via -r requirements/testing.in
 packaging==24.0
- # via pytest
+ # via
+ # deprecation
+ # pytest
 pluggy==1.4.0
  # via pytest
 pydantic==2.6.4
@@ -47,9 +61,23 @@ pytz==2024.1
  # via
  # -r requirements/testing.in
  # dirty-equals
+requests==2.32.3
+ # via docker
 rich==13.7.1
  # via pytest-pretty
+testcontainers==3.7.1
+ # via -r requirements/testing.in
+tomli==2.0.1
+ # via
+ # coverage
+ # pytest
 typing-extensions==4.10.0
  # via
  # pydantic
  # pydantic-core
+urllib3==2.2.2
+ # via
+ # docker
+ # requests
+wrapt==1.16.0
+ # via testcontainers

tests/conftest.py

@@ -2,27 +2,55 @@
 import functools
 import os
 import sys
+from typing import Generator
 
 import msgpack
 import pytest
 import redis.exceptions
 from redis.asyncio.retry import Retry
 from redis.backoff import NoBackoff
+from testcontainers.redis import RedisContainer
 
-from arq.connections import ArqRedis, create_pool
+from arq.connections import ArqRedis, RedisSettings, create_pool
 from arq.worker import Worker
 
 
 @pytest.fixture(name='loop')
-def _fix_loop(event_loop):
+def _fix_loop(event_loop: asyncio.AbstractEventLoop) -> asyncio.AbstractEventLoop:
  return event_loop
 
 
+@pytest.fixture(scope='session')
+def redis_version() -> str:
+ return os.getenv('ARQ_TEST_REDIS_VERSION', 'latest')
+
+
+@pytest.fixture(scope='session')
+def redis_container(redis_version: str) -> Generator[RedisContainer, None, None]:
+ with RedisContainer(f'redis:{redis_version}') as redis:
+ yield redis
+
+
+@pytest.fixture(scope='session')
+def test_redis_host(redis_container: RedisContainer) -> str:
+ return redis_container.get_container_host_ip()
+
+
+@pytest.fixture(scope='session')
+def test_redis_port(redis_container: RedisContainer) -> int:
+ return redis_container.get_exposed_port(redis_container.port_to_expose)
+
+
+@pytest.fixture(scope='session')
+def test_redis_settings(test_redis_host: str, test_redis_port: int) -> RedisSettings:
+ return RedisSettings(host=test_redis_host, port=test_redis_port)
+
+
 @pytest.fixture
-async def arq_redis(loop):
+async def arq_redis(test_redis_host: str, test_redis_port: int):
  redis_ = ArqRedis(
- host='localhost',
- port=6379,
+ host=test_redis_host,
+ port=test_redis_port,
  encoding='utf-8',
  )
 
@@ -34,10 +62,10 @@ async def arq_redis(loop):
 
 
 @pytest.fixture
-async def arq_redis_msgpack(loop):
+async def arq_redis_msgpack(test_redis_host: str, test_redis_port: int):
  redis_ = ArqRedis(
- host='localhost',
- port=6379,
+ host=test_redis_host,
+ port=test_redis_port,
  encoding='utf-8',
  job_serializer=msgpack.packb,
  job_deserializer=functools.partial(msgpack.unpackb, raw=False),
@@ -48,10 +76,10 @@ async def arq_redis_msgpack(loop):
 
 
 @pytest.fixture
-async def arq_redis_retry(loop):
+async def arq_redis_retry(test_redis_host: str, test_redis_port: int):
  redis_ = ArqRedis(
- host='localhost',
- port=6379,
+ host=test_redis_host,
+ port=test_redis_port,
  encoding='utf-8',
  retry=Retry(backoff=NoBackoff(), retries=3),
  retry_on_timeout=True,

tests/test_cli.py

@@ -3,6 +3,7 @@
 
 from arq import logs
 from arq.cli import cli
+from arq.connections import RedisSettings
 
 
 async def foobar(ctx):
@@ -14,6 +15,11 @@ class WorkerSettings:
  functions = [foobar]
 
 
+@pytest.fixture(scope='module', autouse=True)
+def setup_worker_connection(test_redis_host: str, test_redis_port: int):
+ WorkerSettings.redis_settings = RedisSettings(host=test_redis_host, port=test_redis_port)
+
+
 def test_help():
  runner = CliRunner()
  result = runner.invoke(cli, ['--help'])

tests/test_jobs.py

@@ -114,9 +114,9 @@ async def test_enqueue_job_alt_queue(arq_redis: ArqRedis, worker):
  await test_enqueue_job(arq_redis, worker, queue_name='custom_queue')
 
 
-async def test_enqueue_job_nondefault_queue(worker):
+async def test_enqueue_job_nondefault_queue(test_redis_settings: RedisSettings, worker):
  """Test initializing arq_redis with a queue name, and the worker using it."""
- arq_redis = await create_pool(RedisSettings(), default_queue_name='test_queue')
+ arq_redis = await create_pool(test_redis_settings, default_queue_name='test_queue')
  await test_enqueue_job(
  arq_redis,
  lambda functions, **_: worker(functions=functions, arq_redis=arq_redis, queue_name=None),

tests/test_main.py

@@ -11,7 +11,7 @@
 import pytest
 from dirty_equals import IsInt, IsNow
 
-from arq.connections import ArqRedis
+from arq.connections import ArqRedis, RedisSettings
 from arq.constants import default_queue_name
 from arq.jobs import Job, JobDef, SerializationError
 from arq.utils import timestamp_ms
@@ -65,7 +65,9 @@ async def parent_job(ctx):
  assert inner_result == 42
 
 
-async def test_enqueue_job_nested_custom_serializer(arq_redis_msgpack: ArqRedis, worker):
+async def test_enqueue_job_nested_custom_serializer(
+ arq_redis_msgpack: ArqRedis, test_redis_settings: RedisSettings, worker
+):
  async def foobar(ctx):
  return 42
 
@@ -78,6 +80,7 @@ async def parent_job(ctx):
  worker: Worker = worker(
  functions=[func(parent_job, name='parent_job'), func(foobar, name='foobar')],
  arq_redis=None,
+ redis_settings=test_redis_settings,
  job_serializer=msgpack.packb,
  job_deserializer=functools.partial(msgpack.unpackb, raw=False),
  )
@@ -90,7 +93,7 @@ async def parent_job(ctx):
  assert inner_result == 42
 
 
-async def test_enqueue_job_custom_queue(arq_redis: ArqRedis, worker):
+async def test_enqueue_job_custom_queue(arq_redis: ArqRedis, test_redis_settings: RedisSettings, worker):
  async def foobar(ctx):
  return 42
 
@@ -103,6 +106,7 @@ async def parent_job(ctx):
  worker: Worker = worker(
  functions=[func(parent_job, name='parent_job'), func(foobar, name='foobar')],
  arq_redis=None,
+ redis_settings=test_redis_settings,
  queue_name='spanner',
  )
 

tests/test_utils.py

@@ -55,20 +55,19 @@ async def test_redis_sentinel_failure(create_pool, cancel_remaining_task, mocker
  await create_pool(settings)
 
 
-async def test_redis_success_log(caplog, create_pool):
+async def test_redis_success_log(test_redis_settings: RedisSettings, caplog, create_pool):
  caplog.set_level(logging.INFO)
- settings = RedisSettings()
- pool = await create_pool(settings)
+ pool = await create_pool(test_redis_settings)
  assert 'redis connection successful' not in [r.message for r in caplog.records]
  await pool.close(close_connection_pool=True)
 
- pool = await create_pool(settings, retry=1)
+ pool = await create_pool(test_redis_settings, retry=1)
  assert 'redis connection successful' in [r.message for r in caplog.records]
  await pool.close(close_connection_pool=True)
 
 
-async def test_redis_log(create_pool):
- redis = await create_pool(RedisSettings())
+async def test_redis_log(test_redis_settings: RedisSettings, create_pool):
+ redis = await create_pool(test_redis_settings)
  await redis.flushall()
  await redis.set(b'a', b'1')
  await redis.set(b'b', b'2')
@@ -110,7 +109,7 @@ def test_typing():
  assert 'OptionType' in arq.typing.__all__
 
 
-def test_redis_settings_validation():
+def redis_settings_validation():
  class Settings(BaseModel, arbitrary_types_allowed=True):
  redis_settings: RedisSettings
 

tests/test_worker.py

@@ -35,12 +35,13 @@ async def fails(ctx):
  raise TypeError('my type error')
 
 
-def test_no_jobs(arq_redis: ArqRedis, loop, mocker):
+def test_no_jobs(test_redis_settings: RedisSettings, arq_redis: ArqRedis, loop, mocker):
  class Settings:
  functions = [func(foobar, name='foobar')]
  burst = True
  poll_delay = 0
  queue_read_limit = 10
+ redis_settings = test_redis_settings
 
  loop.run_until_complete(arq_redis.enqueue_job('foobar'))
  mocker.patch('asyncio.get_event_loop', lambda: loop)
@@ -49,21 +50,21 @@ class Settings:
  assert str(worker) == '<Worker j_complete=1 j_failed=0 j_retried=0 j_ongoing=0>'
 
 
-def test_health_check_direct(loop):
+def test_health_check_direct(test_redis_settings: RedisSettings, loop):
  class Settings:
- pass
+ redis_settings = test_redis_settings
 
  asyncio.set_event_loop(loop)
  assert check_health(Settings) == 1
 
 
-async def test_health_check_fails():
- assert 1 == await async_check_health(None)
+async def test_health_check_fails(test_redis_settings: RedisSettings):
+ assert 1 == await async_check_health(test_redis_settings)
 
 
-async def test_health_check_pass(arq_redis):
+async def test_health_check_pass(test_redis_settings: RedisSettings, arq_redis: ArqRedis):
  await arq_redis.set(default_queue_name + health_check_key_suffix, b'1')
- assert 0 == await async_check_health(None)
+ assert 0 == await async_check_health(test_redis_settings)
 
 
 async def test_set_health_check_key(arq_redis: ArqRedis, worker):
@@ -479,8 +480,8 @@ async def test_log_health_check(arq_redis: ArqRedis, worker, caplog):
  assert 'recording health' in caplog.text
 
 
-async def test_remain_keys(arq_redis: ArqRedis, worker, create_pool):
- redis2 = await create_pool(RedisSettings())
+async def test_remain_keys(test_redis_settings: RedisSettings, arq_redis: ArqRedis, worker, create_pool):
+ redis2 = await create_pool(test_redis_settings)
  await arq_redis.enqueue_job('foobar', _job_id='testing')
  assert sorted(await redis2.keys('*')) == [b'arq:job:testing', b'arq:queue']
  worker: Worker = worker(functions=[foobar])