pathogens: use top level workflows key

List workflows under a top level `workflows` key in the `nextstrain-pathogen.yaml` 
file where each workflow has it's own `compatibility` key:

```yaml
workflows:
  ingest:
    compatibility:
      nextstrain run: True
  phylogenetic:
    compatibility:
      nextstrain run: True
```

The top level `compatibility['nextstrain run']` boolean is still supported 
for backwards compatibility.

From discussion with @victorlin in
<#462 (comment)>

Author: joverlee521

CHANGES.md

@@ -18,7 +18,7 @@ development source code and as such may not be routinely kept up to date.
 * `nextstrain setup <pathogen>` and `nextstrain version --pathogens` now list
   the available workflows for a pathogen if the pathogen lists the workflows
   in the top-level `nextstrain-pathogen.yaml` file.
-  ([#461](https://github.com/nextstrain/cli/pull/461))
+  ([#461](https://github.com/nextstrain/cli/pull/461), [#472](https://github.com/nextstrain/cli/pull/472))
 
 * Snakemake's storage support downloaded files (stored in `.snakemake/storage/`)
   are now downloaded from AWS Batch builds by default.

doc/changes.md

@@ -22,7 +22,7 @@ development source code and as such may not be routinely kept up to date.
 * `nextstrain setup <pathogen>` and `nextstrain version --pathogens` now list
   the available workflows for a pathogen if the pathogen lists the workflows
   in the top-level `nextstrain-pathogen.yaml` file.
-  ([#461](https://github.com/nextstrain/cli/pull/461))
+  ([#461](https://github.com/nextstrain/cli/pull/461), [#472](https://github.com/nextstrain/cli/pull/472))
 
 * Snakemake's storage support downloaded files (stored in `.snakemake/storage/`)
   are now downloaded from AWS Batch builds by default.

nextstrain/cli/command/run.py

@@ -228,7 +228,7 @@ def run(opts):
     # Resolve pathogen and workflow names to a local workflow directory.
     pathogen = PathogenVersion(opts.pathogen)
 
-    if opts.workflow not in pathogen.registered_workflows():
+    if opts.workflow not in pathogen.compatible_workflows("nextstrain run"):
         print(f"The {opts.workflow!r} workflow is not registered as a compatible workflow, but trying to run anyways.")
 
     workflow_directory = pathogen.workflow_path(opts.workflow)

nextstrain/cli/command/version.py

@@ -72,9 +72,9 @@ def run(opts):
                     if opts.verbose:
                         print("      " + str(version.path))
 
-                    if registered_workflows := version.registered_workflows():
-                        print("      " + "Available workflows:")
-                        for workflow in registered_workflows:
+                    if compatible_workflows := version.compatible_workflows("nextstrain run"):
+                        print("      " + "`nextstrain run` compatible workflows:")
+                        for workflow in compatible_workflows:
                             print("        " + workflow)
                     else:
                         print("      " + "No workflows listed, please refer to pathogen docs.")

nextstrain/cli/pathogens.py

@@ -308,20 +308,32 @@ def __init__(self, name_version_url: str, new_setup: bool = False):
     def registered_workflows(self) -> Dict[str, Dict]:
         """
         Parses :attr:`.registration` to return a dict of registered
-        compatible workflows, where the keys are workflow names.
+        workflows, where the keys are workflow names.
         """
         if self.registration is None:
             debug("pathogen does not have a registration")
             return {}
 
-        workflows = self.registration.get("compatibility", {}).get("nextstrain run")
+        workflows = self.registration.get("workflows")
         if not isinstance(workflows, dict):
-            debug(f"pathogen registration.compatibility['nextstrain runs'] is not a dict (got a {type(workflows).__name__})")
+            debug(f"pathogen registration.workflows is not a dict (got a {type(workflows).__name__})")
             return {}
 
         return workflows
 
 
+    def compatible_workflows(self, feature: str) -> Dict[str, Dict]:
+        """
+        Parses registered workflows to return a subset of workflows that are
+        compatible with the provided *feature*.
+        """
+        return {
+            workflow: workflow_config
+            for workflow, workflow_config in self.registered_workflows().items()
+            if workflow_config.get("compatibility", {}).get(feature, False)
+        }
+
+
     def workflow_path(self, workflow: str) -> Path:
         return self.path / workflow
 
@@ -481,20 +493,19 @@ def test_compatibility() -> SetupTestResult:
             if self.registration is None:
                 return msg + "\n(couldn't read registration)", False
 
+            if compatible_workflows := self.compatible_workflows("nextstrain run"):
+                return msg + f"\nCompatible workflows: {list(compatible_workflows.keys())}", True
+
+            # If no compatible workflows are listed, then check for the top level
+            # boolean compatibility declaration
             try:
                 compatibility = self.registration["compatibility"]["nextstrain run"]
             except (KeyError, IndexError, TypeError):
                 if DEBUGGING:
                     traceback.print_exc()
                 return msg + "\n(couldn't find 'compatibility: nextstrain run: …' field)", False
 
-            if compatibility:
-                if workflows := self.registered_workflows():
-                    msg += f"\nAvailable workflows: {list(workflows.keys())}"
-                else:
-                    msg += f"\nNo workflows listed, please refer to pathogen docs."
-
-            return msg, bool(compatibility)
+            return msg + "\nNo compatible workflows listed, please refer to pathogen docs.", bool(compatibility)
 
         return [
             ('downloaded',