added integration test for wildcard subscriptions

Fixes: #414
Adds an integration test that verifies that the all
signals when starting and stopping a unit will be
received by a wildcard monitor subscription with a
setup of multiple nodes.

Signed-off-by: Michael Engel <mengel@redhat.com>

tests/tests/tier0/monitor-wildcard-unit-changes/main.fmf

@@ -0,0 +1 @@
+summary: Test if the proper virtual signals are emitted when a monitor with wildcard subscription is active and a node disconnects and reconnects again

tests/tests/tier0/monitor-wildcard-unit-changes/python/monitor.py

@@ -0,0 +1,95 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+import unittest
+
+from dasbus.loop import EventLoop
+
+from bluechi.api import Manager, Monitor, Node, Structure
+
+node_name_foo = "node-foo"
+node_name_bar = "node-bar"
+
+service_simple = "simple.service"
+service_also_simple = "also-simple.service"
+
+
+class CallCounter:
+
+    def __init__(self) -> None:
+        self.times_new_has_been_called = 0
+        self.times_removed_has_been_called = 0
+        self.times_state_change_has_been_called = 0
+        self.times_property_change_has_been_called = 0
+
+
+class TestMonitorWildcardUnitChanges(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.call_counter = {
+            node_name_foo: {service_simple: CallCounter()},
+            node_name_bar: {service_also_simple: CallCounter()},
+        }
+
+        self.loop = EventLoop()
+        self.mgr = Manager()
+        self.monitor = Monitor(self.mgr.create_monitor())
+
+        def on_unit_new(node: str, unit: str, reason: str) -> None:
+            if node in self.call_counter and unit in self.call_counter[node]:
+                self.call_counter[node][unit].times_new_has_been_called += 1
+
+        def on_unit_removed(node: str, unit: str, reason: str) -> None:
+            if node in self.call_counter and unit in self.call_counter[node]:
+                self.call_counter[node][unit].times_removed_has_been_called += 1
+
+                foo_simple = self.call_counter[node_name_foo][service_simple]
+                bar_also_simple = self.call_counter[node_name_bar][service_also_simple]
+                if foo_simple.times_removed_has_been_called > 0 and bar_also_simple.times_removed_has_been_called > 0:
+                    self.loop.quit()
+
+        def on_unit_state_changed(node: str, unit: str, active_state: str, sub_state: str, reason: str) -> None:
+            if node in self.call_counter and unit in self.call_counter[node]:
+                self.call_counter[node][unit].times_state_change_has_been_called += 1
+
+        def on_unit_property_changed(node: str, unit: str, interface: str, props: Structure) -> None:
+            if node in self.call_counter and unit in self.call_counter[node]:
+                self.call_counter[node][unit].times_property_change_has_been_called += 1
+
+        self.monitor.on_unit_new(on_unit_new)
+        self.monitor.on_unit_removed(on_unit_removed)
+        self.monitor.on_unit_state_changed(on_unit_state_changed)
+        self.monitor.on_unit_properties_changed(on_unit_property_changed)
+
+    def test_monitor_wildcard_unit_changes(self):
+        node_foo = Node(node_name_foo)
+        node_bar = Node(node_name_bar)
+
+        # start subscription on all nodes and units
+        self.monitor.subscribe('*', '*')
+
+        assert node_foo.start_unit(service_simple, "replace") != ""
+        assert node_bar.start_unit(service_also_simple, "replace") != ""
+
+        # will stop when the unit removed signal has been received for
+        # both services on the respective nodes
+        self.loop.run()
+
+        # verify that the signals for those events were received
+        # (at least once since the number of calls is a systemd internal)
+
+        foo_simple = self.call_counter[node_name_foo][service_simple]
+        bar_also_simple = self.call_counter[node_name_bar][service_also_simple]
+
+        assert foo_simple.times_new_has_been_called >= 1
+        assert foo_simple.times_removed_has_been_called >= 1
+        assert foo_simple.times_state_change_has_been_called >= 1
+        assert foo_simple.times_property_change_has_been_called >= 1
+
+        assert bar_also_simple.times_new_has_been_called >= 1
+        assert bar_also_simple.times_removed_has_been_called >= 1
+        assert bar_also_simple.times_state_change_has_been_called >= 1
+        assert bar_also_simple.times_property_change_has_been_called >= 1
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/tests/tier0/monitor-wildcard-unit-changes/systemd/also-simple.service

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

tests/tests/tier0/monitor-wildcard-unit-changes/systemd/simple.service

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

tests/tests/tier0/monitor-wildcard-unit-changes/test_monitor_wildcard_unit_changes.py

@@ -0,0 +1,55 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+import os
+import pytest
+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"
+node_name_bar = "node-bar"
+
+service_simple = "simple.service"
+service_also_simple = "also-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")
+
+    nodes[node_name_bar].copy_systemd_service(
+        service_also_simple, "systemd", os.path.join("/", "etc", "systemd", "system"))
+    assert nodes[node_name_bar].wait_for_unit_state_to_be(service_also_simple, "inactive")
+
+    result, output = ctrl.run_python(os.path.join("python", "monitor.py"))
+    if result != 0:
+        raise Exception(output)
+
+
+@pytest.mark.timeout(15)
+def test_monitor_wildcard_node_reconnect(
+        bluechi_test: BluechiTest,
+        bluechi_ctrl_default_config: BluechiControllerConfig,
+        bluechi_node_default_config: BluechiNodeConfig):
+
+    config_node_foo = bluechi_node_default_config.deep_copy()
+    config_node_bar = bluechi_node_default_config.deep_copy()
+
+    config_node_foo.node_name = node_name_foo
+    config_node_bar.node_name = node_name_bar
+
+    bluechi_ctrl_default_config.allowed_node_names = [
+        node_name_foo,
+        node_name_bar,
+    ]
+
+    bluechi_test.set_bluechi_controller_config(bluechi_ctrl_default_config)
+    bluechi_test.add_bluechi_node_config(config_node_foo)
+    bluechi_test.add_bluechi_node_config(config_node_bar)
+
+    bluechi_test.run(exec)