From 0e294b223da1c18c16a2dfc0079b57de4c42218a Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Sat, 12 Jul 2025 22:59:21 +0100 Subject: [PATCH] Fix the global unique ID generator recursive constructor call --- amqp/adapter/consul_global_id_generator.py | 74 +++--- amqp/config/amq_configuration.py | 88 ++++--- amqp/config/application.properties | 10 +- amqp/service/amq_service.py | 19 +- amqp/service/consul_global_id_generator.py | 233 ------------------ .../perf_test_consul_global_id_generator.py | 5 +- tests/adapter/test_backpressure_handler.py | 18 +- tests/service/test_amq_service.py | 1 - 8 files changed, 122 insertions(+), 326 deletions(-) delete mode 100644 amqp/service/consul_global_id_generator.py diff --git a/amqp/adapter/consul_global_id_generator.py b/amqp/adapter/consul_global_id_generator.py index 6f2dbc4..2a19ea8 100644 --- a/amqp/adapter/consul_global_id_generator.py +++ b/amqp/adapter/consul_global_id_generator.py @@ -1,51 +1,46 @@ -import os import logging +import os import random import socket import time -from typing import Optional -import consul -from amqp.config.amq_configuration import AMQConfiguration +import consul_kv + +from amqp.config.amq_configuration import AMQAdapter logger = logging.getLogger(__name__) -# Default configuration values for Adapter -DEFAULT_CONSUL_HOST = 'localhost' -DEFAULT_CONSUL_PORT = 8500 -DEFAULT_CONSUL_COUNTER_KEY = 'adapter/ids/counter' -DEFAULT_MAX_RETRIES = 5 -DEFAULT_RETRY_DELAY_SECONDS = 0.1 -def get_config_values(): +def get_config_values(config: AMQAdapter) -> tuple: """ Get configuration values from AMQConfiguration or environment variables. - + Returns: tuple: (consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds) """ try: - config = AMQConfiguration("application.properties") - consul_host = os.environ.get('CONSUL_HOST', DEFAULT_CONSUL_HOST) - consul_port = int(os.environ.get('CONSUL_PORT', DEFAULT_CONSUL_PORT)) - consul_counter_key = os.environ.get('CONSUL_COUNTER_KEY', DEFAULT_CONSUL_COUNTER_KEY) - max_retries = int(os.environ.get('CONSUL_MAX_RETRIES', DEFAULT_MAX_RETRIES)) - retry_delay_seconds = float(os.environ.get('CONSUL_RETRY_DELAY_SECONDS', DEFAULT_RETRY_DELAY_SECONDS)) + consul_host = config.consul_host + consul_port = config.consul_port + consul_counter_key = config.consul_counter_key + max_retries = config.consul_max_retries + retry_delay_seconds = config.consul_initial_retry_delay return consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds except Exception as e: logger.warning(f"Failed to load configuration: {str(e)}. Using default values.") return ( - os.environ.get('CONSUL_HOST', DEFAULT_CONSUL_HOST), - int(os.environ.get('CONSUL_PORT', DEFAULT_CONSUL_PORT)), - os.environ.get('CONSUL_COUNTER_KEY', DEFAULT_CONSUL_COUNTER_KEY), - int(os.environ.get('CONSUL_MAX_RETRIES', DEFAULT_MAX_RETRIES)), - float(os.environ.get('CONSUL_RETRY_DELAY_SECONDS', DEFAULT_RETRY_DELAY_SECONDS)) + os.environ.get("CONSUL_HOST", consul_kv.settings.DEFAULT_ENDPOINT.split(":")[0]), + int(os.environ.get("CONSUL_PORT", consul_kv.settings.DEFAULT_ENDPOINT.split(":")[1])), + os.environ.get("CONSUL_COUNTER_KEY", "service/ids/counter"), + int(os.environ.get("CONSUL_MAX_RETRIES", 5)), + float(os.environ.get("CONSUL_RETRY_DELAY_SECONDS", 0.2)), ) + + def generate_fallback_id() -> int: """ Generate a fallback ID when Consul is not available. Uses a combination of timestamp, hostname hash, and random number. - + Returns: int: A reasonably unique integer ID """ @@ -57,44 +52,43 @@ def generate_fallback_id() -> int: logger.warning(f"Using fallback ID generation method: {unique_id}") return unique_id -def get_unique_instance_id() -> int: + +def get_unique_instance_id(config: AMQAdapter) -> int: """ Get a globally unique ID from Consul. Uses Consul's atomic Compare-And-Set operations to safely increment a counter. Falls back to a local generation method if Consul is unavailable. - + Returns: int: A globally unique integer ID """ - consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds = get_config_values() + consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds = ( + get_config_values(config) + ) try: - c = consul.Consul(host=consul_host, port=consul_port) - index, data = c.kv.get(consul_counter_key) + c = consul_kv.Connection(endpoint=f"{consul_host}:{consul_port}", timeout=5) + index, data = c.get(consul_counter_key) if data is None: logger.info(f"Initializing Consul counter at {consul_counter_key}") - if c.kv.put(consul_counter_key, "1"): + if c.put(consul_counter_key, "1", cas=1): return 1 else: logger.error("Failed to initialize Consul counter") return generate_fallback_id() - current_value = int(data['Value'].decode('utf-8')) + current_value = int(data["Value"].decode("utf-8")) new_value = current_value + 1 for attempt in range(max_retries): - success = c.kv.put( - consul_counter_key, - str(new_value), - cas=data['ModifyIndex'] - ) + success = c.put(consul_counter_key, str(new_value), cas=data["ModifyIndex"]) if success: logger.debug(f"Successfully obtained unique ID: {new_value}") return new_value - logger.debug(f"CAS update failed on attempt {attempt+1}, retrying...") - time.sleep(retry_delay_seconds) - index, data = c.kv.get(consul_counter_key) + logger.debug(f"CAS update failed on attempt {attempt + 1}, retrying...") + time.sleep(retry_delay_seconds * (1 << attempt)) # Exponential backoff + index, data = c.get(consul_counter_key) if data is None: logger.error("Counter disappeared during update") return generate_fallback_id() - current_value = int(data['Value'].decode('utf-8')) + current_value = int(data["Value"].decode("utf-8")) new_value = current_value + 1 logger.error(f"Failed to update counter after {max_retries} attempts") return generate_fallback_id() diff --git a/amqp/config/amq_configuration.py b/amqp/config/amq_configuration.py index 9f33d6d..3f96fb3 100644 --- a/amqp/config/amq_configuration.py +++ b/amqp/config/amq_configuration.py @@ -24,9 +24,7 @@ class AMQAdapter: :param config: config parser to read configuration values """ - self.generator_id = config.getint( - "CleverMicro-AMQ", self.PREFIX + "generator-id", fallback=164 - ) + self.generator_id = 0 self.service_name = config.get( "CleverMicro-AMQ", self.PREFIX + "service-name", @@ -50,49 +48,48 @@ class AMQAdapter: self.swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter") # # Consul related configuration - self.consul_port = int(os.getenv( - "CONSUL_PORT", - config.get( - "CleverMicro-AMQ", - self.PREFIX + "consul-port", - fallback="8500" + self.consul_port = ( + int( + os.getenv( + "CONSUL_PORT", + config.get("CleverMicro-AMQ", self.PREFIX + "consul-port", fallback="8500"), + ) ) - )) if os.getenv("CONSUL_PORT") is not None else config.getint( - "CleverMicro-AMQ", - self.PREFIX + "consul-port", - fallback=8500 + if os.getenv("CONSUL_PORT") is not None + else config.getint("CleverMicro-AMQ", self.PREFIX + "consul-port", fallback=8500) ) self.consul_host = os.getenv( "CONSUL_HOST", - config.get( - "CleverMicro-AMQ", - self.PREFIX + "consul-host", - fallback="consul" + config.get("CleverMicro-AMQ", self.PREFIX + "consul-host", fallback="consul"), + ) + self.consul_max_retries = ( + int( + os.getenv( + "CONSUL_MAX_RETRIES", + config.get("CleverMicro-AMQ", self.PREFIX + "consul-max-retries", fallback="5"), + ) + ) + if os.getenv("CONSUL_MAX_RETRIES") is not None + else config.getint("CleverMicro-AMQ", self.PREFIX + "consul-max-retries", fallback=5) + ) + self.consul_initial_retry_delay = ( + float( + os.getenv( + "CONSUL_INITIAL_RETRY_DELAY", + config.get( + "CleverMicro-AMQ", + self.PREFIX + "consul-initial-retry-delay", + fallback="0.2", + ), + ) + ) + if os.getenv("CONSUL_INITIAL_RETRY_DELAY") is not None + else config.getfloat( + "CleverMicro-AMQ", self.PREFIX + "consul-initial-retry-delay", fallback=0.2 ) ) - self.consul_max_retries = int(os.getenv( - "CONSUL_MAX_RETRIES", - config.get( - "CleverMicro-AMQ", - self.PREFIX + "consul-max-retries", - fallback="5" - ) - )) if os.getenv("CONSUL_MAX_RETRIES") is not None else config.getint( - "CleverMicro-AMQ", - self.PREFIX + "consul-max-retries", - fallback=5 - ) - self.consul_initial_retry_delay = float(os.getenv( - "CONSUL_INITIAL_RETRY_DELAY", - config.get( - "CleverMicro-AMQ", - self.PREFIX + "consul-initial-retry-delay", - fallback="0.2" - ) - )) if os.getenv("CONSUL_INITIAL_RETRY_DELAY") is not None else config.getfloat( - "CleverMicro-AMQ", - self.PREFIX + "consul-initial-retry-delay", - fallback=0.2 + self.consul_counter_key = config.get( + "CleverMicro-AMQ", self.PREFIX + "consul-counter-key", fallback="service/ids/counter" ) def adapter_prefix(self) -> str: @@ -218,3 +215,14 @@ class AMQConfiguration: self.amq_adapter: AMQAdapter = AMQAdapter(config) self.dispatch: Dispatch = Dispatch(config) self.backpressure: Backpressure = Backpressure(config) + + def get_unique_instance_id(self) -> int: + """ + Get a unique instance ID for this AMQ service. + + Returns: + int: A unique instance ID + """ + if self.instance_id == -1: + logging_warning(f"Generated unique instance ID: {self.instance_id}") + return self.instance_id diff --git a/amqp/config/application.properties b/amqp/config/application.properties index 1a70c64..7a0c3c7 100644 --- a/amqp/config/application.properties +++ b/amqp/config/application.properties @@ -14,8 +14,12 @@ cm.amq-adapter.generator-id=1 cm.amq-adapter.route-mapping= # Indicate if AMQ Adapter should respond with HTTP 401 Unauthorized if the request does not contain valid JWT token cm.amq-adapter.require-authenticated-user=false -# Indicate if AMQ Adapter should use inbuilt backpressure monitor (True) or rely only on the service backpressure updates (False) -cm.amq-adapter.default-backpressure-enabled=false +# Consul configuration for CleverMicro AMQ Adapter +cm.amq-adapter.consul-host=consul +cm.amq-adapter.consul-port=8500 +cm.amq-adapter.consul-max-retries=5 +cm.amq-adapter.consul-initial-retry-delay=0.2 +cm.amq-adapter.consul-id-generator-key=services/ids/counter # CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding cm.dispatch.use-dlq=true @@ -42,3 +46,5 @@ cm.backpressure.time-window=10 cm.backpressure.cpu-overload-duration=30 # Define the time window length, in seconds, during which, if no request is made to the Service, the system will trigger IDLE alert cm.backpressure.cpu-idle-duration=30 +# AMQ Adapter can monitor CPU usage and trigger backpressure alerts based on CPU usage. Enable this feature explicitly +cm.backpressure.cpu-monitoring-enabled=false diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index a6bf2d6..a400057 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -10,6 +10,7 @@ from aio_pika.abc import AbstractRobustExchange from amqp.adapter.amq_route_factory import AMQRouteFactory from amqp.adapter.backpressure_handler import BackpressureHandler +from amqp.adapter.consul_global_id_generator import get_unique_instance_id from amqp.adapter.data_message_factory import DataMessageFactory from amqp.adapter.logging_utils import logging_info, logging_warning from amqp.adapter.service_message_factory import ServiceMessageFactory @@ -37,14 +38,17 @@ class AMQService: def __init__(self, amq_configuration: AMQConfiguration, service_adapter=None): logging.info("***********************************************") - logging.info("********** AMQ Service *******************") + logging.info("************** AMQ Service ***************") logging.info("***********************************************") self.loop = asyncio.new_event_loop() self.amq_configuration = amq_configuration self.service_adapter = service_adapter + self.unique_id = get_unique_instance_id(amq_configuration.amq_adapter) + amq_configuration.amq_adapter.generator_id = self.unique_id + self.data_message_factory = DataMessageFactory.get_instance( - self.amq_configuration.amq_adapter.generator_id + self.unique_id & DataMessageFactory.MACHINE_ID_MASK ) self.router = RouterConsumer(self.amq_configuration) self.rabbit_mq_client = RabbitMQClient(self.amq_configuration, self.router) @@ -63,7 +67,7 @@ class AMQService: # ======================================================================= # ======================================================================= - # ====================== A P I M e t h o d s ========================= + # ============== P u b l i c A P I M e t h o d s =================== # ======================================================================= # ======================================================================= def run(self): @@ -123,6 +127,15 @@ class AMQService: ) return result + def get_unique_instance_id(self) -> int: + """ + Provides cluster-wide unique instance ID for this AMQ service instance. + This ID is used to identify the service instance in the cluster and is unique across all instances. + IDs are managed by shared Consul KV store, ensuring that even if multiple instances of the service are running, + they will each have unique ID. + """ + return self.unique_id + # ======================================================================= # ====================== Internal Methods ============================ # ======================================================================= diff --git a/amqp/service/consul_global_id_generator.py b/amqp/service/consul_global_id_generator.py deleted file mode 100644 index 1c7ba8b..0000000 --- a/amqp/service/consul_global_id_generator.py +++ /dev/null @@ -1,233 +0,0 @@ -import consul_kv -import socket -import os -import time -import logging -import threading - -# --- Configuration --- -CONSUL_HOST = os.environ.get('CONSUL_HOST', 'localhost') -CONSUL_PORT = int(os.environ.get('CONSUL_PORT', 8500)) -CONSUL_COUNTER_KEY = 'service/ids/counter' -MAX_RETRIES = 5 # Max attempts for atomic update -RETRY_DELAY_SECONDS = 0.1 # Delay between retries for atomic update - -# --- Logging Setup --- -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s') -logger = logging.getLogger(__name__) - - -# --- Fallback ID Generation --- -def generate_fallback_id() -> int: - """ - Generates a fallback ID based on hostname and PID if Consul is unavailable. - This provides a unique-ish ID for local testing/development but is not guaranteed - to be globally unique across different machines or after restarts. - """ - hostname = socket.gethostname() - pid = os.getpid() - # A simple way to combine them into a number. - # Hash hostname to an integer and combine with PID. - # Note: This is a simple fallback. For production, consider a more robust - # local ID generation or a different strategy if Consul is critical. - fallback_id = (hash(hostname) % 1000000) * 100000 + pid - logger.warning( - f"Consul is unavailable. Generating fallback ID: {fallback_id} (based on hostname: {hostname}, PID: {pid})") - return fallback_id - - -# --- Unique Instance ID Generation Function --- -def get_unique_instance_id() -> int: - """ - Reads and atomically increments a counter in Consul KV to get a unique instance ID. - If Consul is unavailable or the atomic update fails after retries, - it falls back to generating an ID based on hostname and PID. - """ - try: - # Initialize Consul client - client = consul_kv.Connection(endpoint=f"{CONSUL_HOST}:{CONSUL_PORT}") - logger.info(f"Attempting to connect to Consul at {CONSUL_HOST}:{CONSUL_PORT}") - - for attempt in range(MAX_RETRIES): - try: - # 1. Read the current value and its modifyIndex (for CAS) - # client.get returns (value, modifyIndex) - current_value_str, modify_index = client.get(CONSUL_COUNTER_KEY) - - # If key doesn't exist, initialize it to 0 - if current_value_str is None: - current_value = 0 - modify_index = 0 # For a non-existent key, modify_index is 0 for initial CAS - logger.info(f"Consul key '{CONSUL_COUNTER_KEY}' not found. Initializing to 0.") - else: - current_value = int(current_value_str) - - new_value = current_value + 1 - - # 2. Attempt atomic update using CAS (Check-And-Set) - # client.set returns True on success, False on failure (if modify_index changed) - success = client.put(CONSUL_COUNTER_KEY, str(new_value), cas=modify_index) - - if success: - logger.info(f"Successfully obtained unique ID: {new_value} (Attempt {attempt + 1}/{MAX_RETRIES})") - return new_value - else: - logger.warning( - f"CAS failed for '{CONSUL_COUNTER_KEY}' (modify_index changed). Retrying... (Attempt {attempt + 1}/{MAX_RETRIES})") - time.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) # Exponential backoff for retries - except ValueError: - # Handle case where the stored value is not an integer - logger.error( - f"Value stored at '{CONSUL_COUNTER_KEY}' is not a valid integer. Please check Consul KV. Attempting to overwrite.") - # To recover, you might force a set without CAS, but that risks data loss. - # For this demo, we'll just let it retry or fall back. - time.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) - except Exception as e: - # Catch specific ConsulKV errors (e.g., connection issues, invalid response) - logger.error(f"ConsulKV error during atomic update (Attempt {attempt + 1}/{MAX_RETRIES}): {e}") - time.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) # Exponential backoff - - logger.error(f"Failed to obtain unique ID from Consul after {MAX_RETRIES} attempts.") - return generate_fallback_id() # Fallback if all retries fail - - except Exception as e: - # Catch broader connection errors or unexpected issues with Consul client initialization - logger.error(f"Could not connect to Consul or an unexpected error occurred: {e}") - return generate_fallback_id() # Fallback if Consul is completely unreachable - -import os -import logging -import random -import socket -import time -from typing import Optional - -import consul - -from amqp.config.amq_configuration import AMQConfiguration - -from amqp.config.amq_configuration import AMQConfiguration - -# Configuration from environment variables with defaults -CONSUL_HOST = os.environ.get('CONSUL_HOST', 'localhost') -CONSUL_PORT = int(os.environ.get('CONSUL_PORT', 8500)) -CONSUL_COUNTER_KEY = 'service/ids/counter' -MAX_RETRIES = 5 # Max attempts for atomic update -RETRY_DELAY_SECONDS = 0.1 # Delay between retries for atomic update - -logger = logging.getLogger(__name__) - - -def generate_fallback_id() -> int: - """ - Generate a fallback ID when Consul is not available. - Uses a combination of timestamp, hostname hash, and random number. - - Returns: - int: A reasonably unique integer ID - """ - # Get current timestamp in milliseconds - timestamp = int(time.time() * 1000) - - # Get hostname and hash it to an integer - hostname = socket.gethostname() - hostname_hash = hash(hostname) % 10000 # Limit to 4 digits - - # Generate a random number - random_part = random.randint(0, 9999) # 4 digits - - # Combine all parts into a single integer - # Format: TTTTTTTTTTT-HHHH-RRRR (T=timestamp, H=hostname hash, R=random) - unique_id = (timestamp * 100000000) + (hostname_hash * 10000) + random_part - - logger.warning(f"Using fallback ID generation method: {unique_id}") - return unique_id - - -def get_unique_instance_id() -> int: - """ - Get a globally unique ID from Consul. - Uses Consul's atomic Compare-And-Set operations to safely increment a counter. - Falls back to a local generation method if Consul is unavailable. - - Returns: - int: A globally unique integer ID - """ - consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds = get_config_values() - try: - # Connect to Consul - c = consul.Consul(host=consul_host, port=consul_port) - - # Try to get the current counter value - index, data = c.kv.get(CONSUL_COUNTER_KEY) - - # If the key doesn't exist yet, create it with initial value - if data is None: - logger.info(f"Initializing Consul counter at {CONSUL_COUNTER_KEY}") - if c.kv.put(CONSUL_COUNTER_KEY, "1"): - return 1 - else: - logger.error("Failed to initialize Consul counter") - return generate_fallback_id() - - # Get the current value - current_value = int(data['Value'].decode('utf-8')) - new_value = current_value + 1 - - # Try to atomically update the counter - for attempt in range(MAX_RETRIES): - # Use Compare-And-Set to ensure atomicity - success = c.kv.put( - CONSUL_COUNTER_KEY, - str(new_value), - cas=data['ModifyIndex'] - ) - - if success: - logger.debug(f"Successfully obtained unique ID: {new_value}") - return new_value - - # If CAS failed, someone else updated the value, retry - logger.debug(f"CAS update failed on attempt {attempt+1}, retrying...") - time.sleep(RETRY_DELAY_SECONDS) - - # Get the latest value for the next attempt - index, data = c.kv.get(CONSUL_COUNTER_KEY) - if data is None: - logger.error("Counter disappeared during update") - return generate_fallback_id() - - current_value = int(data['Value'].decode('utf-8')) - new_value = current_value + 1 - - # If we've exhausted all retries, use the fallback - logger.error(f"Failed to update counter after {MAX_RETRIES} attempts") - return generate_fallback_id() - - except Exception as e: - logger.error(f"Error accessing Consul: {str(e)}") - return generate_fallback_id() -def get_config_values(): - """ - Get configuration values from AMQConfiguration or environment variables. - - Returns: - tuple: (consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds) - """ - try: - config = AMQConfiguration("application.properties") - consul_host = os.environ.get('CONSUL_HOST', 'localhost') - consul_port = int(os.environ.get('CONSUL_PORT', 8500)) - consul_counter_key = os.environ.get('CONSUL_COUNTER_KEY', 'service/ids/counter') - max_retries = int(os.environ.get('CONSUL_MAX_RETRIES', 5)) - retry_delay_seconds = float(os.environ.get('CONSUL_RETRY_DELAY_SECONDS', 0.1)) - return consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds - except Exception as e: - logger.warning(f"Failed to load configuration: {str(e)}. Using default values.") - return ( - os.environ.get('CONSUL_HOST', 'localhost'), - int(os.environ.get('CONSUL_PORT', 8500)), - os.environ.get('CONSUL_COUNTER_KEY', 'service/ids/counter'), - int(os.environ.get('CONSUL_MAX_RETRIES', 5)), - float(os.environ.get('CONSUL_RETRY_DELAY_SECONDS', 0.1)) - ) diff --git a/tests/adapter/perf_test_consul_global_id_generator.py b/tests/adapter/perf_test_consul_global_id_generator.py index 9a5b36e..50520b2 100644 --- a/tests/adapter/perf_test_consul_global_id_generator.py +++ b/tests/adapter/perf_test_consul_global_id_generator.py @@ -2,7 +2,8 @@ import logging import os import threading -from amqp.service.consul_global_id_generator import get_unique_instance_id +from amqp.adapter.consul_global_id_generator import get_unique_instance_id +from amqp.config.amq_configuration import AMQConfiguration CONSUL_HOST = os.environ.get("CONSUL_HOST", "localhost") CONSUL_PORT = int(os.environ.get("CONSUL_PORT", 8500)) @@ -20,7 +21,7 @@ logger = logging.getLogger(__name__) # --- Demo: Simulate Concurrent Access --- def worker_thread(results_list): """A function to be run by multiple threads to get unique IDs.""" - instance_id = get_unique_instance_id() + instance_id = get_unique_instance_id(AMQConfiguration("").amq_adapter) results_list.append(instance_id) logger.info(f"Thread {threading.current_thread().name} got ID: {instance_id}") diff --git a/tests/adapter/test_backpressure_handler.py b/tests/adapter/test_backpressure_handler.py index b1987fa..49762b5 100644 --- a/tests/adapter/test_backpressure_handler.py +++ b/tests/adapter/test_backpressure_handler.py @@ -275,7 +275,9 @@ class TestBackpressureHandler: """ Pytest fixture to create a mock AMQConfiguration. """ - return AMQConfiguration("") + amq_cfg = AMQConfiguration("") + amq_cfg.backpressure.cpu_monitoring_enabled = True + return amq_cfg def test_backpressure_handler_initialization(self, mock_channel, mock_loop, mock_config): """ @@ -649,8 +651,12 @@ class TestBackpressureHandler: mock_time.return_value = current_time # Set last data message time to be older than idle_duration - handler.last_data_message_time = current_time - mock_config.backpressure.idle_duration - 10 - handler.last_backpressure_event_time = current_time - mock_config.backpressure.time_window - 10 + handler.last_data_message_time = ( + current_time - mock_config.backpressure.idle_duration - 10 + ) + handler.last_backpressure_event_time = ( + current_time - mock_config.backpressure.time_window - 10 + ) # Call the method await handler.check_overload_condition() @@ -679,7 +685,9 @@ class TestBackpressureHandler: mock_time.return_value = current_time # Set last event time to be older than time_window - handler.last_backpressure_event_time = current_time - mock_config.backpressure.time_window - 10 + handler.last_backpressure_event_time = ( + current_time - mock_config.backpressure.time_window - 10 + ) # Call the method await handler.check_overload_condition() @@ -701,7 +709,7 @@ class TestBackpressureHandler: # Verify no thread was started assert thread is None - assert not hasattr(handler, 'thread') or handler.thread is None + assert not hasattr(handler, "thread") or handler.thread is None def test_start_backpressure_monitor_enabled(self, mock_channel, mock_loop, mock_config): """ diff --git a/tests/service/test_amq_service.py b/tests/service/test_amq_service.py index 0c57e9c..7dc4ad6 100644 --- a/tests/service/test_amq_service.py +++ b/tests/service/test_amq_service.py @@ -10,7 +10,6 @@ class TestAMQService(unittest.IsolatedAsyncioTestCase): def setUp(self): # Create a mock configuration self.mock_adapter = Mock(spec=AMQAdapter) - self.mock_adapter.generator_id = 12345 self.mock_adapter.service_name = "test-service" self.mock_adapter.route_mapping = {} self.mock_adapter.swarm_service_id = "test-swarm-id"