added intregration tests for AddPeer on Monitor API

src/bindings/python/setup.py

@@ -9,7 +9,7 @@ def readme():
 
 setup(
     name="bluechi",
-    version="0.6.0",
+    version="0.7.0",
     description="Python bindings for BlueChi's D-Bus API",
     long_description=readme(),
     long_description_content_type="text/markdown",

src/manager/monitor.c

@@ -214,7 +214,7 @@ void monitor_close(Monitor *monitor) {
         MonitorPeer *peer = NULL;
         MonitorPeer *next_peer = NULL;
         LIST_FOREACH_SAFE(peers, peer, next_peer, monitor->peers) {
-                monitor_emit_peer_removed(monitor, peer, "closed");
+                monitor_emit_peer_removed(monitor, peer, "monitor closed");
                 LIST_REMOVE(peers, monitor->peers, peer);
                 free_and_null(peer->name);
                 free_and_null(peer);
@@ -421,6 +421,23 @@ static MonitorPeer *monitor_add_peer(Monitor *monitor, const char *peer_name) {
         return peer;
 }
 
+static bool monitor_has_peer_with_name(Monitor *monitor, const char *peer_name) {
+        if (streq(monitor->owner, peer_name)) {
+                return true;
+        }
+
+        MonitorPeer *peer = NULL;
+        MonitorPeer *next_peer = NULL;
+        LIST_FOREACH_SAFE(peers, peer, next_peer, monitor->peers) {
+                if (streq(peer->name, peer_name)) {
+                        return true;
+                }
+        }
+        bc_log_debugf("Peer with name '%s' for monitor '%s' not found", peer_name, monitor->object_path);
+
+        return false;
+}
+
 static int monitor_method_add_peer(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
         Monitor *monitor = userdata;
 
@@ -443,6 +460,11 @@ static int monitor_method_add_peer(sd_bus_message *m, void *userdata, UNUSED sd_
                                 peer_name);
         }
 
+        if (monitor_has_peer_with_name(monitor, peer_name)) {
+                return sd_bus_reply_method_errorf(
+                                m, SD_BUS_ERROR_INVALID_ARGS, "Peer name '%s' has already been added", peer_name);
+        }
+
         MonitorPeer *peer = monitor_add_peer(monitor, peer_name);
         if (peer == NULL) {
                 return sd_bus_reply_method_errorf(
@@ -534,6 +556,8 @@ int monitor_on_unit_property_changed(
                                       monitor->object_path,
                                       strerror(-r));
                 }
+                sd_bus_message_unrefp(&sig);
+                sig = NULL;
         }
 
         MonitorPeer *peer = NULL;
@@ -548,6 +572,8 @@ int monitor_on_unit_property_changed(
                                               peer->name,
                                               strerror(-r));
                         }
+                        sd_bus_message_unrefp(&sig);
+                        sig = NULL;
                 }
         }
 
@@ -569,6 +595,8 @@ int monitor_on_unit_new(void *userdata, const char *node, const char *unit, cons
                                       monitor->object_path,
                                       strerror(-r));
                 }
+                sd_bus_message_unrefp(&sig);
+                sig = NULL;
         }
 
         MonitorPeer *peer = NULL;
@@ -583,6 +611,8 @@ int monitor_on_unit_new(void *userdata, const char *node, const char *unit, cons
                                               peer->name,
                                               strerror(-r));
                         }
+                        sd_bus_message_unrefp(&sig);
+                        sig = NULL;
                 }
         }
 
@@ -610,6 +640,8 @@ int monitor_on_unit_state_changed(
                                       monitor->object_path,
                                       strerror(-r));
                 }
+                sd_bus_message_unrefp(&sig);
+                sig = NULL;
         }
 
         MonitorPeer *peer = NULL;
@@ -624,6 +656,8 @@ int monitor_on_unit_state_changed(
                                               peer->name,
                                               strerror(-r));
                         }
+                        sd_bus_message_unrefp(&sig);
+                        sig = NULL;
                 }
         }
 
@@ -645,6 +679,8 @@ int monitor_on_unit_removed(void *userdata, const char *node, const char *unit,
                                       monitor->object_path,
                                       strerror(-r));
                 }
+                sd_bus_message_unrefp(&sig);
+                sig = NULL;
         }
 
         MonitorPeer *peer = NULL;
@@ -659,6 +695,8 @@ int monitor_on_unit_removed(void *userdata, const char *node, const char *unit,
                                               peer->name,
                                               strerror(-r));
                         }
+                        sd_bus_message_unrefp(&sig);
+                        sig = NULL;
                 }
         }
 

tests/bluechi_test/container.py

@@ -107,6 +107,17 @@ def copy_systemd_service(self, service_file_name: str, source_dir: str, target_d
         self.create_file(target_dir, service_file_name, content)
         self.systemctl_daemon_reload()
 
+    def copy_container_script(self, script_file_name: str):
+        curr_dir = os.getcwd()
+        source_path = os.path.join(curr_dir, "..", "..", "..", "bluechi_test", "container_scripts", script_file_name)
+        content = read_file(source_path)
+
+        target_dir = os.path.join("/", "var")
+
+        LOGGER.info(f"Copy container script '{source_path}' to container path '{curr_dir}'\
+             with content:\n{content}")
+        self.create_file(target_dir, script_file_name, content)
+
     def service_is_active(self, unit_name: str) -> bool:
         result, _ = self.exec_run(f"systemctl is-active {unit_name}")
         return result == 0

tests/bluechi_test/container_scripts/create_monitor.py

@@ -0,0 +1,22 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+import sys
+from bluechi.api import Manager, Monitor
+from dasbus.loop import EventLoop
+
+if len(sys.argv) != 3:
+    print("Usage: python create_monitor.py <node-to-watch> <service-to-watch>")
+    exit(1)
+
+node = sys.argv[1]
+service = sys.argv[2]
+
+loop = EventLoop()
+mgr = Manager()
+monitor_path = mgr.create_monitor()
+monitor = Monitor(monitor_path)
+monitor.subscribe(node, service)
+
+print(monitor_path, flush=True)
+
+loop.run()

tests/bluechi_test/container_scripts/listen.py

@@ -0,0 +1,48 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+import sys
+from bluechi.api import Monitor
+from dasbus.loop import EventLoop
+
+if len(sys.argv) != 2:
+    sys.exit(1)
+
+monitor_path = sys.argv[1]
+
+loop = EventLoop()
+
+monitor = Monitor(monitor_path)
+
+
+def on_new(node, unit, reason):
+    print(f"Unit New: {node} -- {unit}", flush=True)
+
+
+def on_removed(node, unit, reason):
+    print(f"Unit Removed: {node} -- {unit}", flush=True)
+
+
+def on_state(node, unit, active, sub, reason):
+    print(f"Unit state: {node} -- {unit}", flush=True)
+
+
+def on_props(node, unit, interface, props):
+    print(f"Unit props: {node} -- {unit}", flush=True)
+
+
+monitor.on_unit_new(on_new)
+monitor.on_unit_removed(on_removed)
+monitor.on_unit_state_changed(on_state)
+monitor.on_unit_properties_changed(on_props)
+
+
+def on_removed(reason):
+    print(f"Removed, reason: {reason}", flush=True)
+    loop.quit()
+
+
+monitor.get_proxy().PeerRemoved.connect(on_removed)
+
+print(monitor.bus.connection.get_unique_name(), flush=True)
+
+loop.run()

tests/tests/tier0/monitor-listener-added-as-peer/main.fmf

@@ -0,0 +1,2 @@
+summary: Test if a listener on a monitor does receive signals from it when added as peer
+id: 624ba8ae-cf8e-4b17-a8c5-bb09a47ae507

tests/tests/tier0/monitor-listener-added-as-peer/python/added_as_peer.py

@@ -0,0 +1,102 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+import time
+import threading
+import unittest
+from typing import List
+from queue import Queue, Empty
+from subprocess import Popen, PIPE
+
+
+from bluechi.api import Node, Monitor
+
+node = "node-foo"
+service = "simple.service"
+
+
+class TestListenerAddedAsPeer(unittest.TestCase):
+
+    def test_listener_added_as_peer(self):
+
+        process_monitor_owner = Popen(["python3", "/var/create_monitor.py", node,
+                                      service], stdout=PIPE, universal_newlines=True)
+        monitor_path = process_monitor_owner.stdout.readline().strip()
+
+        process_monitor_listener = Popen(["python3", "/var/listen.py", monitor_path],
+                                         stdout=PIPE, universal_newlines=True)
+        bus_id = process_monitor_listener.stdout.readline().strip()
+
+        # add listener as peer so it receives signals from the monitor
+        monitor = Monitor(monitor_path)
+        monitor.add_peer(bus_id)
+
+        # start the unit on the node monitored
+        node_foo = Node(node)
+        node_foo.start_unit(service, "replace")
+
+        events = self.read_no_wait(process_monitor_listener.stdout)
+        assert len(events) > 0
+
+        unit_new_found = False
+        unit_removed_found = False
+        unit_state_changed_found = False
+        unit_prop_changed_found = False
+        for event in events:
+            if event.startswith("Unit New:"):
+                unit_new_found = True
+            elif event.startswith("Unit Removed:"):
+                unit_removed_found = True
+            elif event.startswith("Unit state:"):
+                unit_state_changed_found = True
+            elif event.startswith("Unit props:"):
+                unit_prop_changed_found = True
+
+        assert unit_new_found
+        assert unit_removed_found
+        assert unit_state_changed_found
+        assert unit_prop_changed_found
+
+        process_monitor_owner.terminate()
+        process_monitor_listener.terminate()
+
+    ##################
+    # Helper functions
+
+    worker_queue: Queue = None
+    worker: threading.Thread = None
+
+    def read_no_wait(self, out) -> List[str]:
+        """
+        Helper function.
+        stdout.readline() is a blocking call, use threads to make it non-blocking.
+        """
+
+        def enqueue_output(out, queue):
+            while True:
+                event = out.readline().strip()
+                if event == "":
+                    break
+                queue.put(event)
+
+        if self.worker_queue is None:
+            self.worker_queue = Queue()
+        if self.worker is None:
+            self.worker = threading.Thread(target=enqueue_output, args=(out, self.worker_queue))
+            self.worker.daemon = True
+            self.worker.start()
+
+            # wait a bit for the new worker to catch up
+            time.sleep(0.5)
+
+        res = []
+        while True:
+            try:
+                res.append(self.worker_queue.get_nowait())
+            except Empty:
+                break
+
+        return res
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/tests/tier0/monitor-listener-added-as-peer/systemd/simple.service

@@ -0,0 +1,6 @@
+[Unit]
+Description=Just being true once
+
+[Service]
+Type=simple
+ExecStart=/bin/true

tests/tests/tier0/monitor-listener-added-as-peer/test_monitor_listener_added.py

@@ -0,0 +1,44 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+import os
+from typing import Dict
+
+from bluechi_test.test import BluechiTest
+from bluechi_test.container import BluechiControllerContainer, BluechiNodeContainer
+from bluechi_test.config import BluechiControllerConfig, BluechiNodeConfig
+
+
+node_name_foo = "node-foo"
+service_simple = "simple.service"
+
+
+def exec(ctrl: BluechiControllerContainer, nodes: Dict[str, BluechiNodeContainer]):
+
+    nodes[node_name_foo].copy_systemd_service(
+        service_simple, "systemd", os.path.join("/", "etc", "systemd", "system"))
+    assert nodes[node_name_foo].wait_for_unit_state_to_be(service_simple, "inactive")
+
+    # copy prepared python scripts into container
+    # will be run by not_added_as_peer.py script to create two processes with different bus ids
+    ctrl.copy_container_script("create_monitor.py")
+    ctrl.copy_container_script("listen.py")
+
+    result, output = ctrl.run_python(os.path.join("python", "added_as_peer.py"))
+    if result != 0:
+        raise Exception(output)
+
+
+def test_monitor_listener_added(
+        bluechi_test: BluechiTest,
+        bluechi_ctrl_default_config: BluechiControllerConfig,
+        bluechi_node_default_config: BluechiNodeConfig):
+
+    config_node_foo = bluechi_node_default_config.deep_copy()
+    config_node_foo.node_name = node_name_foo
+
+    bluechi_ctrl_default_config.allowed_node_names = [node_name_foo]
+
+    bluechi_test.set_bluechi_controller_config(bluechi_ctrl_default_config)
+    bluechi_test.add_bluechi_node_config(config_node_foo)
+
+    bluechi_test.run(exec)

tests/tests/tier0/monitor-listener-added-more-than-once/main.fmf

@@ -0,0 +1,2 @@
+summary: Test if a listener on a monitor can only be added once added
+id: 5e694ded-ca05-48f9-94d4-c801e9595f5e