broadcast: report invalid settings

* We already report invalid cycle points and namespaces.
* However, we have been relying on client-side validation for settings
  (which doesn't apply to the GraphQL mutation).
* This also raises the potential for inter-version compatibility issues
  going unreported.
* This commit explicitly handles invalid settings in the same way as
  invalid cycle points and namespaces so that they are reported back to
  the user.
* Closes the issue part of #6429.
* Additionally:
  * This change also strips duplicate entries from broadcast reports.
  * And fixes the CLI options in broadcast report to match `cylc
    broadcast`.

Author: oliver-sanders

changes.d/6574.fix.md

@@ -0,0 +1 @@
+Broadcast: Report any settings that are not compatible with the scheduler Cylc version.

cylc/flow/broadcast_mgr.py

@@ -37,7 +37,7 @@
     fail_if_platform_and_host_conflict,
     PlatformLookupError,
 )
-
+from cylc.flow.util import uniq
 
 if TYPE_CHECKING:
     from cylc.flow.id import Tokens
@@ -151,7 +151,7 @@ def clear_broadcast(
             LOG.error(get_broadcast_bad_options_report(bad_options))
         if modified_settings:
             self.data_store_mgr.delta_broadcast()
-        return modified_settings, bad_options
+        return uniq(modified_settings), bad_options
 
     def expire_broadcast(self, cutoff=None, **kwargs):
         """Clear all broadcasts targeting cycle points earlier than cutoff."""
@@ -284,18 +284,24 @@ def put_broadcast(
           bad_options is as described in the docstring for self.clear().
         """
         modified_settings = []
-        bad_point_strings = []
-        bad_namespaces = []
+        bad_settings = []
+        bad_point_strings = set()
+        bad_namespaces = set()
 
         with self.lock:
             for setting in settings or []:
                 # Coerce setting to cylc runtime object,
                 # i.e. str to  DurationFloat.
                 coerced_setting = deepcopy(setting)
-                BroadcastConfigValidator().validate(
-                    coerced_setting,
-                    SPEC['runtime']['__MANY__'],
-                )
+                try:
+                    BroadcastConfigValidator().validate(
+                        coerced_setting,
+                        SPEC['runtime']['__MANY__'],
+                    )
+                except Exception as exc:
+                    LOG.error(exc)
+                    bad_settings.append(setting)
+                    continue
 
                 # Skip and warn if a run mode is broadcast to a workflow
                 # running in simulation or dummy mode.
@@ -308,22 +314,24 @@ def put_broadcast(
                         f' running in {self.workflow_run_mode.value} mode'
                         ' will have no effect, and will not be actioned.'
                     )
+                    bad_settings.append(setting)
                     continue
 
                 for point_string in point_strings or []:
                     # Standardise the point and check its validity.
                     bad_point = False
                     try:
                         point_string = standardise_point_string(point_string)
+
                     except PointParsingError:
                         if point_string != '*':
-                            bad_point_strings.append(point_string)
+                            bad_point_strings.add(point_string)
                             bad_point = True
                     if not bad_point and point_string not in self.broadcasts:
                         self.broadcasts[point_string] = {}
                     for namespace in namespaces or []:
                         if namespace not in self.linearized_ancestors:
-                            bad_namespaces.append(namespace)
+                            bad_namespaces.add(namespace)
                         elif not bad_point:
                             # Check broadcast against config and against
                             # existing broadcasts:
@@ -340,6 +348,7 @@ def put_broadcast(
                                 namespace,
                                 coerced_setting,
                             ):
+                                bad_settings.append(setting)
                                 continue
 
                             if namespace not in self.broadcasts[point_string]:
@@ -362,13 +371,15 @@ def put_broadcast(
         LOG.info(get_broadcast_change_report(modified_settings))
 
         bad_options = {}
+        if bad_settings:
+            bad_options["settings"] = uniq(bad_settings)
         if bad_point_strings:
-            bad_options["point_strings"] = bad_point_strings
+            bad_options["point_strings"] = sorted(bad_point_strings)
         if bad_namespaces:
-            bad_options["namespaces"] = bad_namespaces
+            bad_options["namespaces"] = sorted(bad_namespaces)
         if modified_settings:
             self.data_store_mgr.delta_broadcast()
-        return modified_settings, bad_options
+        return uniq(modified_settings), bad_options
 
     @staticmethod
     def bc_mixes_old_and_new_platform_settings(

cylc/flow/broadcast_report.py

@@ -26,6 +26,13 @@
 CHANGE_TITLE_CANCEL = "Broadcast cancelled:"
 CHANGE_TITLE_SET = "Broadcast set:"
 
+CLI_OPT_MAP = {
+    # broadcast field: cli option
+    'point_strings': 'point',
+    'namespaces': 'namespace',
+    'settings': 'set',
+}
+
 
 def get_broadcast_bad_options_report(bad_options, is_set=False):
     """Return a string to report bad options for broadcast cancel/clear."""
@@ -48,7 +55,11 @@ def get_broadcast_bad_options_report(bad_options, is_set=False):
                         value_str += val
             else:
                 value_str = value
-            msg += BAD_OPTIONS_FMT % (key, value_str)
+            if isinstance(value, dict):
+                value_str = ', '.join(
+                    f'{key}={val}' for key, val in value.items()
+                )
+            msg += BAD_OPTIONS_FMT % (CLI_OPT_MAP.get(key, key), value_str)
     return msg
 
 

cylc/flow/network/resolvers.py

@@ -55,6 +55,7 @@
     runtime_schema_to_cfg,
     sort_elements,
 )
+from cylc.flow.util import uniq, iter_uniq
 
 if TYPE_CHECKING:
     from uuid import UUID
@@ -112,45 +113,6 @@ def collate_workflow_atts(workflow):
     }
 
 
-def uniq(iterable):
-    """Return a unique collection of the provided items preserving item order.
-
-    Useful for unhashable things like dicts, relies on __eq__ for testing
-    equality.
-
-    Examples:
-        >>> uniq([1, 1, 2, 3, 5, 8, 1])
-        [1, 2, 3, 5, 8]
-
-    """
-    ret = []
-    for item in iterable:
-        if item not in ret:
-            ret.append(item)
-    return ret
-
-
-def iter_uniq(iterable):
-    """Iterate over an iterable omitting any duplicate entries.
-
-    Useful for unhashable things like dicts, relies on __eq__ for testing
-    equality.
-
-    Note:
-        More efficient than "uniq" for iteration use cases.
-
-    Examples:
-        >>> list(iter_uniq([1, 1, 2, 3, 5, 8, 1]))
-        [1, 2, 3, 5, 8]
-
-    """
-    cache = set()
-    for item in iterable:
-        if item not in cache:
-            cache.add(item)
-            yield item
-
-
 def workflow_ids_filter(workflow_tokens, items) -> bool:
     """Match id arguments with workflow attributes.
 
@@ -787,24 +749,31 @@ def broadcast(
         cutoff: Any = None
     ):
         """Put or clear broadcasts."""
-        if settings is not None:
-            # Convert schema field names to workflow config setting names if
-            # applicable:
-            for i, dict_ in enumerate(settings):
-                settings[i] = runtime_schema_to_cfg(dict_)
-
-        if mode == 'put_broadcast':
-            return self.schd.task_events_mgr.broadcast_mgr.put_broadcast(
-                cycle_points, namespaces, settings)
-        if mode == 'clear_broadcast':
-            return self.schd.task_events_mgr.broadcast_mgr.clear_broadcast(
-                point_strings=cycle_points,
-                namespaces=namespaces,
-                cancel_settings=settings)
-        if mode == 'expire_broadcast':
-            return self.schd.task_events_mgr.broadcast_mgr.expire_broadcast(
-                cutoff)
-        raise ValueError(f"Unsupported broadcast mode: '{mode}'")
+        try:
+            if settings is not None:
+                # Convert schema field names to workflow config setting names
+                # if applicable:
+                for i, dict_ in enumerate(settings):
+                    settings[i] = runtime_schema_to_cfg(dict_)
+
+            if mode == 'put_broadcast':
+                return self.schd.task_events_mgr.broadcast_mgr.put_broadcast(
+                    cycle_points, namespaces, settings)
+            if mode == 'clear_broadcast':
+                return self.schd.task_events_mgr.broadcast_mgr.clear_broadcast(
+                    point_strings=cycle_points,
+                    namespaces=namespaces,
+                    cancel_settings=settings)
+            if mode == 'expire_broadcast':
+                return (
+                    self.schd.task_events_mgr.broadcast_mgr.expire_broadcast(
+                        cutoff
+                    )
+                )
+            raise ValueError(f'Unsupported broadcast mode: {mode}')
+        except Exception as exc:
+            LOG.error(exc)
+            return {'error': {'message': str(exc)}}
 
     def put_ext_trigger(
         self,

cylc/flow/network/schema.py

@@ -1795,17 +1795,6 @@ class Arguments:
             )
         )
 
-        # TODO: work out how to implement this feature, it needs to be
-        #       handled client-side which makes it slightly awkward in
-        #       api-on-the-fly land
-
-        # files = graphene.List(
-        #    String,
-        #    description=sstrip('''
-        #        File with config to broadcast. Can be used multiple times
-        #    ''')
-        # )
-
     result = GenericScalar()
 
 

cylc/flow/util.py

@@ -45,6 +45,45 @@
 _NAT_SORT_SPLIT = re.compile(r'([\d\.]+)')
 
 
+def uniq(iterable):
+    """Return a unique collection of the provided items preserving item order.
+
+    Useful for unhashable things like dicts, relies on __eq__ for testing
+    equality.
+
+    Examples:
+        >>> uniq([1, 1, 2, 3, 5, 8, 1])
+        [1, 2, 3, 5, 8]
+
+    """
+    ret = []
+    for item in iterable:
+        if item not in ret:
+            ret.append(item)
+    return ret
+
+
+def iter_uniq(iterable):
+    """Iterate over an iterable omitting any duplicate entries.
+
+    Useful for unhashable things like dicts, relies on __eq__ for testing
+    equality.
+
+    Note:
+        More efficient than "uniq" for iteration use cases.
+
+    Examples:
+        >>> list(iter_uniq([1, 1, 2, 3, 5, 8, 1]))
+        [1, 2, 3, 5, 8]
+
+    """
+    cache = set()
+    for item in iterable:
+        if item not in cache:
+            cache.add(item)
+            yield item
+
+
 def sstrip(text):
     """Simple function to dedent and strip text.
 

tests/functional/broadcast/00-simple/expected-prep.err

@@ -1,14 +1,14 @@
 ERROR: No broadcast to cancel/clear for these options:
   --cancel=[environment]BCAST
-  --namespaces=m4
-  --point_strings=20100809T00
+  --namespace=m4
+  --point=20100809T00
 ERROR: No broadcast to cancel/clear for these options:
   --cancel=[environment]BCAST
-  --namespaces=m5
-  --point_strings=*
+  --namespace=m5
+  --point=*
 ERROR: No broadcast to cancel/clear for these options:
-  --namespaces=m6
+  --namespace=m6
 ERROR: No broadcast to cancel/clear for these options:
-  --namespaces=m7
+  --namespace=m7
 ERROR: No broadcast to cancel/clear for these options:
-  --namespaces=ENS3
+  --namespace=ENS3

tests/integration/run_modes/test_simulation.py

@@ -424,10 +424,12 @@ async def test_settings_broadcast(
 
         # Assert that setting run mode on a simulation mode task fails with
         # warning:
-        schd.broadcast_mgr.put_broadcast(
+        good, bad = schd.broadcast_mgr.put_broadcast(
             ['1066'], ['one'], [{
                 'run mode': 'live',
             }])
+        assert good == []
+        assert bad == {'settings': [{'run mode': 'live'}]}
         record = log_filter(contains='will not be actioned')[0]
         assert record[0] == logging.WARNING
         assert 'run mode' not in schd.broadcast_mgr.broadcasts