Make core event registration able to support multiple topics (#2505)

* Make core event registration able to support multiple topics

Co-authored-by: Đỗ Trường Giang <do.giang@avaiga.com>

taipy/core/notification/_registration.py

@@ -10,7 +10,7 @@
 # specific language governing permissions and limitations under the License.
 
 from queue import SimpleQueue
-from typing import Optional
+from typing import Optional, Set
 from uuid import uuid4
 
 from ._topic import _Topic
@@ -23,17 +23,21 @@ class _Registration:
     _ID_PREFIX = "REGISTRATION"
     __SEPARATOR = "_"
 
-    def __init__(
-        self,
+    def __init__(self) -> None:
+        self.registration_id: RegistrationId = self._new_id()
+        self.queue: SimpleQueue = SimpleQueue()
+        self.topics: Set[_Topic] = set()
+
+    @staticmethod
+    def from_topic(
         entity_type: Optional[EventEntityType] = None,
         entity_id: Optional[str] = None,
         operation: Optional[EventOperation] = None,
         attribute_name: Optional[str] = None,
-    ):
-
-        self.registration_id: str = self._new_id()
-        self.topic: _Topic = _Topic(entity_type, entity_id, operation, attribute_name)
-        self.queue: SimpleQueue = SimpleQueue()
+    ) -> "_Registration":
+        reg = _Registration()
+        reg.topics.add(_Topic(entity_type, entity_id, operation, attribute_name))
+        return reg
 
     @staticmethod
     def _new_id() -> RegistrationId:
@@ -42,3 +46,37 @@ class _Registration:
 
     def __hash__(self) -> int:
         return hash(self.registration_id)
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, _Registration):
+            return False
+        return self.registration_id == other.registration_id
+
+    def __ne__(self, other: object) -> bool:
+        return not self.__eq__(other)
+
+    def __str__(self) -> str:
+        return f"Registration ID: {self.registration_id}, Topics: {self.topics}"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
+    def add_topic(
+        self,
+        entity_type: Optional[EventEntityType] = None,
+        entity_id: Optional[str] = None,
+        operation: Optional[EventOperation] = None,
+        attribute_name: Optional[str] = None,
+    ) -> None:
+        """Add a topic to the registration."""
+        self.topics.add(_Topic(entity_type, entity_id, operation, attribute_name))
+
+    def remove_topic(
+        self,
+        entity_type: Optional[EventEntityType] = None,
+        entity_id: Optional[str] = None,
+        operation: Optional[EventOperation] = None,
+        attribute_name: Optional[str] = None,
+    ) -> None:
+        """Remove a topic from the registration."""
+        self.topics.remove(_Topic(entity_type, entity_id, operation, attribute_name))

taipy/core/notification/_topic.py

@@ -53,3 +53,12 @@ class _Topic:
         ):
             return True
         return False
+
+    def __ne__(self, __value) -> bool:
+        return not self.__eq__(__value)
+
+    def __str__(self) -> str:
+        return (f"Topic: {self.entity_type}, "
+                f" {self.entity_id}, "
+                f" {self.operation}, "
+                f" {self.attribute_name}")

taipy/core/notification/notifier.py

@@ -122,14 +122,8 @@ class Notifier:
         Returns:
             A tuple containing the registration id and the event queue.
         """
-        registration = _Registration(entity_type, entity_id, operation, attribute_name)
-
-        if registrations := cls._topics_registrations_list.get(registration.topic, None):
-            registrations.add(registration)
-        else:
-            cls._topics_registrations_list[registration.topic] = {registration}
-
-        return registration.registration_id, registration.queue
+        registration = _Registration.from_topic(entity_type, entity_id, operation, attribute_name)
+        return cls.__do_register(registration)
 
     @classmethod
     def unregister(cls, registration_id: str) -> None:
@@ -153,16 +147,20 @@ class Notifier:
         to_remove_registration: Optional[_Registration] = None
 
         for _, registrations in cls._topics_registrations_list.items():
-            for registration in registrations:
-                if registration.registration_id == registration_id:
-                    to_remove_registration = registration
+            for reg in registrations:
+                if reg.registration_id == registration_id:
+                    to_remove_registration = reg
                     break
+            if to_remove_registration:
+                break
 
         if to_remove_registration:
-            registrations = cls._topics_registrations_list[to_remove_registration.topic]
-            registrations.remove(to_remove_registration)
-            if len(registrations) == 0:
-                del cls._topics_registrations_list[to_remove_registration.topic]
+            for topic in to_remove_registration.topics:
+                if topic in cls._topics_registrations_list:
+                    registrations = cls._topics_registrations_list[topic]
+                    registrations.remove(to_remove_registration)
+                    if len(registrations) == 0:
+                        del cls._topics_registrations_list[topic]
 
     @classmethod
     def publish(cls, event: Event) -> None:
@@ -171,10 +169,25 @@ class Notifier:
         Arguments:
             event (`Event^`): The event to publish.
         """
+        processed_registrations: Set[_Registration] = set()
         for topic, registrations in cls._topics_registrations_list.items():
             if Notifier._is_matching(event, topic):
                 for registration in registrations:
-                    registration.queue.put(event)
+                    if registration not in processed_registrations:
+                        registration.queue.put(event)
+                        processed_registrations.add(registration)
+
+    @classmethod
+    def _register_from_registration(cls, registration: _Registration) -> Tuple[str, SimpleQueue]:
+        """Register a listener from a registration object.
+
+        Arguments:
+            registration (`_Registration`): The registration object.
+
+        Returns:
+            A tuple containing the registration id and the event queue.
+        """
+        return cls.__do_register(registration)
 
     @staticmethod
     def _is_matching(event: Event, topic: _Topic) -> bool:
@@ -188,3 +201,12 @@ class Notifier:
         if topic.attribute_name is not None and event.attribute_name and event.attribute_name != topic.attribute_name:
             return False
         return True
+
+    @classmethod
+    def __do_register(cls, registration: _Registration) -> Tuple[str, SimpleQueue]:
+        for topic in registration.topics:
+            if registrations := cls._topics_registrations_list.get(topic, None):
+                registrations.add(registration)
+            else:
+                cls._topics_registrations_list[topic] = {registration}
+        return registration.registration_id, registration.queue

tests/core/conftest.py

@@ -320,7 +320,7 @@ def cleanup_files():
 
 
 @pytest.fixture(scope="function", autouse=True)
-def clean_repository(init_config, init_managers, init_orchestrator, init_notifier, clean_argparser):
+def clean_core(init_config, init_managers, init_orchestrator, init_notifier, clean_argparser):
     clean_argparser()
     close_all_sessions()
     init_config()

tests/core/notification/test_notifier.py

@@ -15,7 +15,7 @@ from taipy.common.config import Config
 from taipy.core import taipy as tp
 from taipy.core._version._version_manager_factory import _VersionManagerFactory
 from taipy.core.common.frequency import Frequency
-from taipy.core.notification import EventEntityType, EventOperation
+from taipy.core.notification import EventEntityType, EventOperation, _Registration
 from taipy.core.notification._topic import _Topic
 from taipy.core.notification.event import Event
 from taipy.core.notification.notifier import Notifier
@@ -23,7 +23,7 @@ from taipy.core.scenario._scenario_manager_factory import _ScenarioManagerFactor
 from taipy.core.submission.submission_status import SubmissionStatus
 
 
-def test_register():
+def test_register_unregister():
     def find_registration_and_topic(registration_id):
         for topic, registrations in Notifier._topics_registrations_list.items():
             for registration in registrations:
@@ -87,6 +87,8 @@ def test_register():
     registration_id_5, register_queue_5 = Notifier.register(EventEntityType.SCENARIO)
     topic_5, registration_5 = find_registration_and_topic(registration_id_5)
 
+    assert topic_5 == topic_2
+    assert registration_5 != registration_2
     assert isinstance(registration_id_5, str) and registration_id_5 == registration_5.registration_id
     assert isinstance(register_queue_5, SimpleQueue)
     assert len(Notifier._topics_registrations_list.keys()) == 4
@@ -95,12 +97,15 @@ def test_register():
     assert register_queue_5 in [registration.queue for registration in Notifier._topics_registrations_list[topic_5]]
 
     registration_id_6, register_queue_6 = Notifier.register()
+    topic_6, registration_6 = find_registration_and_topic(registration_id_6)
+    assert topic_6 == topic_0
+    assert registration_6 != registration_0
     assert len(Notifier._topics_registrations_list.keys()) == 4
-    assert len(Notifier._topics_registrations_list[topic_0]) == 3
+    assert len(Notifier._topics_registrations_list[topic_6]) == 3
 
     Notifier.unregister(registration_id_6)
     assert len(Notifier._topics_registrations_list.keys()) == 4
-    assert len(Notifier._topics_registrations_list[topic_0]) == 2
+    assert len(Notifier._topics_registrations_list[topic_6]) == 2
 
     Notifier.unregister(registration_id_4)
     assert len(Notifier._topics_registrations_list.keys()) == 3
@@ -249,9 +254,9 @@ def test_matching():
         ),
         _Topic(EventEntityType.SEQUENCE, "sequence_id", EventOperation.UPDATE, "tasks"),
     )
-    assert Notifier._is_matching(Event(EventEntityType.TASK, "task_id", EventOperation.DELETION), _Topic())
+    assert Notifier._is_matching(Event(EventEntityType.TASK, EventOperation.DELETION, "task_id"), _Topic())
     assert Notifier._is_matching(
-        Event(EventEntityType.TASK, "task_id", EventOperation.DELETION), _Topic(EventEntityType.TASK)
+        Event(EventEntityType.TASK, EventOperation.DELETION, "task_id"), _Topic(EventEntityType.TASK)
     )
     assert Notifier._is_matching(
         Event(entity_type=EventEntityType.TASK, entity_id="task_id", operation=EventOperation.DELETION),
@@ -315,7 +320,7 @@ def test_publish_creation_event():
     dn_config = Config.configure_data_node("dn_config")
     task_config = Config.configure_task("task_config", print, [dn_config])
     scenario_config = Config.configure_scenario(
-        "scenario_config", [task_config], frequency=Frequency.DAILY, flag="test"
+        "scenario_config", [task_config], frequency=Frequency.DAILY, properties={"flag": "test"}
     )
     scenario_config.add_sequences({"sequence_config": [task_config]})
 
@@ -352,15 +357,15 @@ def test_publish_creation_event():
 
 
 def test_publish_update_event():
-    _, registration_queue = Notifier.register()
 
     dn_config = Config.configure_data_node("dn_config")
     task_config = Config.configure_task("task_config", print, [dn_config])
     scenario_config = Config.configure_scenario(
-        "scenario_config", [task_config], frequency=Frequency.DAILY, flag="test"
+        "scenario_config", [task_config], frequency=Frequency.DAILY, properties={"flag": "test"}
     )
     scenario_config.add_sequences({"sequence_config": [task_config]})
 
+    _, registration_queue = Notifier.register()
     scenario = _ScenarioManagerFactory._build_manager()._create(scenario_config)
     cycle = scenario.cycle
     task = scenario.tasks[task_config.id]
@@ -549,15 +554,14 @@ def test_publish_update_event():
 
 
 def test_publish_update_event_in_context_manager():
-    _, registration_queue = Notifier.register()
-
     dn_config = Config.configure_data_node("dn_config")
     task_config = Config.configure_task("task_config", print, [dn_config])
     scenario_config = Config.configure_scenario(
-        "scenario_config", [task_config], frequency=Frequency.DAILY, flag="test"
+        "scenario_config", [task_config], frequency=Frequency.DAILY, properties={"flag": "test"}
     )
     scenario_config.add_sequences({"sequence_config": [task_config]})
 
+    _, registration_queue = Notifier.register()
     scenario = _ScenarioManagerFactory._build_manager()._create(scenario_config)
     cycle = scenario.cycle
     task = scenario.tasks[task_config.id]
@@ -700,7 +704,7 @@ def test_publish_submission_event():
     dn_config = Config.configure_data_node("dn_config")
     task_config = Config.configure_task("task_config", print, [dn_config])
     scenario_config = Config.configure_scenario(
-        "scenario_config", [task_config], frequency=Frequency.DAILY, flag="test"
+        "scenario_config", [task_config], frequency=Frequency.DAILY, properties={"flag": "test"}
     )
     scenario_config.add_sequences({"sequence_config": [task_config]})
     scenario = _ScenarioManagerFactory._build_manager()._create(scenario_config)
@@ -767,7 +771,7 @@ def test_publish_deletion_event():
     dn_config = Config.configure_data_node("dn_config")
     task_config = Config.configure_task("task_config", print, [dn_config])
     scenario_config = Config.configure_scenario(
-        "scenario_config", [task_config], frequency=Frequency.DAILY, flag="test"
+        "scenario_config", [task_config], frequency=Frequency.DAILY, properties={"flag": "test"}
     )
     scenario_config.add_sequences({"sequence_config": [task_config]})
     scenario = _ScenarioManagerFactory._build_manager()._create(scenario_config)
@@ -844,3 +848,101 @@ def test_publish_deletion_event():
         and event.attribute_name is None
         for i, event in enumerate(published_events)
     )
+
+
+def find_registration_and_topics(registration_id: str):
+    topics = set()
+    registration = None
+    for topic, registrations in Notifier._topics_registrations_list.items():
+        for reg in registrations:
+            if reg.registration_id == registration_id:
+                topics.add(topic)
+                registration = reg
+                break
+    return topics, registration
+
+
+def test_register_no_topic():
+    assert len(Notifier._topics_registrations_list) == 0
+    empty_registration = _Registration()
+
+    id, queue = Notifier._register_from_registration(empty_registration)
+
+    assert id == empty_registration.registration_id
+    assert queue == empty_registration.queue
+    assert len(Notifier._topics_registrations_list) == 0
+    retrieved_topics, retrieved_registration = find_registration_and_topics(id)
+    assert retrieved_topics == set()
+    assert retrieved_registration is None
+
+    # Unregister the registration
+    Notifier.unregister(id)
+    assert len(Notifier._topics_registrations_list) == 0
+    retrieved_topics, retrieved_registration = find_registration_and_topics(id)
+    assert retrieved_topics == set()
+    assert retrieved_registration is None
+
+
+def test_register_multiple_topics():
+    assert len(Notifier._topics_registrations_list) == 0
+
+    # Registration_0 with 4 topics
+    registration_0 = _Registration()
+    registration_0.add_topic(EventEntityType.SCENARIO)
+    registration_0.add_topic(EventEntityType.TASK, "task_id")
+    registration_0.add_topic(EventEntityType.DATA_NODE, operation=EventOperation.CREATION)
+    registration_0.add_topic(operation=EventOperation.DELETION)
+    id_0, queue_0 = Notifier._register_from_registration(registration_0)
+
+    assert id_0 == registration_0.registration_id
+    assert queue_0 == registration_0.queue
+    assert len(registration_0.topics) == 4
+    assert len(Notifier._topics_registrations_list) == 4
+    for topic in registration_0.topics:
+        assert topic in Notifier._topics_registrations_list
+        assert len(Notifier._topics_registrations_list[topic]) == 1
+        assert registration_0 in Notifier._topics_registrations_list[topic]
+    retrieved_topics, retrieved_registration = find_registration_and_topics(id_0)
+    assert retrieved_topics == registration_0.topics
+    assert retrieved_registration == registration_0
+
+    # Registration_1 with 3 common topics and 1 new topic
+    registration_1 = _Registration()
+    registration_1.add_topic(EventEntityType.SCENARIO)
+    registration_1.add_topic(EventEntityType.TASK, "task_id")
+    registration_1.add_topic(EventEntityType.DATA_NODE, operation=EventOperation.UPDATE)
+    registration_1.add_topic(operation=EventOperation.DELETION)
+    id_1, queue_1 = Notifier._register_from_registration(registration_1)
+
+    retrieved_topics, retrieved_registration = find_registration_and_topics(id_1)
+    assert retrieved_topics == registration_1.topics
+    assert retrieved_registration == registration_1 != registration_0
+    assert id_1 == registration_1.registration_id != id_0
+    assert queue_1 == registration_1.queue != queue_0
+    assert len(registration_1.topics) == 4
+    assert len(Notifier._topics_registrations_list) == 5
+    for topic in registration_1.topics:
+        assert topic in Notifier._topics_registrations_list
+        if topic.operation == EventOperation.UPDATE:
+            assert len(Notifier._topics_registrations_list[topic]) == 1
+        else:
+            assert len(Notifier._topics_registrations_list[topic]) == 2
+        assert registration_1 in Notifier._topics_registrations_list[topic]
+
+    # Unregister registration_0
+    Notifier.unregister(id_0)
+    retrieved_topics, retrieved_registration = find_registration_and_topics(id_0)
+    assert retrieved_topics == set()
+    assert retrieved_registration is None
+    assert len(Notifier._topics_registrations_list) == 4 # The 4 topics from registration_1
+    for topic in registration_1.topics:
+        assert topic in Notifier._topics_registrations_list
+        assert len(Notifier._topics_registrations_list[topic]) == 1
+        assert registration_1 in Notifier._topics_registrations_list[topic]
+
+    # Unregister registration_1
+    Notifier.unregister(id_1)
+    retrieved_topics, retrieved_registration = find_registration_and_topics(id_1)
+    assert retrieved_topics == set()
+    assert retrieved_registration is None
+    assert len(Notifier._topics_registrations_list) == 0

tests/core/notification/test_registration.py

@@ -18,31 +18,45 @@ from taipy.core.notification._topic import _Topic
 
 
 def test_create_registration():
-    registration_0 = _Registration()
+    registration = _Registration()
+    assert isinstance(registration.registration_id, str)
+    assert registration.registration_id.startswith(_Registration._ID_PREFIX)
+    assert isinstance(registration.queue, SimpleQueue)
+    assert registration.queue.qsize() == 0
+    assert len(registration.topics) == 0
+
+def test_create_registration_from_topic():
+    registration_0 = _Registration.from_topic()
     assert isinstance(registration_0.registration_id, str)
     assert registration_0.registration_id.startswith(_Registration._ID_PREFIX)
     assert isinstance(registration_0.queue, SimpleQueue)
     assert registration_0.queue.qsize() == 0
-    assert isinstance(registration_0.topic, _Topic)
-    assert registration_0.topic.entity_type is None
-    assert registration_0.topic.entity_id is None
-    assert registration_0.topic.operation is None
-    assert registration_0.topic.attribute_name is None
+    assert len(registration_0.topics) == 1
+    topic_0 = registration_0.topics.pop()
+    assert isinstance(topic_0, _Topic)
+    assert topic_0.entity_type is None
+    assert topic_0.entity_id is None
+    assert topic_0.operation is None
+    assert topic_0.attribute_name is None
 
-    registration_1 = _Registration(
-        entity_type=EventEntityType.SCENARIO, entity_id="SCENARIO_scenario_id", operation=EventOperation.CREATION
+    registration_1 = _Registration.from_topic(
+        entity_type=EventEntityType.SCENARIO,
+        entity_id="SCENARIO_scenario_id",
+        operation=EventOperation.CREATION
     )
     assert isinstance(registration_1.registration_id, str)
     assert registration_1.registration_id.startswith(_Registration._ID_PREFIX)
     assert isinstance(registration_1.queue, SimpleQueue)
     assert registration_1.queue.qsize() == 0
-    assert isinstance(registration_1.topic, _Topic)
-    assert registration_1.topic.entity_type == EventEntityType.SCENARIO
-    assert registration_1.topic.entity_id == "SCENARIO_scenario_id"
-    assert registration_1.topic.operation == EventOperation.CREATION
-    assert registration_1.topic.attribute_name is None
+    assert len(registration_1.topics) == 1
+    topic_1 = registration_1.topics.pop()
+    assert isinstance(topic_1, _Topic)
+    assert topic_1.entity_type == EventEntityType.SCENARIO
+    assert topic_1.entity_id == "SCENARIO_scenario_id"
+    assert topic_1.operation == EventOperation.CREATION
+    assert topic_1.attribute_name is None
 
-    registration_2 = _Registration(
+    registration_2 = _Registration.from_topic(
         entity_type=EventEntityType.SEQUENCE,
         entity_id="SEQUENCE_scenario_id",
         operation=EventOperation.UPDATE,
@@ -52,8 +66,102 @@ def test_create_registration():
     assert registration_2.registration_id.startswith(_Registration._ID_PREFIX)
     assert isinstance(registration_2.queue, SimpleQueue)
     assert registration_2.queue.qsize() == 0
-    assert isinstance(registration_2.topic, _Topic)
-    assert registration_2.topic.entity_type == EventEntityType.SEQUENCE
-    assert registration_2.topic.entity_id == "SEQUENCE_scenario_id"
-    assert registration_2.topic.operation == EventOperation.UPDATE
-    assert registration_2.topic.attribute_name == "tasks"
+    topic_2 = registration_2.topics.pop()
+    assert isinstance(topic_2, _Topic)
+    assert topic_2.entity_type == EventEntityType.SEQUENCE
+    assert topic_2.entity_id == "SEQUENCE_scenario_id"
+    assert topic_2.operation == EventOperation.UPDATE
+    assert topic_2.attribute_name == "tasks"
+
+def test_eq():
+    registration = _Registration.from_topic(operation=EventOperation.DELETION)
+    other_registration = _Registration()
+    other_registration.registration_id = registration.registration_id
+    assert registration == other_registration
+
+def test_ne():
+    registration = _Registration.from_topic(operation=EventOperation.DELETION)
+    other_registration = _Registration()
+    assert registration != other_registration
+
+def test_add_topic():
+    registration = _Registration()
+    assert len(registration.topics) == 0
+    registration.add_topic()
+    assert len(registration.topics) == 1
+    topic = registration.topics.pop()
+    assert isinstance(topic, _Topic)
+    assert topic.entity_type is None
+    assert topic.entity_id is None
+    assert topic.operation is None
+    assert topic.attribute_name is None
+
+    registration.add_topic(
+        entity_type=EventEntityType.SCENARIO,
+        entity_id="SCENARIO_scenario_id",
+        operation=EventOperation.CREATION
+    )
+    assert len(registration.topics) == 1
+    topic = registration.topics.pop()
+    assert isinstance(topic, _Topic)
+    assert topic.entity_type == EventEntityType.SCENARIO
+    assert topic.entity_id == "SCENARIO_scenario_id"
+    assert topic.operation == EventOperation.CREATION
+    assert topic.attribute_name is None
+
+    registration.add_topic(
+        entity_type=EventEntityType.SEQUENCE,
+        entity_id="SEQUENCE_scenario_id",
+        operation=EventOperation.UPDATE,
+        attribute_name="tasks",
+    )
+    assert len(registration.topics) == 1
+    topic = registration.topics.pop()
+    assert isinstance(topic, _Topic)
+    assert topic.entity_type == EventEntityType.SEQUENCE
+    assert topic.entity_id == "SEQUENCE_scenario_id"
+    assert topic.operation == EventOperation.UPDATE
+    assert topic.attribute_name == "tasks"
+
+def test_add_remove_topic():
+    registration = _Registration()
+    registration.add_topic()
+    topic_0 = _Topic()
+    registration.add_topic(
+        entity_type=EventEntityType.SCENARIO,
+        entity_id="SCENARIO_scenario_id",
+        operation=EventOperation.CREATION
+    )
+    topic_1 = _Topic(EventEntityType.SCENARIO, "SCENARIO_scenario_id", EventOperation.CREATION)
+    registration.add_topic(
+        entity_type=EventEntityType.SEQUENCE,
+        entity_id="SEQUENCE_scenario_id",
+        operation=EventOperation.UPDATE,
+        attribute_name="tasks",
+    )
+    topic_2 = _Topic(EventEntityType.SEQUENCE, "SEQUENCE_scenario_id", EventOperation.UPDATE, "tasks")
+    assert len(registration.topics) == 3
+    assert topic_0 in registration.topics
+    assert topic_1 in registration.topics
+    assert topic_2 in registration.topics
+
+    registration.remove_topic(
+        entity_type=EventEntityType.SCENARIO,
+        entity_id="SCENARIO_scenario_id",
+        operation=EventOperation.CREATION
+    )
+    assert len(registration.topics) == 2
+    assert topic_0 in registration.topics
+    assert topic_2 in registration.topics
+
+    registration.remove_topic(
+        entity_type=EventEntityType.SEQUENCE,
+        entity_id="SEQUENCE_scenario_id",
+        operation=EventOperation.UPDATE,
+        attribute_name="tasks",
+    )
+    assert len(registration.topics) == 1
+    assert topic_0 in registration.topics
+
+    registration.remove_topic()
+    assert len(registration.topics) == 0

tests/core/notification/test_topic.py

@@ -317,3 +317,15 @@ def test_topic_equal():
     assert _Topic(EventEntityType.JOB, "JOB_id", EventOperation.UPDATE, "status") == _Topic(
         EventEntityType.JOB, "JOB_id", EventOperation.UPDATE, "status"
     )
+
+
+def test_print():
+    topic = _Topic(EventEntityType.TASK, "task_id", EventOperation.UPDATE, "foo")
+    assert "TASK" in str(topic)
+    assert "task_id" in str(topic)
+    assert "UPDATE" in str(topic)
+    assert "foo" in str(topic)
+
+    topic_2 = _Topic(EventEntityType.SCENARIO, None, EventOperation.CREATION, None)
+    assert "SCENARIO" in str(topic_2)
+    assert "CREATION" in str(topic_2)