diff --git a/README.md b/README.md index 3af09d800..1783359db 100644 --- a/README.md +++ b/README.md @@ -268,6 +268,21 @@ The custom configuration options for AMQP 1.0 messaging are listed below: * `IIB_MESSAGING_URLS` - a list of AMQP(S) URLs to use when connecting to the AMQP 1.0 broker. This must be set if messaging is enabled. +The custom configuration options for Kafka messaging are listed below. Kafka messaging is optional. +When configured, IIB publishes the state-change events to configured Kafka topics : + +* `IIB_KAFKA_BROKERS` - a list of Kafka broker addresses (e.g. `['broker1:9096', 'broker2:9096']`). + If this is not set, Kafka messaging is disabled entirely. +* `IIB_KAFKA_USERNAME` - the SASL username for Kafka authentication. Both `IIB_KAFKA_USERNAME` and + `IIB_KAFKA_PASSWORD` must be set together. +* `IIB_KAFKA_PASSWORD` - the SASL password for Kafka authentication. +* `IIB_KAFKA_SSL_CAFILE` - the path to a CA certificate file used to verify the Kafka broker's TLS + certificate. This defaults to `/etc/pki/tls/certs/ca-bundle.crt`. +* `IIB_KAFKA_BUILD_STATE_TOPIC` - the Kafka topic to publish build request state-change messages to. + If this is not set, build state-change messages will not be published to Kafka. +* `IIB_KAFKA_BATCH_STATE_TOPIC` - the Kafka topic to publish batch state-change messages to. + If this is not set, batch state-change messages will not be published to Kafka. + If you wish to configure AWS S3 bucket for storing artifact files, the following **environment variables** must be set along with `IIB_AWS_S3_BUCKET_NAME` config variable: @@ -508,10 +523,10 @@ modifications, such as registry replacement, will still be applied. ## Messaging -IIB has support to send messages to an AMQP 1.0 broker. If configured to do so, IIB will send -messages when a build request state changes and when a batch state changes. Please note that if a -message can't be sent due to an infrastructure issue, the build request will continue as it is not -considered a fatal error. +IIB has support to send messages to an AMQP 1.0 broker (will be depricated in future) and to Apache Kafka topics. +If configured to do so, IIB will send messages when a build request state changes and when a batch +state changes. Please notethat if a message can't be sent due to an infrastructure issue, the build request will +continue as it is not considered a fatal error. The build request state change message body is the JSON representation of the build request in the non-verbose format like in the `/builds` API endpoint. The message has the following keys set in diff --git a/docs/module_documentation/iib.web.rst b/docs/module_documentation/iib.web.rst index 634c50278..c7807f369 100644 --- a/docs/module_documentation/iib.web.rst +++ b/docs/module_documentation/iib.web.rst @@ -66,6 +66,14 @@ iib.web.iib\_static\_types module :undoc-members: :show-inheritance: +iib.web.kafka\_producer module +------------------------------ + +.. automodule:: iib.web.kafka_producer + :members: + :undoc-members: + :show-inheritance: + iib.web.manage module --------------------- diff --git a/docs/requirements.txt b/docs/requirements.txt index 6e179f1a4..2d43f8f95 100644 --- a/docs/requirements.txt +++ b/docs/requirements.txt @@ -5,6 +5,7 @@ flask flask-login flask-migrate flask-sqlalchemy +kafka-python opentelemetry-api opentelemetry-exporter-otlp opentelemetry-instrumentation diff --git a/iib/web/config.py b/iib/web/config.py index dfbadd5db..5c1545ae3 100644 --- a/iib/web/config.py +++ b/iib/web/config.py @@ -38,6 +38,13 @@ class Config(object): IIB_REQUEST_RELATED_BUNDLES_DIR: Optional[str] = None IIB_REQUEST_RECURSIVE_RELATED_BUNDLES_DIR: Optional[str] = None IIB_USER_TO_QUEUE: Union[Dict[str, str], Dict[str, Dict[str, str]]] = _get_empty_dict_str_str() + # Kafka messaging configuration + IIB_KAFKA_BATCH_STATE_TOPIC: Optional[str] = None + IIB_KAFKA_BROKERS: Optional[List[str]] = None + IIB_KAFKA_BUILD_STATE_TOPIC: Optional[str] = None + IIB_KAFKA_SSL_CAFILE: str = '/etc/pki/tls/certs/ca-bundle.crt' + IIB_KAFKA_USERNAME: Optional[str] = None + IIB_KAFKA_PASSWORD: Optional[str] = None IIB_WORKER_USERNAMES: List[str] = [] SQLALCHEMY_TRACK_MODIFICATIONS: bool = False diff --git a/iib/web/kafka_producer.py b/iib/web/kafka_producer.py new file mode 100644 index 000000000..ca572a46d --- /dev/null +++ b/iib/web/kafka_producer.py @@ -0,0 +1,95 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +import json +import threading +from typing import Any, Dict, Optional, Union + +from flask import current_app +from kafka import KafkaProducer + +from iib.web.iib_static_types import ( + BaseClassRequestResponse, + BatchRequestResponseList, +) + +_kafka_producer: Optional[KafkaProducer] = None +_producer_lock = threading.Lock() + + +def get_kafka_producer() -> Optional[KafkaProducer]: + """Return a cached, thread-safe KafkaProducer.""" + global _kafka_producer + + # First check without lock for performance + if _kafka_producer is not None: + return _kafka_producer + + with _producer_lock: + # Second check inside lock to prevent race conditions + if _kafka_producer is not None: + return _kafka_producer + + conf = current_app.config + brokers = conf.get('IIB_KAFKA_BROKERS') + if not brokers: + return None + + try: + username = conf.get('IIB_KAFKA_USERNAME') + password = conf.get('IIB_KAFKA_PASSWORD') + + if not username or not password: + current_app.logger.error('Kafka credentials missing, cannot initialize producer') + return None + + producer_kwargs = { + 'bootstrap_servers': brokers, + 'security_protocol': 'SASL_SSL', + 'sasl_mechanism': 'SCRAM-SHA-512', + 'sasl_plain_username': username, + 'sasl_plain_password': password, + 'value_serializer': lambda v: json.dumps(v).encode('utf-8'), + 'key_serializer': lambda k: str(k).encode('utf-8') if k is not None else None, + 'linger_ms': 5, + } + + ca_file = conf.get('IIB_KAFKA_SSL_CAFILE') + if ca_file: + producer_kwargs['ssl_cafile'] = ca_file + + _kafka_producer = KafkaProducer(**producer_kwargs) + + current_app.logger.info('Kafka producer initialised') + return _kafka_producer + except Exception: + current_app.logger.exception('Failed to initialise KafkaProducer') + return None + + +def on_send_error(excp): + """Log failed Kafka message delivery in a background-thread-safe way.""" + import logging + + logging.getLogger(__name__).error('Failed to deliver message to Kafka', exc_info=excp) + + +def send_kafka_message( + producer: KafkaProducer, + topic: str, + content: Union[BaseClassRequestResponse, BatchRequestResponseList], + properties: Optional[Dict[str, Any]] = None, +) -> None: + """Send a single message to a Kafka topic asynchronously.""" + properties = properties or {} + headers = [(k, str(v).encode('utf-8')) for k, v in properties.items() if v is not None] + message_key = properties.get('id', properties.get('batch')) + + try: + current_app.logger.info(f'Queuing message for Kafka topic {topic}') + + future = producer.send(topic, key=message_key, value=content, headers=headers) + + # Attach callbacks for async failure handling + future.add_errback(on_send_error) + + except Exception: + current_app.logger.exception(f'Failed to queue message for Kafka topic {topic}') diff --git a/iib/web/messaging.py b/iib/web/messaging.py index 4e0a3a202..a20aa8948 100644 --- a/iib/web/messaging.py +++ b/iib/web/messaging.py @@ -16,6 +16,7 @@ BatchRequestResponseList, ) from iib.web.models import Batch, Request, RequestStateMapping +from iib.web.kafka_producer import get_kafka_producer, send_kafka_message __all__ = ['Envelope', 'json_to_envelope', 'send_messages', 'send_message_for_state_change'] @@ -23,6 +24,74 @@ Envelope = namedtuple('Envelope', 'address message') +def _build_request_state_change_data( + request: Request, +) -> tuple: + """ + Build the content and properties for a request state change message. + + Used by both the legacy AMQP and Kafka paths. + + :param iib.web.models.Request request: the request that changed states + :return: a tuple of (content, properties) + :rtype: tuple + """ + request_json = cast(BaseClassRequestResponse, request.to_json(verbose=False)) + properties = { + 'batch': request_json['batch'], + 'id': request_json['id'], + 'state': request_json['state'], + 'user': request_json['user'], + } + return request_json, properties + + +def _build_batch_state_change_data( + batch: Batch, + new_batch: Optional[bool] = False, +) -> Optional[tuple]: + """ + Build the content and properties for a batch state change message. + + Returns ``None`` when no message should be sent. + Used by both the legacy AMQP and Kafka paths. + + :param iib.web.models.Batch batch: the batch that changed states + :param bool new_batch: if ``True``, a new batch message will be generated + :return: a tuple of (content, properties) or None + :rtype: tuple or None + """ + if new_batch: + batch_state = 'in_progress' + else: + batch_state = batch.state + + if not (new_batch or batch_state in RequestStateMapping.get_final_states()): + return None + + batch_username = getattr(batch.user, 'username', None) + content: BatchRequestResponseList = { + 'batch': batch.id, + 'annotations': batch.annotations, + 'requests': [ + { + 'id': r.id, + 'organization': getattr(r, 'organization', None), + 'request_type': r.type_name, + } + for r in batch.requests + ], + 'state': batch_state, + 'user': batch_username, + } + properties = { + 'batch': batch.id, + 'state': batch_state, + 'user': batch_username, + } + return content, properties + + def _get_batch_state_change_envelope( batch: Batch, new_batch: Optional[bool] = False, @@ -47,34 +116,10 @@ def _get_batch_state_change_envelope( ) return None - if new_batch: - # Avoid querying the database for the batch state since we know it's a new batch - batch_state = 'in_progress' - else: - batch_state = batch.state - - if new_batch or batch_state in RequestStateMapping.get_final_states(): + data = _build_batch_state_change_data(batch, new_batch) + if data: current_app.logger.debug('Preparing to send a state change message for batch %d', batch.id) - batch_username = getattr(batch.user, 'username', None) - content: BatchRequestResponseList = { - 'batch': batch.id, - 'annotations': batch.annotations, - 'requests': [ - { - 'id': request.id, - 'organization': getattr(request, 'organization', None), - 'request_type': request.type_name, - } - for request in batch.requests - ], - 'state': batch_state, - 'user': batch_username, - } - properties = { - 'batch': batch.id, - 'state': batch_state, - 'user': batch_username, - } + content, properties = data return json_to_envelope(batch_address, content, properties) return None @@ -98,15 +143,8 @@ def _get_request_state_change_envelope(request: Request) -> Optional[Envelope]: return None current_app.logger.debug('Preparing to send a state change message for request %d', request.id) - # cast from Union - see Request.to_json - request_json = cast(BaseClassRequestResponse, request.to_json(verbose=False)) - properties = { - 'batch': request_json['batch'], - 'id': request_json['id'], - 'state': request_json['state'], - 'user': request_json['user'], - } - return json_to_envelope(request_address, request_json, properties) + content, properties = _build_request_state_change_data(request) + return json_to_envelope(request_address, content, properties) def _get_ssl_domain() -> Optional[proton.SSLDomain]: @@ -199,6 +237,41 @@ def send_messages(envelopes: List[Envelope]) -> None: connection.close() +def _send_kafka_messages( + requests: List[Request], + batch: Batch, + new_batch: Optional[bool] = False, +) -> None: + """ + Send request and batch state-change messages to Kafka. + + :param list requests: one or more requests whose state changed + :param iib.web.models.Batch batch: the batch associated with the requests + :param bool new_batch: if ``True``, a batch-creation message is sent + """ + try: + producer = get_kafka_producer() + if not producer: + return + + conf = current_app.config + + build_topic = conf.get('IIB_KAFKA_BUILD_STATE_TOPIC') + if build_topic: + for request in requests: + content, properties = _build_request_state_change_data(request) + send_kafka_message(producer, build_topic, content, properties) + + batch_topic = conf.get('IIB_KAFKA_BATCH_STATE_TOPIC') + if batch_topic: + batch_data = _build_batch_state_change_data(batch, new_batch) + if batch_data: + content, properties = batch_data + send_kafka_message(producer, batch_topic, content, properties) + except: # noqa: E722 + current_app.logger.exception('Failed to send one or more messages') + + def send_message_for_state_change(request: Request, new_batch_msg: Optional[bool] = False) -> None: """ Send the appropriate message(s) based on a build request state change. @@ -223,6 +296,8 @@ def send_message_for_state_change(request: Request, new_batch_msg: Optional[bool if envelopes: send_messages(envelopes) + _send_kafka_messages([request], request.batch, new_batch=new_batch_msg) + def send_messages_for_new_batch_of_requests(requests: List[Request]) -> None: """ @@ -250,3 +325,5 @@ def send_messages_for_new_batch_of_requests(requests: List[Request]) -> None: if envelopes: send_messages(envelopes) + + _send_kafka_messages(requests, batch, new_batch=True) diff --git a/requirements.txt b/requirements.txt index 6bab2dc7f..32a5f4c0e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -538,6 +538,10 @@ jmespath==1.1.0 \ # via # boto3 # botocore +kafka-python==3.0.11 \ + --hash=sha256:9d10cab4e11e02545d82c7e5af5702da5aa46dd4eccd11ad92a50bf6dbbecd14 \ + --hash=sha256:a003d927e79c801d6cfd1e59ceaaf78807351e75cdb5b8ee9ce4262586f9780f + # via iib (setup.py) kombu==5.6.2 \ --hash=sha256:8060497058066c6f5aed7c26d7cd0d3b574990b09de842a8c5aaed0b92cc5a55 \ --hash=sha256:efcfc559da324d41d61ca311b0c64965ea35b4c55cc04ee36e55386145dace93 diff --git a/setup.py b/setup.py index a3c1fd898..fe910f56a 100644 --- a/setup.py +++ b/setup.py @@ -17,6 +17,7 @@ 'flask-migrate', 'flask-sqlalchemy', 'importlib-resources', + 'kafka-python', 'kubernetes', 'operator-manifest==0.0.5', 'psycopg2-binary', diff --git a/tests/test_web/test_kafka_producer.py b/tests/test_web/test_kafka_producer.py new file mode 100644 index 000000000..beae41f11 --- /dev/null +++ b/tests/test_web/test_kafka_producer.py @@ -0,0 +1,424 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +from unittest import mock + +import pytest +from kafka.errors import KafkaError + +from iib.web import kafka_producer + + +@pytest.fixture(autouse=True) +def reset_kafka_producer(): + """Reset the module-level cached producer before each test.""" + kafka_producer._kafka_producer = None + yield + kafka_producer._kafka_producer = None + + +class TestGetKafkaProducer: + """Tests for the get_kafka_producer function.""" + + @mock.patch('iib.web.kafka_producer.KafkaProducer') + def test_get_kafka_producer_success(self, mock_kp_class, app): + """Verify the producer is created with correct config values.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker1:9096', 'broker2:9096'] + app.config['IIB_KAFKA_USERNAME'] = 'test_user' + app.config['IIB_KAFKA_PASSWORD'] = 'test_pass' + + producer = kafka_producer.get_kafka_producer() + + assert producer is mock_kp_class.return_value + mock_kp_class.assert_called_once() + call_kwargs = mock_kp_class.call_args[1] + assert call_kwargs['bootstrap_servers'] == ['broker1:9096', 'broker2:9096'] + # security_protocol and sasl_mechanism are hardcoded, not configurable + assert call_kwargs['security_protocol'] == 'SASL_SSL' + assert call_kwargs['sasl_mechanism'] == 'SCRAM-SHA-512' + assert call_kwargs['sasl_plain_username'] == 'test_user' + assert call_kwargs['sasl_plain_password'] == 'test_pass' + assert call_kwargs['linger_ms'] == 5 + + @mock.patch('iib.web.kafka_producer.KafkaProducer') + def test_get_kafka_producer_ssl_cafile_default(self, mock_kp_class, app): + """Verify the default ssl_cafile from config is passed to the producer.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker:9096'] + app.config['IIB_KAFKA_USERNAME'] = 'u' + app.config['IIB_KAFKA_PASSWORD'] = 'p' + + kafka_producer.get_kafka_producer() + + call_kwargs = mock_kp_class.call_args[1] + assert call_kwargs['ssl_cafile'] == '/etc/pki/tls/certs/ca-bundle.crt' + + @mock.patch('iib.web.kafka_producer.KafkaProducer') + def test_get_kafka_producer_ssl_cafile_custom(self, mock_kp_class, app): + """Verify a custom ssl_cafile is passed when configured.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker:9096'] + app.config['IIB_KAFKA_USERNAME'] = 'u' + app.config['IIB_KAFKA_PASSWORD'] = 'p' + app.config['IIB_KAFKA_SSL_CAFILE'] = '/custom/ca-bundle.crt' + + kafka_producer.get_kafka_producer() + + call_kwargs = mock_kp_class.call_args[1] + assert call_kwargs['ssl_cafile'] == '/custom/ca-bundle.crt' + + @mock.patch('iib.web.kafka_producer.KafkaProducer') + def test_get_kafka_producer_ssl_cafile_omitted_when_falsy(self, mock_kp_class, app): + """Verify ssl_cafile is omitted from kwargs when explicitly unset.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker:9096'] + app.config['IIB_KAFKA_USERNAME'] = 'u' + app.config['IIB_KAFKA_PASSWORD'] = 'p' + app.config['IIB_KAFKA_SSL_CAFILE'] = None + + kafka_producer.get_kafka_producer() + + call_kwargs = mock_kp_class.call_args[1] + assert 'ssl_cafile' not in call_kwargs + + def test_get_kafka_producer_missing_username(self, app): + """Verify None is returned and no producer is created when username is missing.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker:9096'] + app.config['IIB_KAFKA_PASSWORD'] = 'pass' + + producer = kafka_producer.get_kafka_producer() + + assert producer is None + + def test_get_kafka_producer_missing_password(self, app): + """Verify None is returned and no producer is created when password is missing.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker:9096'] + app.config['IIB_KAFKA_USERNAME'] = 'user' + + producer = kafka_producer.get_kafka_producer() + + assert producer is None + + def test_get_kafka_producer_missing_credentials(self, app): + """Verify None is returned when neither username nor password is configured.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker:9096'] + + producer = kafka_producer.get_kafka_producer() + + assert producer is None + + def test_get_kafka_producer_no_brokers(self, app): + """Verify None is returned when brokers is None.""" + app.config['IIB_KAFKA_BROKERS'] = None + + producer = kafka_producer.get_kafka_producer() + + assert producer is None + + def test_get_kafka_producer_empty_brokers(self, app): + """Verify None is returned when brokers list is empty.""" + app.config['IIB_KAFKA_BROKERS'] = [] + + producer = kafka_producer.get_kafka_producer() + + assert producer is None + + @mock.patch('iib.web.kafka_producer.KafkaProducer') + def test_get_kafka_producer_cached(self, mock_kp_class, app): + """Verify a second call returns the cached producer without creating a new one.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker:9096'] + app.config['IIB_KAFKA_USERNAME'] = 'u' + app.config['IIB_KAFKA_PASSWORD'] = 'p' + + producer1 = kafka_producer.get_kafka_producer() + producer2 = kafka_producer.get_kafka_producer() + + assert producer1 is producer2 + mock_kp_class.assert_called_once() + + @mock.patch('iib.web.kafka_producer.KafkaProducer') + def test_get_kafka_producer_init_failure(self, mock_kp_class, app): + """Verify None is returned when KafkaProducer init raises an exception.""" + app.config['IIB_KAFKA_BROKERS'] = ['broker:9096'] + app.config['IIB_KAFKA_USERNAME'] = 'u' + app.config['IIB_KAFKA_PASSWORD'] = 'p' + mock_kp_class.side_effect = KafkaError('connection failed') + + producer = kafka_producer.get_kafka_producer() + + assert producer is None + + +class TestSendKafkaMessage: + """Tests for the send_kafka_message function.""" + + def test_send_kafka_message_success(self, app): + """Verify message is sent with correct topic, key, value, and headers.""" + mock_producer = mock.Mock() + mock_future = mock.Mock() + mock_producer.send.return_value = mock_future + + content = {'id': 1, 'state': 'complete', 'batch': 1} + properties = {'id': 1, 'state': 'complete', 'user': 'tbrady'} + + kafka_producer.send_kafka_message( + mock_producer, 'dev.eng.iib.build.state', content, properties + ) + + mock_producer.send.assert_called_once_with( + 'dev.eng.iib.build.state', + key=1, + value=content, + headers=[ + ('id', b'1'), + ('state', b'complete'), + ('user', b'tbrady'), + ], + ) + mock_future.add_errback.assert_called_once_with(kafka_producer.on_send_error) + + def test_send_kafka_message_none_values_filtered_from_headers(self, app): + """Verify None property values are excluded from Kafka headers.""" + mock_producer = mock.Mock() + mock_future = mock.Mock() + mock_producer.send.return_value = mock_future + + content = {'batch': 5, 'state': 'in_progress'} + properties = {'batch': 5, 'state': 'in_progress', 'user': None} + + kafka_producer.send_kafka_message( + mock_producer, 'dev.eng.iib.batch.state', content, properties + ) + + call_kwargs = mock_producer.send.call_args[1] + assert call_kwargs['headers'] == [ + ('batch', b'5'), + ('state', b'in_progress'), + ] + + def test_send_kafka_message_batch_key(self, app): + """When no 'id' in properties, the message key should fall back to 'batch'.""" + mock_producer = mock.Mock() + mock_future = mock.Mock() + mock_producer.send.return_value = mock_future + + content = {'batch': 5, 'state': 'in_progress'} + properties = {'batch': 5, 'state': 'in_progress', 'user': None} + + kafka_producer.send_kafka_message( + mock_producer, 'dev.eng.iib.batch.state', content, properties + ) + + call_kwargs = mock_producer.send.call_args[1] + assert call_kwargs['key'] == 5 + + def test_send_kafka_message_no_properties(self, app): + """Verify message works with no properties passed.""" + mock_producer = mock.Mock() + mock_future = mock.Mock() + mock_producer.send.return_value = mock_future + + kafka_producer.send_kafka_message(mock_producer, 'dev.eng.iib.build.state', {'id': 1}) + + call_kwargs = mock_producer.send.call_args[1] + assert call_kwargs['headers'] == [] + assert call_kwargs['key'] is None + + def test_send_kafka_message_kafka_error(self, app): + """Verify KafkaError is caught and does not raise.""" + mock_producer = mock.Mock() + mock_producer.send.side_effect = KafkaError('send failed') + + kafka_producer.send_kafka_message( + mock_producer, 'dev.eng.iib.build.state', {'id': 1}, {'id': 1} + ) + + # Should not raise — error is logged + + def test_send_kafka_message_non_kafka_error(self, app): + """ + Verify non-KafkaError exceptions are also caught and do not raise. + + ``send_kafka_message`` catches ``Exception`` broadly (not just ``KafkaError``) + since this is a best-effort, fire-and-forget notification path that must never + crash the caller (e.g. a JSON serialization ``TypeError`` from a bad payload). + """ + mock_producer = mock.Mock() + mock_producer.send.side_effect = TypeError('serialization error') + + kafka_producer.send_kafka_message( + mock_producer, 'dev.eng.iib.build.state', {'id': 1}, {'id': 1} + ) + + # Should not raise — error is logged + + def test_send_kafka_message_unexpected_error(self, app): + """Verify arbitrary unexpected exceptions (not just TypeError) are also swallowed.""" + mock_producer = mock.Mock() + mock_producer.send.side_effect = RuntimeError('unexpected failure') + + kafka_producer.send_kafka_message( + mock_producer, 'dev.eng.iib.build.state', {'id': 1}, {'id': 1} + ) + + # Should not raise — error is logged + + def test_on_send_error_logs(self): + """Verify on_send_error logs the exception without raising.""" + exc = KafkaError('async delivery failed') + with mock.patch('logging.getLogger') as mock_get_logger: + kafka_producer.on_send_error(exc) + + mock_get_logger.return_value.error.assert_called_once() + + +class TestSendKafkaMessages: + """Tests for the _send_kafka_messages function in messaging.py.""" + + @mock.patch('iib.web.messaging.send_kafka_message') + @mock.patch('iib.web.messaging.get_kafka_producer') + def test_send_kafka_messages_no_producer( + self, mock_get_producer, mock_send_msg, app, db, minimal_request_add + ): + """When the producer is None, no messages should be sent.""" + mock_get_producer.return_value = None + + from iib.web.messaging import _send_kafka_messages + + _send_kafka_messages([minimal_request_add], minimal_request_add.batch) + + mock_send_msg.assert_not_called() + + @mock.patch('iib.web.messaging.send_kafka_message') + @mock.patch('iib.web.messaging.get_kafka_producer') + def test_send_kafka_messages_build_and_batch( + self, mock_get_producer, mock_send_msg, app, db, minimal_request_add + ): + """Verify both build and batch messages are sent on new batch.""" + mock_producer = mock.Mock() + mock_get_producer.return_value = mock_producer + app.config['IIB_KAFKA_BUILD_STATE_TOPIC'] = 'dev.eng.iib.build.state' + app.config['IIB_KAFKA_BATCH_STATE_TOPIC'] = 'dev.eng.iib.batch.state' + + minimal_request_add.add_state('in_progress', 'Starting') + db.session.commit() + + from iib.web.messaging import _send_kafka_messages + + _send_kafka_messages([minimal_request_add], minimal_request_add.batch, new_batch=True) + + assert mock_send_msg.call_count == 2 + build_call = mock_send_msg.call_args_list[0] + assert build_call[0][0] is mock_producer + assert build_call[0][1] == 'dev.eng.iib.build.state' + + batch_call = mock_send_msg.call_args_list[1] + assert batch_call[0][0] is mock_producer + assert batch_call[0][1] == 'dev.eng.iib.batch.state' + + @mock.patch('iib.web.messaging.send_kafka_message') + @mock.patch('iib.web.messaging.get_kafka_producer') + def test_send_kafka_messages_no_topics( + self, mock_get_producer, mock_send_msg, app, db, minimal_request_add + ): + """When topics are not configured, no messages should be sent.""" + mock_get_producer.return_value = mock.Mock() + app.config.pop('IIB_KAFKA_BUILD_STATE_TOPIC', None) + app.config.pop('IIB_KAFKA_BATCH_STATE_TOPIC', None) + + from iib.web.messaging import _send_kafka_messages + + _send_kafka_messages([minimal_request_add], minimal_request_add.batch) + + mock_send_msg.assert_not_called() + + @mock.patch('iib.web.messaging.send_kafka_message') + @mock.patch('iib.web.messaging.get_kafka_producer') + def test_send_kafka_messages_batch_not_final( + self, mock_get_producer, mock_send_msg, app, db, minimal_request_add + ): + """Batch message should not be sent when batch is in_progress and new_batch is False.""" + mock_get_producer.return_value = mock.Mock() + app.config['IIB_KAFKA_BUILD_STATE_TOPIC'] = 'dev.eng.iib.build.state' + app.config['IIB_KAFKA_BATCH_STATE_TOPIC'] = 'dev.eng.iib.batch.state' + + minimal_request_add.add_state('in_progress', 'Starting') + db.session.commit() + + from iib.web.messaging import _send_kafka_messages + + _send_kafka_messages([minimal_request_add], minimal_request_add.batch, new_batch=False) + + # Only build message, no batch message (batch is in_progress, not final) + assert mock_send_msg.call_count == 1 + assert mock_send_msg.call_args_list[0][0][1] == 'dev.eng.iib.build.state' + + @mock.patch('iib.web.messaging.send_kafka_message') + @mock.patch('iib.web.messaging.get_kafka_producer') + def test_send_kafka_messages_exception_nonfatal( + self, mock_get_producer, mock_send_msg, app, db, minimal_request_add + ): + """Verify that exceptions in _send_kafka_messages do not propagate.""" + mock_get_producer.return_value = mock.Mock() + app.config['IIB_KAFKA_BUILD_STATE_TOPIC'] = 'dev.eng.iib.build.state' + mock_send_msg.side_effect = RuntimeError('unexpected error') + + from iib.web.messaging import _send_kafka_messages + + # Should not raise — error is caught and logged + _send_kafka_messages([minimal_request_add], minimal_request_add.batch) + + +class TestDualPublishIntegration: + """ + Verify dual-publish calls _send_kafka_messages alongside AMQP. + + Covers send_message_for_state_change and send_messages_for_new_batch_of_requests. + """ + + @mock.patch('iib.web.messaging._send_kafka_messages') + @mock.patch('iib.web.messaging.send_messages') + @mock.patch('iib.web.messaging._get_batch_state_change_envelope') + @mock.patch('iib.web.messaging._get_request_state_change_envelope') + def test_send_message_for_state_change_calls_kafka( + self, + mock_req_env, + mock_batch_env, + mock_send_amqp, + mock_send_kafka, + app, + db, + minimal_request_add, + ): + """Verify send_message_for_state_change invokes _send_kafka_messages.""" + mock_req_env.return_value = mock.Mock() + mock_batch_env.return_value = None + + from iib.web.messaging import send_message_for_state_change + + send_message_for_state_change(minimal_request_add, new_batch_msg=False) + + mock_send_kafka.assert_called_once_with( + [minimal_request_add], minimal_request_add.batch, new_batch=False + ) + + @mock.patch('iib.web.messaging._send_kafka_messages') + @mock.patch('iib.web.messaging.send_messages') + @mock.patch('iib.web.messaging._get_batch_state_change_envelope') + @mock.patch('iib.web.messaging._get_request_state_change_envelope') + def test_send_messages_for_new_batch_calls_kafka( + self, + mock_req_env, + mock_batch_env, + mock_send_amqp, + mock_send_kafka, + app, + db, + minimal_request_add, + ): + """Verify send_messages_for_new_batch_of_requests invokes _send_kafka_messages.""" + mock_req_env.return_value = mock.Mock() + mock_batch_env.return_value = None + + from iib.web.messaging import send_messages_for_new_batch_of_requests + + send_messages_for_new_batch_of_requests([minimal_request_add]) + + mock_send_kafka.assert_called_once_with( + [minimal_request_add], minimal_request_add.batch, new_batch=True + ) diff --git a/tests/test_web/test_messaging.py b/tests/test_web/test_messaging.py index a3d93c456..113c4112d 100644 --- a/tests/test_web/test_messaging.py +++ b/tests/test_web/test_messaging.py @@ -231,6 +231,7 @@ def test_send_messages_nonfatal(mock_gsd, mock_bc, app): @pytest.mark.parametrize('request_msg_expected', (True, False)) @pytest.mark.parametrize('batch_msg_expected', (True, False)) +@mock.patch('iib.web.messaging._send_kafka_messages') @mock.patch('iib.web.messaging._get_request_state_change_envelope') @mock.patch('iib.web.messaging._get_batch_state_change_envelope') @mock.patch('iib.web.messaging.send_messages') @@ -238,6 +239,7 @@ def test_send_message_for_state_change( mock_sm, mock_gbsce, mock_grstce, + mock_skm, batch_msg_expected, request_msg_expected, app, @@ -269,11 +271,18 @@ def test_send_message_for_state_change( @pytest.mark.parametrize('request_msg_expected', (True, False)) @pytest.mark.parametrize('batch_msg_expected', (True, False)) +@mock.patch('iib.web.messaging._send_kafka_messages') @mock.patch('iib.web.messaging._get_request_state_change_envelope') @mock.patch('iib.web.messaging._get_batch_state_change_envelope') @mock.patch('iib.web.messaging.send_messages') def test_send_messages_for_new_batch_of_requests( - mock_sm, mock_gbsce, mock_grsce, batch_msg_expected, request_msg_expected, minimal_request_add + mock_sm, + mock_gbsce, + mock_grsce, + mock_skm, + batch_msg_expected, + request_msg_expected, + minimal_request_add, ): expected_msgs = [] if request_msg_expected: @@ -300,8 +309,11 @@ def test_send_messages_for_new_batch_of_requests( mock_sm.assert_not_called() +@mock.patch('iib.web.messaging._send_kafka_messages') @mock.patch('iib.web.messaging.send_messages') -def test_send_messages_for_new_batch_of_requests_no_requests(mock_sm, minimal_request_add): +def test_send_messages_for_new_batch_of_requests_no_requests( + mock_sm, mock_skm, minimal_request_add +): messaging.send_messages_for_new_batch_of_requests([]) mock_sm.assert_not_called()