Pub/Sub links: distinguish bad channel based on centre-id

Author: tomkralidis

pywcmp/wcmp2/ets.py

@@ -499,14 +499,20 @@ def test_requirement_links(self):
                 status['message'] = f"invalid link relation {link['rel']}"
                 return status
 
-            LOGGER.debug('Checking that Pub/Sub links have a channel')
+            LOGGER.debug('Checking that Pub/Sub links have a valid channel')
             if link['href'].startswith('mqtt'):
                 if 'channel' not in link:
                     status['code'] = 'FAILED'
                     status['message'] = 'missing channel for Pub/Sub link'
                     return status
 
                 if link['channel'].startswith(('origin/a/wis2', 'cache/a/wis2')):  # noqa
+                    centre_id = link['channel'].split('/')[3]
+                    if centre_id not in self.th.topics[3]:
+                        status['code'] = 'FAILED'
+                        status['message'] = 'Invalid WIS2 topic (bad centre-id) for Pub/Sub link channel'  # noqa
+                        return status
+
                     LOGGER.debug('Validating topic in link channel')
                     if not self.th.validate(link['channel']):
                         status['code'] = 'FAILED'

tests/run_tests.py

@@ -234,6 +234,23 @@ def test_fail_invalid_geometry_range(self):
             self.assertEqual(codes.count('PASSED'), 11)
             self.assertEqual(codes.count('SKIPPED'), 0)
 
+    def test_fail_invalid_link_channel_centre_id(self):
+        """
+        Simple tests for a failing record with an invalid
+        centre-id in a link channel
+        """
+
+        with open(get_test_file_path('data/wcmp2-failing-invalid-link-channel-centre-id.json')) as fh:  # noqa
+            record = json.load(fh)
+            ts = WMOCoreMetadataProfileTestSuite2(record)
+            results = ts.run_tests()
+
+            codes = [r['code'] for r in results['tests']]
+
+            self.assertEqual(codes.count('FAILED'), 1)
+            self.assertEqual(codes.count('PASSED'), 11)
+            self.assertEqual(codes.count('SKIPPED'), 0)
+
 
 class WCMP2KPITest(unittest.TestCase):
     """WCMP KPI tests of tests"""