fix bugs: address rabbitmq initialization failed when doing pytest

Author: tangg555

src/memos/mem_scheduler/task_schedule_modules/redis_queue.py

@@ -249,8 +249,25 @@ def ack_message(
         stream_key = self.get_stream_key(
             user_id=user_id, mem_cube_id=mem_cube_id, task_label=task_label
         )
+        # No-op if not connected or message doesn't come from Redis
+        if not self._redis_conn:
+            logger.debug(
+                f"Skip ack: Redis not connected for stream '{stream_key}', msg_id='{redis_message_id}'"
+            )
+            return
+        if not redis_message_id:
+            logger.debug(
+                f"Skip ack: Empty redis_message_id for stream '{stream_key}', user_id='{user_id}', label='{task_label}'"
+            )
+            return
 
-        self.redis.xack(stream_key, self.consumer_group, redis_message_id)
+        try:
+            self._redis_conn.xack(stream_key, self.consumer_group, redis_message_id)
+        except Exception as e:
+            logger.warning(
+                f"xack failed for stream '{stream_key}', msg_id='{redis_message_id}': {e}"
+            )
+            return
 
         # Optionally delete the message from the stream to keep it clean
         if self.auto_delete_acked:

src/memos/mem_scheduler/webservice_modules/rabbitmq_service.py

@@ -70,6 +70,16 @@ def initialize_rabbitmq(
         Establish connection to RabbitMQ using pika.
         """
         try:
+            # Skip remote initialization in CI/pytest unless explicitly enabled
+            enable_env = os.getenv("MEMOS_ENABLE_RABBITMQ", "").lower() == "true"
+            in_ci = os.getenv("CI", "").lower() == "true"
+            in_pytest = os.getenv("PYTEST_CURRENT_TEST") is not None
+            if (in_ci or in_pytest) and not enable_env:
+                logger.info(
+                    "Skipping RabbitMQ initialization in CI/test environment. Set MEMOS_ENABLE_RABBITMQ=true to enable."
+                )
+                return
+
             from pika.adapters.select_connection import SelectConnection
 
             if config is None:

src/memos/mem_scheduler/webservice_modules/redis_service.py

@@ -111,6 +111,16 @@ def auto_initialize_redis(self) -> bool:
         Returns:
             bool: True if Redis connection is successfully established, False otherwise
         """
+        # Skip remote initialization in CI/pytest unless explicitly enabled
+        enable_env = os.getenv("MEMOS_ENABLE_REDIS", "").lower() == "true"
+        in_ci = os.getenv("CI", "").lower() == "true"
+        in_pytest = os.getenv("PYTEST_CURRENT_TEST") is not None
+        if (in_ci or in_pytest) and not enable_env:
+            logger.info(
+                "Skipping Redis auto-initialization in CI/test environment. Set MEMOS_ENABLE_REDIS=true to enable."
+            )
+            return False
+
         import redis
 
         # Strategy 1: Try to initialize from config