优化全能创作

core/backend/apps/common/test_celery_health.py

@@ -1,3 +1,4 @@
+import os
 from unittest.mock import Mock, patch
 
 from django.test import SimpleTestCase, override_settings
@@ -7,6 +8,24 @@ from apps.common import celery_health
 from apps.common.celery_health import require_worker_task, worker_consumes_queue
 
 
+class RedisLocalUrlTests(SimpleTestCase):
+    def test_macos_maps_docker_only_host_to_local_redis_for_celery_and_django(self):
+        from airshelf.settings import base
+
+        with patch.dict(os.environ, {"CELERY_BROKER_URL": "redis://:example@yingqing-redis:6379/1"}):
+            with patch.object(base.sys, "platform", "darwin"):
+                self.assertEqual(base.redis_url("CELERY_BROKER_URL", 1), "redis://127.0.0.1:6379/1")
+                self.assertEqual(os.environ["CELERY_BROKER_URL"], "redis://127.0.0.1:6379/1")
+
+    def test_container_keeps_configured_redis_host(self):
+        from airshelf.settings import base
+
+        configured = "redis://:example@yingqing-redis:6379/1"
+        with patch.dict(os.environ, {"CELERY_BROKER_URL": configured}):
+            with patch.object(base.sys, "platform", "linux"):
+                self.assertEqual(base.redis_url("CELERY_BROKER_URL", 1), configured)
+
+
 class RequireWorkerTaskTests(SimpleTestCase):
     def setUp(self):
         celery_health._registered_task_cache.clear()