Merge branch Master into 8.4.x-sync

Author: wxtim

changes.d/6711.fix.md

@@ -0,0 +1 @@
+Stop broadcast allowing `[remote]host` if `platform` set, or vice-versa

cylc/flow/broadcast_mgr.py

@@ -33,6 +33,10 @@
 from cylc.flow.parsec.util import listjoin, pdeepcopy, poverride
 from cylc.flow.parsec.validate import BroadcastConfigValidator
 from cylc.flow.run_modes import WORKFLOW_ONLY_MODES
+from cylc.flow.platforms import (
+    fail_if_platform_and_host_conflict,
+    PlatformLookupError,
+)
 
 
 if TYPE_CHECKING:
@@ -64,10 +68,11 @@ class BroadcastMgr:
 
     REC_SECTION = re.compile(r"\[([^\]]+)\]")
 
-    def __init__(self, workflow_db_mgr, data_store_mgr, run_mode):
-        self.workflow_run_mode = run_mode
-        self.workflow_db_mgr = workflow_db_mgr
-        self.data_store_mgr = data_store_mgr
+    def __init__(self, schd):
+        self.schd = schd
+        self.workflow_run_mode = schd.get_run_mode()
+        self.workflow_db_mgr = schd.workflow_db_mgr
+        self.data_store_mgr = schd.data_store_mgr
         self.linearized_ancestors = {}
         self.broadcasts = {}
         self.ext_triggers = {}  # Can use collections.Counter in future
@@ -320,6 +325,23 @@ def put_broadcast(
                         if namespace not in self.linearized_ancestors:
                             bad_namespaces.append(namespace)
                         elif not bad_point:
+                            # Check broadcast against config and against
+                            # existing broadcasts:
+                            newconfig = pdeepcopy(self.schd.config.get_config(
+                                ['runtime', namespace]
+                            ))
+                            poverride(
+                                newconfig,
+                                self.broadcasts.get(point_string, {})
+                                .get(namespace, {})
+                            )
+                            if self.bc_mixes_old_and_new_platform_settings(
+                                newconfig,
+                                namespace,
+                                coerced_setting,
+                            ):
+                                continue
+
                             if namespace not in self.broadcasts[point_string]:
                                 self.broadcasts[point_string][namespace] = {}
 
@@ -348,6 +370,23 @@ def put_broadcast(
             self.data_store_mgr.delta_broadcast()
         return modified_settings, bad_options
 
+    @staticmethod
+    def bc_mixes_old_and_new_platform_settings(
+        task_config, namespace, coerced_setting
+    ):
+        """Check for combination of old ([remote]host) and new (platform)
+        settings in the task config as it will be after merger.
+        """
+        task_config.update(coerced_setting)
+        try:
+            fail_if_platform_and_host_conflict(
+                task_config, namespace
+            )
+            return False
+        except PlatformLookupError as exc:
+            LOG.error('Cannot apply broadcast:\n' + '\n    '.join(exc.args))
+            return True
+
     @staticmethod
     def _cancel_keys_in_prunes(prunes, cancel_keys):
         """Is cancel_keys pruned?"""

cylc/flow/scheduler.py

@@ -402,8 +402,7 @@ async def initialise(self):
 
         """
         self.data_store_mgr = DataStoreMgr(self)
-        self.broadcast_mgr = BroadcastMgr(
-            self.workflow_db_mgr, self.data_store_mgr, self.get_run_mode())
+        self.broadcast_mgr = BroadcastMgr(self)
 
         self.server = WorkflowRuntimeServer(self)
 

tests/integration/test_broadcast_mgr.py

@@ -0,0 +1,70 @@
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Tests for Broadcast Manager."""
+
+
+async def test_reject_valid_broadcast_is_remote_clash_with_config(
+    one_conf, flow, start, scheduler, log_filter
+):
+    """`put_broadcast` gracefully rejects invalid broadcast:
+
+    Existing config = [task][remote]host = foo
+    Broadcast       = [task]platform = bar
+
+    https://github.com/cylc/cylc-flow/issues/6693
+    """
+    one_conf.update({'runtime': {'root': {'platform': 'foo'}}})
+    wid = flow(one_conf)
+    schd = scheduler(wid)
+    async with start(schd):
+        bc_mgr = schd.broadcast_mgr
+        bc_mgr.put_broadcast(
+            point_strings=['1'],
+            namespaces=['one'],
+            settings=[{'remote': {'host': 'bar'}}]
+        )
+        assert log_filter(contains='Cannot apply broadcast')
+        assert bc_mgr.broadcasts == {'1': {}}
+
+
+async def test_reject_valid_broadcast_is_remote_clash_with_broadcast(
+    one_conf, flow, start, scheduler, log_filter
+):
+    """`put_broadcast` gracefully rejects invalid broadcast:
+
+    Existing Broadcast = [task][remote]host = foo
+    New Broadcast      = [task]platform = bar
+
+    https://github.com/cylc/cylc-flow/pull/6711/files#r2033457964
+    """
+    schd = scheduler(flow(one_conf))
+    async with start(schd):
+        bc_mgr = schd.broadcast_mgr
+        bc_mgr.put_broadcast(
+            point_strings=['1'],
+            namespaces=['one'],
+            settings=[{'remote': {'host': 'bar'}}]
+        )
+        # this should not be allowed, if it is the scheduler will crash
+        # when unpaused:
+        bc_mgr.put_broadcast(
+            point_strings=['1'],
+            namespaces=['one'],
+            settings=[{'platform': 'foo'}]
+        )
+        assert log_filter(contains='Cannot apply broadcast')
+        assert bc_mgr.broadcasts == {'1': {'one': {'remote': {'host': 'bar'}}}}