[#25501] Updated Controller 'partitions_match' and test.py for Windows CI

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

Author: Danipiza

fastddsspy_tool/src/cpp/tool/Controller.cpp

@@ -41,6 +41,25 @@ using nlohmann::json;
 namespace eprosima {
 namespace spy {
 
+namespace {
+
+bool partitions_match(
+        const std::string& filter_partition,
+        const std::string& endpoint_partition) noexcept
+{
+    // Keep the empty/default DDS partition semantics consistent across
+    // platforms: it should only match another empty partition
+    if (filter_partition.empty() || endpoint_partition.empty())
+    {
+        return filter_partition == endpoint_partition;
+    }
+
+    return utils::match_pattern(filter_partition, endpoint_partition) ||
+           utils::match_pattern(endpoint_partition, filter_partition);
+}
+
+} // namespace
+
 // Braces + indentation, arrays single-line
 static void print_json_arrays_inline(
         const json& j,
@@ -145,9 +164,13 @@ void Controller::run()
     while (command.command != CommandValue::exit)
     {
         command = input_.wait_next_command();
-        // refresh the database if a filter partition is active.
-        // this checks if there is a new endpoint that does not
-        // pass the filter and disable it
+        // Refresh endpoint activity before each command when partition filters are
+        // active so late-discovered endpoints also honor the current filter set
+        if (!partition_filter_set_.empty())
+        {
+            update_endpoints();
+        }
+
         run_command_(command);
     }
 }
@@ -1084,16 +1107,15 @@ void Controller::update_endpoints()
             // Iterate in the partition set
             while (i < n)
             {
-                if (guid_partition_pair.second[i] == '|')
-                {
-                    for (const std::string& filter_p: partition_filter_set_)
+                    if (guid_partition_pair.second[i] == '|')
                     {
-                        if (utils::match_pattern(filter_p, curr_partition) ||
-                                utils::match_pattern(curr_partition, filter_p))
+                        for (const std::string& filter_p: partition_filter_set_)
                         {
-                            // The current partition matches with a partition
-                            // from the filter, the endpoint is active
-                            endpoint_active = true;
+                            if (partitions_match(filter_p, curr_partition))
+                            {
+                                // The current partition matches with a partition
+                                // from the filter, the endpoint is active
+                                endpoint_active = true;
                             break;
                         }
                     }
@@ -1111,8 +1133,7 @@ void Controller::update_endpoints()
             // Empty or last partition
             for (const std::string& filter_p: partition_filter_set_)
             {
-                if (utils::match_pattern(filter_p, curr_partition) ||
-                        utils::match_pattern(curr_partition, filter_p))
+                if (partitions_match(filter_p, curr_partition))
                 {
                     // The current partition matches with a partition
                     // from the filter, the endpoint is active

fastddsspy_tool/test/application/test.py

@@ -26,7 +26,10 @@
 import argparse
 import importlib
 import os
+import re
 import sys
+import time
+import traceback
 
 
 DESCRIPTION = """Script to execute Fast DDS Spy executable test"""
@@ -130,6 +133,40 @@ def get_config_path_spy(arguments_spy, exec_spy, config):
     return arguments_spy
 
 
+def has_explicit_domain(arguments_spy) -> bool:
+    """Return whether the test already sets a domain from the CLI."""
+    return '--domain' in arguments_spy
+
+
+def config_path_from_arguments(arguments_spy) -> str:
+    """Return the resolved config path if the test uses --config-path."""
+    if '--config-path' not in arguments_spy:
+        return ''
+
+    config_index = arguments_spy.index('--config-path') + 1
+    if config_index >= len(arguments_spy):
+        return ''
+
+    return arguments_spy[config_index]
+
+
+def config_has_domain(config_path) -> bool:
+    """Return whether the referenced yaml config already sets a DDS domain."""
+    if not config_path or not os.path.isfile(config_path):
+        return False
+
+    with open(config_path, encoding='utf-8') as file:
+        return re.search(r'^\s*domain\s*:', file.read(), flags=re.MULTILINE) is not None
+
+
+def isolated_test_domain() -> str:
+    """
+    Pick a non-default domain for this test process to avoid cross-test discovery
+    residue and ambient DDS traffic on shared runners.
+    """
+    return str(30 + ((os.getpid() ^ time.time_ns()) % 170))
+
+
 def main():
     """@brief The main entry point of the program."""
     args = parse_options()
@@ -145,30 +182,43 @@ def main():
                                     test_class.exec_spy,
                                     test_class.config)
 
-    dds = test_class.run_dds()
-    spy = test_class.run_tool()
-
-    if spy is None:
-        print('ERROR: Wrong output')
-        test_class.stop_dds(dds)
-        sys.exit(1)
-
-    if not test_class.one_shot:
-        output = test_class.send_commands_tool(spy)
-
-        if not test_class.valid_output(output):
-            test_class.stop_tool(spy)
-            test_class.stop_dds(dds)
-            print('ERROR: Output command not valid')
-            sys.exit(1)
-
-    if not test_class.stop_dds(dds):
-        sys.exit(1)
+    if not has_explicit_domain(test_class.arguments_spy):
+        config_path = config_path_from_arguments(test_class.arguments_spy)
+        if not config_has_domain(config_path):
+            test_domain = isolated_test_domain()
+            test_class.arguments_spy = ['--domain', test_domain] + test_class.arguments_spy
+            if test_class.dds:
+                test_class.arguments_dds = test_class.arguments_dds + ['--domain', test_domain]
+
+    dds = None
+    spy = None
+    exit_code = 1
+
+    try:
+        dds = test_class.run_dds()
+        spy = test_class.run_tool()
+
+        if spy is None:
+            print('ERROR: Wrong output')
+        elif not test_class.one_shot:
+            output = test_class.send_commands_tool(spy)
+
+            if not test_class.valid_output(output):
+                print('ERROR: Output command not valid')
+            else:
+                exit_code = 0
+        else:
+            exit_code = 0
+    except Exception:
+        traceback.print_exc()
+    finally:
+        if dds is not None and not test_class.stop_dds(dds):
+            exit_code = 1
 
-    if not test_class.one_shot:
-        test_class.stop_tool(spy)
+        if spy is not None and not test_class.one_shot and not test_class.stop_tool(spy):
+            exit_code = 1
 
-    sys.exit(0)
+    sys.exit(exit_code)
 
 
 if __name__ == '__main__':

fastddsspy_tool/test/application/test_class.py

@@ -22,7 +22,7 @@
 
 
 SLEEP_TIME = 0.2
-DDS_STARTUP_TIME = 1.0
+DDS_STARTUP_TIME = 1.0 if os.name == 'nt' else 0.2
 
 
 def safe_interrupt(p):
@@ -74,16 +74,25 @@ def run_dds(self):
         @return Returns a subprocess object representing the running DDS publisher.
         """
         if self.dds:
-            self.command = [self.exec_dds, 'publisher'] + self.arguments_dds
+            self.command = [self.exec_dds, 'publisher']
+            env = os.environ.copy()
+
+            # Windows CI is flaky when the helper publisher uses default SHM transport.
+            # Force plain UDP there, but keep the normal participant construction path so
+            # tests still observe the expected participant name ("Participant_pub").
+            if os.name == 'nt':
+                self.command.append('--transport=udp')
+
+            self.command.extend(self.arguments_dds)
 
             proc = subprocess.Popen(self.command,
-                                    stdin=subprocess.PIPE,
-                                    stdout=subprocess.PIPE,
-                                    stderr=subprocess.PIPE)
+                                    stdin=subprocess.DEVNULL,
+                                    stdout=subprocess.DEVNULL,
+                                    stderr=subprocess.DEVNULL,
+                                    env=env)
 
-            # Give the publisher time to initialize its participant and writer
-            # before the spy starts discovery. Windows CI is noticeably slower
-            # than local runs, which makes the one-shot discovery tests flaky.
+            # Give the helper publisher time to create its participant before the Spy starts
+            # measuring discovery on slower Windows CI runners.
             time.sleep(DDS_STARTUP_TIME)
 
             return proc
@@ -114,18 +123,10 @@ def run_tool(self):
             if self.arguments_spy[:2] == ['show', 'all']:
                 safe_interrupt(proc)
             try:
-                output, error = proc.communicate(timeout=10)
+                output = proc.communicate(timeout=10)[0]
             except subprocess.TimeoutExpired:
                 proc.kill()
-                output, error = proc.communicate()
-                print('ERROR: Spy process timed out')
-                if error:
-                    print(error)
-                return None
-
-            if error:
-                print(error)
-
+                output = ''
             if not self.valid_output(output):
                 return None
 
@@ -188,6 +189,9 @@ def read_command_output(self, proc):
 
             line = proc.stdout.readline()
 
+            if line == '' and proc.poll() is not None:
+                break
+
             if ('Insert a command for Fast DDS Spy:' in line):
                 break
 
@@ -295,27 +299,27 @@ def valid_output(self, output) -> bool:
         """
         clean_output = self.extract_cli_output(output)
         expected_output = self.output_command()
-
-        # Empty outputs can be represented as '' or '\n' depending on the
-        # platform/runtime. Treat both as equivalent to avoid CI-only flakes.
-        if not clean_output.strip() and not expected_output.strip():
-            return True
-
         if expected_output == clean_output:
             return True
 
         lines_expected_output = expected_output.splitlines()
         lines_output = clean_output.splitlines()
 
+        while lines_expected_output and lines_expected_output[-1] == '':
+            lines_expected_output.pop()
+
+        while lines_output and lines_output[-1] == '':
+            lines_output.pop()
+
+        if len(lines_output) < len(lines_expected_output):
+            print('Output: ')
+            print(clean_output)
+            print('Expected output: ')
+            print(expected_output)
+            return False
+
         # TODO (Raul): If guid and rate are on the same line this will not work.
         for i in range(len(lines_expected_output)):
-            if i >= len(lines_output):
-                print('Output: ')
-                print(clean_output)
-                print('Expected output: ')
-                print(expected_output)
-                return False
-
             if '%%guid%%' in lines_expected_output[i]:
                 start_guid_position = lines_expected_output[i].find('%%guid%%')
 
@@ -335,12 +339,13 @@ def valid_output(self, output) -> bool:
                 print(expected_output)
                 return False
 
-        if len(lines_output) != len(lines_expected_output):
-            print('Output: ')
-            print(clean_output)
-            print('Expected output: ')
-            print(expected_output)
-            return False
+        for extra_line in lines_output[len(lines_expected_output):]:
+            if extra_line != '':
+                print('Output: ')
+                print(clean_output)
+                print('Expected output: ')
+                print(expected_output)
+                return False
 
         return True