xBis7 commented on code in PR #73276:
URL: https://github.com/apache/airflow/pull/73276#discussion_r4292555013
##########
providers/apache/kafka/tests/unit/apache/kafka/plugins/test_event_producer.py:
##########
@@ -326,56 +379,98 @@ def test_hook_defaults_when_options_unset(self):
class TestCheckTopicExists:
"""Tests for the topic check and its retry-after-failure cooldown."""
+ @pytest.fixture
+ def topic_type(self, request) -> event_producer.EventProducerKafkaTopic:
+ return request.param
+
@pytest.fixture(autouse=True)
- def _topic_conf(self):
- with conf_vars({("kafka_event_producer", "topic"): _KAFKA_TOPIC}):
+ def _topic_conf(self, dagrun_topic, task_instance_topic):
+ with conf_vars(
+ {
+ ("kafka_event_producer", "dagrun_topic"): dagrun_topic,
+ ("kafka_event_producer", "task_instance_topic"):
task_instance_topic,
+ }
+ ):
yield
- def test_topic_exists_kept_for_process_lifetime(self):
+ def test_topic_exists_kept_for_process_lifetime(self, topic_type):
kafka_producer_mock = MagicMock()
kafka_producer_mock.list_topics.return_value.topics.__contains__.return_value =
True
with patch(_PRODUCER_CLS, return_value=kafka_producer_mock):
- assert event_producer._check_topic_exists() is True
+ assert event_producer._check_topic_exists(topic_type) is True
# Once confirmed, the broker is not queried again.
- assert event_producer._check_topic_exists() is True
+ assert event_producer._check_topic_exists(topic_type) is True
kafka_producer_mock.list_topics.assert_called_once()
- def test_topic_doesnt_exist(self, monkeypatch):
+ def test_topic_doesnt_exist(self, monkeypatch, topic_type, dagrun_topic,
task_instance_topic):
kafka_producer_mock = MagicMock()
kafka_producer_mock.list_topics.return_value.topics.__contains__.return_value =
False
log_warning_mock = MagicMock()
monkeypatch.setattr(event_producer.log, "warning", log_warning_mock)
with patch(_PRODUCER_CLS, return_value=kafka_producer_mock):
- assert event_producer._check_topic_exists() is False
+ assert event_producer._check_topic_exists(topic_type) is False
log_warning_mock.assert_called_once()
# Format string + the two formatting args are passed positionally to
log.warning.
warning_format, topic_arg, retry_interval_arg =
log_warning_mock.call_args.args
assert "topic %r not found on the broker" in warning_format
- assert topic_arg == _KAFKA_TOPIC
+ if topic_type == event_producer.EventProducerKafkaTopic.DAG_RUN:
+ assert topic_arg == dagrun_topic
+ elif topic_type ==
event_producer.EventProducerKafkaTopic.TASK_INSTANCE:
+ assert topic_arg == task_instance_topic
assert retry_interval_arg ==
event_producer._get_topic_check_retry_interval()
- def test_list_topics_failure_warns_and_sets_cooldown(self, monkeypatch):
+ def test_one_topic_doesnt_exist(self, monkeypatch, dagrun_topic,
task_instance_topic):
+ kafka_producer_mock = MagicMock()
+
kafka_producer_mock.list_topics.return_value.topics.__contains__.side_effect =
[True, False]
+
+ log_warning_mock = MagicMock()
+ monkeypatch.setattr(event_producer.log, "warning", log_warning_mock)
+
+ with (
+ patch(_PRODUCER_CLS, return_value=kafka_producer_mock),
+
patch("airflow.providers.apache.kafka.plugins.event_producer.time.monotonic",
return_value=60.0),
+ ):
+ assert
event_producer._check_topic_exists(event_producer.EventProducerKafkaTopic.TASK_INSTANCE)
+ assert
event_producer._check_topic_exists(event_producer.EventProducerKafkaTopic.DAG_RUN)
is False
Review Comment:
These two checks should be reversed because what we are trying to check is
that if the Dag run topic doesn't exist, calling the function again will still
check that the task instance topic does.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
