feat(cli/configs.py): adding validation for deployments (#67)

* feat(cli/configs.py): adding validation for deployments

Now `workflow configs deploy <filepath>` will check deployments usage and notify the user

* Trigger

* fix(deployments): check for unused deployments in configs

* refactor(validate): deployments

---------

Co-authored-by: Shiny [ˈʃaɪni] <[email protected]>

tests/conftest.py

@@ -12,3 +12,64 @@ def set_testing_workspace():
     runner = CliRunner()
     runner.invoke(workflow, ["workspace", "set", "development"])
     return True
+
+
+@pytest.fixture(autouse=True, scope="function")
+def config_with_deployments():
+    """Return config with deployments for testing."""
+    return {
+        "version": "2",
+        "name": "example_deployments",
+        "defaults": {"user": "test", "site": "local"},
+        "deployments": [
+            {
+                "name": "ld1",
+                "site": "local",
+                "image": "workflow_local_112312",
+                "sleep": 1,
+                "resources": {"cores": 2, "ram": "1G"},
+                "replicas": 2,
+            },
+            {
+                "name": "ld2",
+                "site": "local",
+                "sleep": 1,
+                "image": "workflow_local_123123",
+                "resources": {"cores": 4, "ram": "2G"},
+                "replicas": 1,
+            },
+            {
+                "name": "ld3",
+                "site": "local",
+                "sleep": 1,
+                "image": "workflow_local",
+                "resources": {"cores": 4, "ram": "2G"},
+                "replicas": 1,
+            },
+            {
+                "name": "ld4",
+                "site": "local",
+                "sleep": 1,
+                "image": "workflow_local",
+                "resources": {"cores": 4, "ram": "2G"},
+                "replicas": 1,
+            },
+        ],
+        "pipeline": {
+            "steps": [
+                {"name": "echo", "stage": 1, "work": {"command": ["ls", "-lah"]}},
+                {"name": "uname", "stage": 2, "work": {"command": ["uname", "-a"]}},
+                {
+                    "name": "printenv",
+                    "runs_on": "ld3",
+                    "stage": 3,
+                    "work": {"command": ["printenv"]},
+                },
+                {
+                    "name": "printenv-2",
+                    "stage": 4,
+                    "work": {"command": ["printenv", "--version"]},
+                },
+            ]
+        },
+    }

tests/test_validate.py → tests/test_config_validation.py

@@ -1,8 +1,9 @@
 from os import chmod
+from typing import Any, Dict
 
 import pytest
 
-from workflow.utils.validate import command, function
+from workflow.utils.validate import command, deployments, function
 
 
 def test_validate_function():
@@ -23,3 +24,13 @@ def test_validate_command():
     # Test invalid command
     result = command("invalid_command")
     assert result is False
+
+
+def test_validate_deployments(config_with_deployments: Dict[str, Any]):
+    """Tests the validate_deployment function."""
+    unused, orphaned = deployments(config=config_with_deployments)
+    assert unused
+    assert orphaned
+    assert unused == ["ld1", "ld2", "ld4"]
+    orphaned.sort()
+    assert orphaned == ["echo", "printenv-2", "uname"]

workflow/cli/configs.py

@@ -1,7 +1,7 @@
 """Manage workflow pipelines."""
 
 import json
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
 import click
 import requests
@@ -15,6 +15,7 @@
 from yaml.loader import SafeLoader
 
 from workflow.http.context import HTTPContext
+from workflow.utils import validate
 from workflow.utils.renderers import render_config
 
 pretty.install()
@@ -80,6 +81,30 @@ def deploy(filename: click.Path):
     data: Dict[str, Any] = {}
     with open(filepath) as reader:
         data = yaml.load(reader, Loader=SafeLoader)  # type: ignore
+
+    # ? Check unused deployments and orphaned steps
+    unused_deployments: List[str] = list()
+    orphaned_steps: List[str] = list()
+    if data.get("deployments", None):
+        unused_deployments, orphaned_steps = validate.deployments(config=data)
+        if any(unused_deployments):
+            answer = console.input(
+                f"The following deployments are not being used: {unused_deployments},"
+                " do you wish to continue? (Y/n):"
+            )
+            if answer.lower() != "y":
+                console.print("Cancelling", style="red")
+                return
+        if any(orphaned_steps):
+            answer = console.input(
+                f"The following steps {orphaned_steps} does not have a runs_on "
+                "even though you have defined deployments, "
+                "do you wish to continue? (Y/n):",
+            )
+            if answer.lower() != "y":
+                console.print("Cancelling", style="red")
+                return
+
     try:
         deploy_result = http.configs.deploy(data)
     except requests.HTTPError as deploy_error:

workflow/utils/validate.py

@@ -94,6 +94,47 @@ def command(command: str) -> bool:
         return False
 
 
+def deployments(config: Dict[str, Any]) -> Tuple[List[str], List[str]]:
+    """Validate deployments.
+
+    Args:
+        config (Dict[str, Any]): Config payload.
+
+    Returns:
+        Tuple[List[str], List[str]]: Unused deployments and orphaned steps.
+    """
+    unused_deployments = []
+    orphaned_steps = []
+    deployments = config.get("deployments", [])
+    steps = config["pipeline"]["steps"]
+
+    for deployment in deployments:
+        n_used = int()
+        top_level = False
+        # ? First case: unused deployments
+        if config["pipeline"].get("runs_on", None):
+            if deployment["name"] in config["pipeline"]["runs_on"]:
+                n_used += 1
+                top_level = True
+        for step in steps:
+            has_runs_on = False
+            if step.get("runs_on", None):
+                used = deployment["name"] in step["runs_on"]
+                has_runs_on = True
+                if used:
+                    n_used += 1
+                # ? Second case: orphaned steps
+                if not used and not has_runs_on:
+                    orphaned_steps.append(step["name"])
+            if not top_level and not has_runs_on:
+                orphaned_steps.append(step["name"])
+
+        if n_used == 0:
+            unused_deployments.append(deployment["name"])
+
+    return (unused_deployments, list(set(orphaned_steps)))
+
+
 def outcome(output: Any) -> Tuple[Dict[str, Any], List[str], List[str]]:
     """Parse the output, returning results, products, and plots.