Merge pull request #7213 from cylc/8.6.x-sync

🤖 Merge 8.6.x-sync into master

Author: MetRonnie

changes.d/7121.fix.md

@@ -0,0 +1 @@
+Fixed `cylc broadcast` traceback when broadcasting an auto-upgraded deprecated setting.

cylc/flow/cfgspec/workflow.py

@@ -21,6 +21,7 @@
 from typing import (
     Any,
     Dict,
+    Literal,
     Optional,
     Set,
 )
@@ -2102,7 +2103,9 @@ def get_script_common_text(this: str, example: Optional[str] = None):
                 ''')
 
 
-def upg(cfg, descr, for_cancel_broadcast=False):
+def upg(
+    cfg: dict, descr: str, broadcast: bool | Literal["cancel"] = False
+) -> upgrader:
     """Upgrade old workflow configuration.
 
     NOTE: We are silencing deprecation (and only deprecation) warnings
@@ -2111,14 +2114,15 @@ def upg(cfg, descr, for_cancel_broadcast=False):
     warnings and upgrade the syntax).
 
     Args:
-        for_cancel_broadcast:
-            If True, extra validation steps which inspect configuration values
-            will be skipped. This is used for "cylc broadcast --cancel" where
-            the values are not known.
+        broadcast:
+            If truthy, tailors warning messages for the context of broadcasts.
+            If "cancel", extra validation steps which inspect configuration
+            values will be skipped. This is used for
+            "cylc broadcast --cancel" where the values are not known.
             See https://github.com/cylc/cylc-flow/issues/6950.
 
     """
-    u = upgrader(cfg, descr)
+    u = upgrader(cfg, descr, broadcast=bool(broadcast))
 
     u.obsolete(
         '7.8.0', ['runtime', '__MANY__', 'suite state polling', 'template']
@@ -2321,7 +2325,7 @@ def upg(cfg, descr, for_cancel_broadcast=False):
     )
     u.upgrade()
 
-    if not for_cancel_broadcast:
+    if broadcast != "cancel":
         upgrade_graph_section(cfg, descr)
         upgrade_param_env_templates(cfg, descr)
         warn_about_depr_platform(cfg)
@@ -2393,7 +2397,7 @@ def upgrade_param_env_templates(cfg, descr):
                 continue
             if not cylc.flow.flags.cylc7_back_compat:
                 if first_warn:
-                    LOG.warning(upgrader.DEPR_MSG)
+                    LOG.warning(upgrader.depr_msg)
                     first_warn = False
                 LOG.warning(
                     f' * (8.0.0) {dep % task_name} contents prepended to '

cylc/flow/config.py

@@ -2848,7 +2848,7 @@ def _upg_wflow_event_names(self) -> None:
                     )
             if upgraded and not cylc.flow.flags.cylc7_back_compat:
                 LOG.warning(
-                    f"{upgrader.DEPR_MSG}\n"
+                    f"{upgrader.depr_msg}\n"
                     f" * (8.0.0) [scheduler][events][{setting}] "
                     + ', '.join(f'{k} -> {v}' for k, v in upgraded.items())
                 )

cylc/flow/parsec/config.py

@@ -45,11 +45,11 @@ class ParsecConfig:
     def __init__(
         self,
         spec: 'ConfigNode',
-        upgrader: Optional[Callable[[dict, str], None]] = None,
-        output_fname: Optional[str] = None,
-        tvars: Optional[dict] = None,
-        validator: Optional[Callable] = None,
-        options: Optional['Values'] = None
+        upgrader: Callable | None = None,
+        output_fname: str | None = None,
+        tvars: dict | None = None,
+        validator: Callable | None = None,
+        options: 'Values | None' = None
     ):
         """Instatiate a parsec config object.
 

cylc/flow/parsec/upgrade.py

@@ -43,17 +43,28 @@ class upgrader:
     SITE_CONFIG = 'site config'
     USER_CONFIG = 'user config'
 
-    DEPR_MSG = (
+    DEPR_TEMPLATE = (
         "Deprecated config items were automatically upgraded. "
-        "Please alter your workflow to use the new syntax."
+        "Please alter your {} to use the new syntax."
     )
+    depr_msg = DEPR_TEMPLATE.format('workflow')
 
-    def __init__(self, cfg, descr):
+    obslt_msg = (
+        "Obsolete config items were automatically deleted. "
+        "Please check your workflow and remove them permanently."
+    )
+
+    def __init__(self, cfg: dict, descr: str, broadcast: bool = False):
         """Store the config dict to be upgraded if necessary."""
         self.cfg = cfg
         self.descr = descr
         # upgrades must be ordered in case several act on the same item
-        self.upgrades = OrderedDict()
+        self.upgrades: OrderedDict[str, list[dict]] = OrderedDict()
+        if broadcast:
+            self.depr_msg = self.DEPR_TEMPLATE.format('broadcast')
+            self.obslt_msg = (
+                "Obsolete config items were rejected by the broadcast."
+            )
 
     def deprecate(
         self, vn, oldkeys, newkeys=None,
@@ -255,13 +266,9 @@ def upgrade(self):
                 # Log at warning level.
                 level = WARNING
             if obsoletions:
-                LOG.log(
-                    level,
-                    "Obsolete config items were automatically deleted."
-                    " Please check your workflow and remove them permanently."
-                )
+                LOG.log(level, self.obslt_msg)
             if deprecations:
-                LOG.log(level, self.DEPR_MSG)
+                LOG.log(level, self.depr_msg)
 
             for vn, msgs in warnings.items():
                 for msg in msgs:

cylc/flow/scripts/broadcast.py

@@ -196,13 +196,36 @@ def get_rdict(left, right=None, for_cancel_broadcast=False):
     upg(
         {'runtime': {'__MANY__': rdict}},
         'test',
-        for_cancel_broadcast=for_cancel_broadcast,
+        broadcast='cancel' if for_cancel_broadcast else True,
     )
+    rdict = strip_empty_sections(rdict)
     # Perform validation, but don't coerce the original (deepcopy).
     cylc_config_validate(deepcopy(rdict), SPEC['runtime']['__MANY__'])
     return rdict
 
 
+def strip_empty_sections(data):
+    """Recursively strip data structure of empty dicts.
+
+    This is needed post-upgrade as the upgrader can leave empty sections
+    behind.
+
+    >>> strip_empty_sections(
+    ...     {'job': {}, 'thing limit': 'PT2S', 'whatever': ''})
+    {'thing limit': 'PT2S', 'whatever': ''}
+
+    >>> strip_empty_sections({'section': {'subsection': {}}})
+    {}
+    """
+    if not isinstance(data, dict):
+        return data
+    return {
+        key: stripped
+        for key, val in data.items()
+        if (stripped := strip_empty_sections(val)) != {}
+    }
+
+
 def files_to_settings(settings, setting_files, cancel_mode=False):
     """Parse setting files, and append to settings."""
     cfg = ParsecConfig(
@@ -437,9 +460,11 @@ async def run(options: 'Values', workflow_id):
                 raise InputError(
                     "--cancel=[SEC]ITEM does not take a value")
             option_item = option_item.strip()
-            setting = get_rdict(option_item, for_cancel_broadcast=True)
-            settings.append(setting)
+            if setting := get_rdict(option_item, for_cancel_broadcast=True):
+                settings.append(setting)
         files_to_settings(settings, options.cancel_files, options.cancel)
+        if not settings:
+            raise InputError("No valid settings to broadcast.")
         mutation_kwargs['variables'].update(
             {
                 'bMode': 'Clear',
@@ -456,9 +481,11 @@ async def run(options: 'Values', workflow_id):
                 raise InputError(
                     "--set=[SEC]ITEM=VALUE requires a value")
             lhs, rhs = [s.strip() for s in option_item.split("=", 1)]
-            setting = get_rdict(lhs, rhs)
-            settings.append(setting)
+            if setting := get_rdict(lhs, rhs):
+                settings.append(setting)
         files_to_settings(settings, options.setting_files)
+        if not settings:
+            raise InputError("No valid settings to broadcast.")
         mutation_kwargs['variables'].update(
             {
                 'bMode': 'Set',
@@ -472,9 +499,9 @@ async def run(options: 'Values', workflow_id):
 
     results = await pclient.async_request('graphql', mutation_kwargs)
     try:
+        bad_options = {}
         for result in results['broadcast']['result']:
-            modified_settings = result['response'][0]
-            bad_options = result['response'][1]
+            modified_settings, bad_options = result['response']
             if modified_settings:
                 ret['stdout'].append(
                     get_broadcast_change_report(
@@ -483,7 +510,7 @@ async def run(options: 'Values', workflow_id):
                     )
                 )
         bad_result = report_bad_options(bad_options, is_set=report_set)
-    except TypeError:
+    except (TypeError, ValueError):
         # Catch internal API server errors
         bad_result = cparse(f'<red>{results}</red>')
 

tests/integration/scripts/test_broadcast.py

@@ -14,12 +14,19 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import logging
+
 from ansimarkup import strip as cstrip
+import pytest
 
 from cylc.flow.network.client import WorkflowRuntimeClient
 from cylc.flow.option_parsers import Options
 from cylc.flow.rundb import CylcWorkflowDAO
-from cylc.flow.scripts.broadcast import _main, get_option_parser
+from cylc.flow.scheduler import Scheduler
+from cylc.flow.scripts.broadcast import (
+    _main,
+    get_option_parser,
+)
 from cylc.flow.util import sstrip
 
 
@@ -314,3 +321,51 @@ def put_broadcast(*args, **kwargs):
 
         # the error should be logged server-side too
         assert log_filter(contains='FooBar')
+
+
+async def test_broadcast_auto_upgraded_settings(
+    one: Scheduler,
+    start,
+    log_filter,
+    caplog: pytest.LogCaptureFixture,
+    capsys: pytest.CaptureFixture,
+):
+    """Test broadcast of deprecated/obsolete settings."""
+    opts = {
+        'point_strings': ['1'],
+        'namespaces': ['root'],
+    }
+    async with start(one):
+        # Test deprecated setting that is moved to a different section by
+        # the auto-upgrader should still work.
+        rets = await _main(
+            BroadcastOptions(
+                settings=['[job]execution time limit=PT1H'], **opts
+            ),
+            one.workflow,
+        )
+        assert set(rets.values()) == {True}
+        assert log_filter(
+            logging.WARNING, regex=r"Deprecated config.* upgraded.* broadcast"
+        )
+
+        stdout, stderr = capsys.readouterr()
+        assert "Traceback" not in stdout + stderr
+        caplog.clear()
+
+        # Test obsolete setting that is removed by the auto-upgrader should
+        # be rejected.
+        rets = await _main(
+            BroadcastOptions(
+                settings=['extra log files=whatever'], **opts
+            ),
+            one.workflow,
+        )
+        assert set(rets.values()) == {False}
+        assert log_filter(
+            logging.WARNING,
+            regex=r"Obsolete config.* rejected by the broadcast",
+        )
+        stdout, stderr = capsys.readouterr()
+        assert "InputError: No valid settings to broadcast" in stderr
+        assert "Traceback" not in stdout + stderr