less code

Author: khvn26

tests/unit/task_processor/conftest.py

@@ -23,32 +23,6 @@ def task_processor_mode_marked(request: pytest.FixtureRequest) -> None:
             request.getfixturevalue("task_processor_mode")
 
 
-class GetTaskProcessorCaplog(typing.Protocol):
-    def __call__(
-        self, log_level: str | int = logging.INFO
-    ) -> pytest.LogCaptureFixture: ...
-
-
-@pytest.fixture
-def get_task_processor_caplog(
-    caplog: pytest.LogCaptureFixture,
-) -> GetTaskProcessorCaplog:
-    # caplog doesn't allow you to capture logging outputs from loggers that don't
-    # propagate to root. Quick hack here to get the task_processor logger to
-    # propagate.
-    # TODO: look into using loguru.
-
-    def _inner(log_level: str | int = logging.INFO) -> pytest.LogCaptureFixture:
-        task_processor_logger = logging.getLogger("task_processor")
-        task_processor_logger.propagate = True
-        # Assume required level for the logger.
-        task_processor_logger.setLevel(log_level)
-        caplog.set_level(log_level)
-        return caplog
-
-    return _inner
-
-
 @pytest.fixture(autouse=True)
 def task_registry() -> typing.Generator[dict[str, RegisteredTask], None, None]:
     from task_processor.task_registry import registered_tasks

tests/unit/task_processor/test_unit_task_processor_decorators.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import typing
 from datetime import timedelta
 from unittest.mock import MagicMock
@@ -18,10 +19,6 @@
 from task_processor.task_registry import get_task, initialise
 from task_processor.task_run_method import TaskRunMethod
 
-if typing.TYPE_CHECKING:
-    # This import breaks private-package-test workflow in core
-    from tests.unit.task_processor.conftest import GetTaskProcessorCaplog
-
 
 @pytest.fixture
 def mock_thread_class(
@@ -36,12 +33,12 @@ def mock_thread_class(
 
 @pytest.mark.django_db
 def test_register_task_handler_run_in_thread__transaction_commit__true__default(
-    get_task_processor_caplog: "GetTaskProcessorCaplog",
+    caplog: pytest.LogCaptureFixture,
     mock_thread_class: MagicMock,
     django_capture_on_commit_callbacks: DjangoCaptureOnCommitCallbacks,
 ) -> None:
     # Given
-    caplog = get_task_processor_caplog()
+    caplog.set_level(logging.DEBUG)
 
     @register_task_handler()
     def my_function(*args: str, **kwargs: str) -> None:
@@ -69,11 +66,11 @@ def my_function(*args: str, **kwargs: str) -> None:
 
 
 def test_register_task_handler_run_in_thread__transaction_commit__false(
-    get_task_processor_caplog: "GetTaskProcessorCaplog",
+    caplog: pytest.LogCaptureFixture,
     mock_thread_class: MagicMock,
 ) -> None:
     # Given
-    caplog = get_task_processor_caplog()
+    caplog.set_level(logging.DEBUG)
 
     @register_task_handler(transaction_on_commit=False)
     def my_function(*args: typing.Any, **kwargs: typing.Any) -> None:

tests/unit/task_processor/test_unit_task_processor_processor.py

@@ -32,11 +32,6 @@
 )
 from task_processor.task_registry import initialise, registered_tasks
 
-if typing.TYPE_CHECKING:
-    # This import breaks private-package-test workflow in core
-    from tests.unit.task_processor.conftest import GetTaskProcessorCaplog
-
-
 DEFAULT_CACHE_KEY = "foo"
 DEFAULT_CACHE_VALUE = "bar"
 
@@ -109,10 +104,9 @@ def test_run_task_runs_task_and_creates_task_run_object_when_success(
 @pytest.mark.task_processor_mode
 def test_run_task_kills_task_after_timeout(
     sleep_task: TaskHandler[[int]],
-    get_task_processor_caplog: "GetTaskProcessorCaplog",
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     # Given
-    caplog = get_task_processor_caplog(logging.ERROR)
     task = Task.create(
         sleep_task.task_identifier,
         scheduled_for=timezone.now(),
@@ -148,11 +142,9 @@ def test_run_task_kills_task_after_timeout(
 @pytest.mark.django_db
 @pytest.mark.task_processor_mode
 def test_run_recurring_task_kills_task_after_timeout(
-    get_task_processor_caplog: "GetTaskProcessorCaplog",
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     # Given
-    caplog = get_task_processor_caplog(logging.ERROR)
-
     @register_recurring_task(
         run_every=timedelta(seconds=1), timeout=timedelta(microseconds=1)
     )
@@ -384,11 +376,10 @@ def _a_task() -> None:
 @pytest.mark.task_processor_mode
 def test_run_task_runs_task_and_creates_task_run_object_when_failure(
     raise_exception_task: TaskHandler[[str]],
-    get_task_processor_caplog: "GetTaskProcessorCaplog",
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     # Given
-    caplog = get_task_processor_caplog(logging.DEBUG)
-
+    caplog.set_level(logging.DEBUG)
     msg = "Error!"
     task = Task.create(
         raise_exception_task.task_identifier, args=(msg,), scheduled_for=timezone.now()

tests/unit/task_processor/test_unit_task_processor_threads.py

@@ -1,5 +1,4 @@
 import logging
-import typing
 from typing import Type
 
 import pytest
@@ -8,10 +7,6 @@
 
 from task_processor.threads import TaskRunner
 
-if typing.TYPE_CHECKING:
-    # This import breaks private-package-test workflow in core
-    from tests.unit.task_processor.conftest import GetTaskProcessorCaplog
-
 
 @pytest.mark.parametrize(
     "exception_class, exception_message",
@@ -20,13 +15,11 @@
 @pytest.mark.django_db
 def test_task_runner_is_resilient_to_errors(
     mocker: MockerFixture,
-    get_task_processor_caplog: "GetTaskProcessorCaplog",
+    caplog: pytest.LogCaptureFixture,
     exception_class: Type[Exception],
     exception_message: str,
 ) -> None:
     # Given
-    caplog = get_task_processor_caplog(logging.DEBUG)
-
     task_runner = TaskRunner()
     mocker.patch(
         "task_processor.threads.run_tasks",