Using mocks instead of real docker containers in unit tests

src/pytest_celery/api/worker.py

@@ -20,10 +20,7 @@ def __init__(self, container: CeleryTestContainer, app: Celery):
 
     @property
     def version(self) -> str:
-        if hasattr(self.container, "version"):
-            return self.container.version()
-        else:
-            return "unknown"
+        return self.container.version()
 
     @property
     def log_level(self) -> str:

tests/integration/vendors/test_memcached.py

@@ -17,6 +17,10 @@ def test_client(self, container: MemcachedContainer):
         assert container.client.get("ready") == "1"
         assert container.client.delete("ready")
 
+    def test_celeryconfig(self, container: MemcachedContainer):
+        expected_keys = {"url", "local_url", "hostname", "port"}
+        assert set(container.celeryconfig.keys()) == expected_keys
+
 
 # @pytest.mark.parametrize("node", [lazy_fixture(CELERY_MEMCACHED_BACKEND)])
 # class test_memcached_test_backend:

tests/integration/vendors/test_rabbitmq.py

@@ -19,6 +19,10 @@ def test_client(self, container: RabbitMQContainer):
         finally:
             c.release()
 
+    def test_celeryconfig(self, container: RabbitMQContainer):
+        expected_keys = {"url", "local_url", "hostname", "port", "vhost"}
+        assert set(container.celeryconfig.keys()) == expected_keys
+
 
 # @pytest.mark.parametrize("node", [lazy_fixture(CELERY_RABBITMQ_BROKER)])
 # class test_rabbitmq_test_broker:

tests/integration/vendors/test_redis.py

@@ -19,6 +19,10 @@ def test_client(self, container: RedisContainer):
         assert container.client.get("ready") == "1"
         assert container.client.delete("ready")
 
+    def test_celeryconfig(self, container: RedisContainer):
+        expected_keys = {"url", "local_url", "hostname", "port", "vhost"}
+        assert set(container.celeryconfig.keys()) == expected_keys
+
 
 # @pytest.mark.parametrize("node", [lazy_fixture(CELERY_REDIS_BACKEND)])
 # class test_redis_test_backend:

tests/integration/vendors/test_worker.py

@@ -13,6 +13,10 @@ def test_client(self, container: CeleryWorkerContainer):
         assert container.client
         assert container.client == container
 
+    def test_celeryconfig(self, container: CeleryWorkerContainer):
+        with pytest.raises(NotImplementedError):
+            container.celeryconfig
+
 
 # @pytest.mark.parametrize("node", [lazy_fixture(CELERY_SETUP_WORKER)])
 # class test_base_test_worker:

tests/unit/api/test_backend.py

@@ -1,25 +1,14 @@
 from pytest_celery import CeleryBackendCluster
 from pytest_celery import CeleryTestBackend
-from pytest_celery import CeleryTestContainer
 
 
 class test_celey_test_backend:
-    def test_default_config_format(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node = CeleryTestBackend(unit_tests_container)
+    def test_default_config_format(self, celery_backend: CeleryTestBackend):
         expected_format = {"url", "local_url"}
-        assert set(node.default_config().keys()) == expected_format
+        assert set(celery_backend.default_config().keys()) == expected_format
 
 
 class test_celery_backend_cluster:
-    def test_default_config_format(
-        self,
-        unit_tests_container: CeleryTestContainer,
-        local_test_container: CeleryTestContainer,
-    ):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestBackend(unit_tests_container)
-        node2 = CeleryTestBackend(local_test_container)
-        cluster = CeleryBackendCluster(node1, node2)
+    def test_default_config_format(self, celery_backend_cluster: CeleryBackendCluster):
         expected_format = {"urls", "local_urls"}
-        assert set(cluster.default_config().keys()) == expected_format
+        assert set(celery_backend_cluster.default_config().keys()) == expected_format

tests/unit/api/test_base.py

@@ -1,90 +1,87 @@
+from unittest.mock import Mock
+from unittest.mock import patch
+
+import pytest
 from celery import Celery
 
 from pytest_celery import CeleryTestCluster
 from pytest_celery import CeleryTestContainer
 from pytest_celery import CeleryTestNode
 
 
+@pytest.fixture
+def mocked_test_container() -> CeleryTestContainer:
+    return Mock(spec=CeleryTestContainer)
+
+
+@pytest.fixture(autouse=True)
+def mock_wait_for_callable():
+    with patch("pytest_celery.api.base.wait_for_callable", new=Mock()):
+        yield
+
+
 class test_celery_test_node:
-    def test_ready(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node = CeleryTestNode(unit_tests_container)
-        assert node.ready()
+    @pytest.fixture
+    def node(self, mocked_test_container: CeleryTestContainer):
+        return CeleryTestNode(mocked_test_container)
 
-    def test_app(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
+    def test_app(self, mocked_test_container: CeleryTestContainer):
         expected_app = Celery()
-        node = CeleryTestNode(unit_tests_container, expected_app)
+        node = CeleryTestNode(mocked_test_container, expected_app)
         assert node.app is expected_app
 
-    def test_default_config_format(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node = CeleryTestNode(unit_tests_container)
+    def test_default_config_format(self, mocked_test_container: CeleryTestContainer):
+        node = CeleryTestNode(mocked_test_container)
         assert node.default_config() == dict()
 
-    def test_eq_opertor(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestNode(unit_tests_container)
-        node2 = CeleryTestNode(unit_tests_container)
+    def test_eq_opertor(self, mocked_test_container: CeleryTestContainer):
+        node1 = CeleryTestNode(mocked_test_container)
+        node2 = CeleryTestNode(mocked_test_container)
         assert node1 == node2
         assert node1 is not node2
 
-    def test_wait_for_log(self, unit_tests_container: CeleryTestContainer, celery_setup_app: Celery):
-        # TODO: Use mock instead of real container
-        node = CeleryTestNode(unit_tests_container, celery_setup_app)
+    def test_ready(self, node: CeleryTestNode):
+        assert node.ready()
+
+    def test_wait_for_log(self, node: CeleryTestNode):
         node.wait_for_log("", "test_celey_test_worker.test_wait_for_log")
 
-    def test_assert_log_exists(self, unit_tests_container: CeleryTestContainer, celery_setup_app: Celery):
-        # TODO: Use mock instead of real container
-        node = CeleryTestNode(unit_tests_container, celery_setup_app)
+    def test_assert_log_exists(self, node: CeleryTestNode):
         node.assert_log_exists("", "test_celey_test_worker.test_assert_log_exists")
 
 
 class test_celery_test_cluster:
-    def test_ready(
-        self,
-        unit_tests_container: CeleryTestContainer,
-        local_test_container: CeleryTestContainer,
-    ):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestNode(unit_tests_container)
+    @pytest.fixture
+    def cluster(self, mocked_test_container: CeleryTestContainer):
+        local_test_container = Mock(spec=CeleryTestContainer)
+        node1 = CeleryTestNode(mocked_test_container)
         node2 = CeleryTestNode(local_test_container)
-        cluster = CeleryTestCluster(node1, node2)
-        assert cluster.ready()
-
-    def test_default_config_format(
-        self,
-        unit_tests_container: CeleryTestContainer,
-        local_test_container: CeleryTestContainer,
-    ):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestNode(unit_tests_container)
-        node2 = CeleryTestNode(local_test_container)
-        cluster = CeleryTestCluster(node1, node2)
-        assert cluster.default_config() == dict()
+        return CeleryTestCluster(node1, node2)
 
-    def test_set_nodes(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestNode(unit_tests_container)
-        node2 = CeleryTestNode(unit_tests_container)
+    def test_set_nodes(self, mocked_test_container: CeleryTestContainer):
+        node1 = CeleryTestNode(mocked_test_container)
+        node2 = CeleryTestNode(mocked_test_container)
         cluster = CeleryTestCluster(node1)
         cluster._set_nodes(node2)
         assert cluster[0] == node2
 
-    def test_iter(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestNode(unit_tests_container)
+    def test_iter(self, mocked_test_container: CeleryTestContainer):
+        node1 = CeleryTestNode(mocked_test_container)
         cluster = CeleryTestCluster(node1)
         assert list(cluster) == [node1]
 
-    def test_len(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestNode(unit_tests_container)
+    def test_len(self, mocked_test_container: CeleryTestContainer):
+        node1 = CeleryTestNode(mocked_test_container)
         cluster = CeleryTestCluster(node1)
         assert len(cluster) == 1
 
-    def test_getitem(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestNode(unit_tests_container)
+    def test_getitem(self, mocked_test_container: CeleryTestContainer):
+        node1 = CeleryTestNode(mocked_test_container)
         cluster = CeleryTestCluster(node1)
         assert cluster[0] == node1
+
+    def test_ready(self, cluster: CeleryTestCluster):
+        assert cluster.ready()
+
+    def test_default_config_format(self, cluster: CeleryTestCluster):
+        assert cluster.default_config() == dict()

tests/unit/api/test_broker.py

@@ -1,25 +1,14 @@
 from pytest_celery import CeleryBrokerCluster
 from pytest_celery import CeleryTestBroker
-from pytest_celery import CeleryTestContainer
 
 
-class test_celery_test_broker:
-    def test_default_config_format(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        node = CeleryTestBroker(unit_tests_container)
+class test_celey_test_broker:
+    def test_default_config_format(self, celery_broker: CeleryTestBroker):
         expected_format = {"url", "local_url"}
-        assert set(node.default_config().keys()) == expected_format
+        assert set(celery_broker.default_config().keys()) == expected_format
 
 
 class test_celery_broker_cluster:
-    def test_default_config_format(
-        self,
-        unit_tests_container: CeleryTestContainer,
-        local_test_container: CeleryTestContainer,
-    ):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestBroker(unit_tests_container)
-        node2 = CeleryTestBroker(local_test_container)
-        cluster = CeleryBrokerCluster(node1, node2)
+    def test_default_config_format(self, celery_broker_cluster: CeleryBrokerCluster):
         expected_format = {"urls", "local_urls"}
-        assert set(cluster.default_config().keys()) == expected_format
+        assert set(celery_broker_cluster.default_config().keys()) == expected_format

tests/unit/api/test_container.py

@@ -1,8 +1,17 @@
+from unittest.mock import Mock
+
+import pytest
+
 from pytest_celery import CeleryTestContainer
 
 
+@pytest.fixture
+def mocked_container() -> CeleryTestContainer:
+    docker_container_mock = Mock()
+    return CeleryTestContainer(container=docker_container_mock)
+
+
 class test_celery_test_container:
-    def test_client(self, unit_tests_container: CeleryTestContainer):
-        # TODO: Use mock instead of real container
-        # TODO: Check raises NotImplementedError and remove the client overload from the unit tests container
-        assert unit_tests_container.client == unit_tests_container
+    def test_client(self, mocked_container: CeleryTestContainer):
+        with pytest.raises(NotImplementedError):
+            mocked_container.client

tests/unit/api/test_worker.py

@@ -1,45 +1,21 @@
 from celery import Celery
 
-from pytest_celery import CeleryTestContainer
+from pytest_celery import DEFAULT_WORKER_VERSION
 from pytest_celery import CeleryTestWorker
 from pytest_celery import CeleryWorkerCluster
 
 
 class test_celey_test_worker:
-    def test_app(self, unit_tests_container: CeleryTestContainer, celery_setup_app: Celery):
-        # TODO: Use mock instead of real container
-        node = CeleryTestWorker(unit_tests_container, celery_setup_app)
-        assert node.app is celery_setup_app
-
-    def test_version(self, unit_tests_container: CeleryTestContainer, celery_setup_app: Celery):
-        # TODO: Use mock instead of real container
-        node = CeleryTestWorker(unit_tests_container, celery_setup_app)
-        assert node.version == "unknown"
+    def test_ready(self, celery_worker: CeleryTestWorker):
+        assert celery_worker.ready()
+        celery_worker.container.ready.assert_called_once()
 
 
 class test_celery_worker_cluster:
-    def test_app(
-        self,
-        unit_tests_container: CeleryTestContainer,
-        local_test_container: CeleryTestContainer,
-        celery_setup_app: Celery,
-    ):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestWorker(unit_tests_container, celery_setup_app)
-        node2 = CeleryTestWorker(local_test_container, celery_setup_app)
-        cluster = CeleryWorkerCluster(node1, node2)
+    def test_app(self, celery_worker_cluster: CeleryWorkerCluster, celery_setup_app: Celery):
         node: CeleryTestWorker
-        for node in cluster:
+        for node in celery_worker_cluster:
             assert node.app is celery_setup_app
 
-    def test_versions(
-        self,
-        unit_tests_container: CeleryTestContainer,
-        local_test_container: CeleryTestContainer,
-        celery_setup_app: Celery,
-    ):
-        # TODO: Use mock instead of real container
-        node1 = CeleryTestWorker(unit_tests_container, celery_setup_app)
-        node2 = CeleryTestWorker(local_test_container, celery_setup_app)
-        cluster = CeleryWorkerCluster(node1, node2)
-        assert cluster.versions == {"unknown"}
+    def test_versions(self, celery_worker_cluster: CeleryWorkerCluster):
+        assert celery_worker_cluster.versions == {DEFAULT_WORKER_VERSION}