Celery logging config

src/palace/manager/celery/app.py

@@ -39,11 +39,13 @@ def import_celery_tasks() -> None:
 
 @setup_logging.connect
 def celery_logger_setup(loglevel: int, logfile: str | None, **kwargs: Any) -> None:
-    container_level = services.logging.config.level().levelno  # type: ignore[attr-defined]
+    services = container_instance()
+    level = services.logging.config.level()  # type: ignore[attr-defined]
+    container_level = level.levelno if level is not None else None
     root_logger = logging.getLogger()
 
     # If celery requested a lower log level, then we update the root logger to use the lower level.
-    if loglevel < container_level:
+    if container_level is None or loglevel < container_level:
         root_logger.setLevel(loglevel)
 
     # If celery requested a log file, then we update the root logger to also log to the file.

tests/manager/celery/tasks/test_search.py

@@ -14,6 +14,7 @@
 )
 from palace.manager.scripts.initialization import InstanceInitializationScript
 from palace.manager.search.external_search import Filter
+from palace.manager.service.logging.configuration import LogLevel
 from tests.fixtures.celery import CeleryFixture
 from tests.fixtures.database import DatabaseTransactionFixture
 from tests.fixtures.search import EndToEndSearchFixture
@@ -216,6 +217,7 @@ def test_index_work_failures(
     caplog: pytest.LogCaptureFixture,
     db: DatabaseTransactionFixture,
 ):
+    caplog.set_level(LogLevel.info)
     # Make sure our backoff function doesn't delay the test.
     mock_backoff.return_value = 0
 

tests/manager/celery/test_monitoring.py

@@ -248,6 +248,7 @@ def test_on_shutter(
         cloudwatch_camera: CloudwatchCameraFixture,
         caplog: pytest.LogCaptureFixture,
     ):
+        caplog.set_level(LogLevel.warning)
         cloudwatch = cloudwatch_camera.create_cloudwatch()
         mock_publish = create_autospec(cloudwatch.publish)
         cloudwatch.publish = mock_publish