feat: add new env parsing and helper module

src/gha_runner/clouddeployment.py

@@ -1,6 +1,6 @@
 from abc import ABC, abstractmethod
 from gha_runner.gh import GitHubInstance, MissingRunnerLabel
-from gha_runner.helper import warning, error
+from gha_runner.helper.workflow_cmds import warning, error
 from dataclasses import dataclass, field
 from typing import Type
 

src/gha_runner/helper/input.py

@@ -0,0 +1,172 @@
+import json
+from typing import Dict, Any, NamedTuple, Optional, Type
+from copy import deepcopy
+
+
+def check_required(env: Dict[str, str], required: list[str]):
+    """Check if required environment variables are set.
+
+    Parameters
+    ----------
+    env : Dict[str, str]
+        The environment variables.
+    required : list[str]
+        A list of required environment variables.
+
+    Raises
+    ------
+    ValueError
+        If any required environment variables are missing.
+
+    """
+    missing = [var for var in required if not env.get(var)]
+    if missing:
+        raise ValueError(f"Missing required environment variables: {missing}")
+
+
+class ParamConfig(NamedTuple):
+    """Configuration for a single parameter."""
+
+    env_var: str
+    key: str
+    is_json: bool
+    allow_empty: bool
+    type_val: Type
+
+
+class EnvVarBuilder:
+    """A builder class for constructing a dictionary of parameters from environment variables.
+
+    This class provides a fluent interface for parsing and transforming environment variables
+    into a structured dictionary, with support for JSON parsing and type conversion.
+
+    Examples
+    --------
+    >>> env = {"MY_VAR": "123", "JSON_VAR": '{"key": "value"}'}
+    >>> builder = EnvVarBuilder(env)
+    >>> result = (builder
+    ...     .with_var("MY_VAR", "my_key", type_hint=int)
+    ...     .with_var("JSON_VAR", "json_key", is_json=True)
+    ...     .build())
+    >>> # result = {"my_key": 123, "json_key": {"key": "value"}}
+
+    Methods
+    -------
+    with_var(var_name: str, key: str, is_json: bool = False,
+            allow_empty: bool = False, type_hint: Type = str) -> EnvVarBuilder:
+        Adds an environment variable to be parsed with specified configuration.
+    build() -> dict:
+        Builds and returns the final dictionary of parsed parameters.
+
+    Notes
+    -----
+    - The builder creates deep copies of values to prevent mutation
+    - JSON parsing is performed before type conversion
+    - Empty strings are ignored by default unless allow_empty is True
+
+    """
+
+    def __init__(self, env: Dict[str, str]):
+        """Initialize the builder.
+
+        The builder is initialized with the environment variables as a dict.
+
+        Parameters
+        ----------
+        env : Dict[str, str]
+            The environment variables.
+
+        """
+        self.env = env
+        self.params = {}
+
+    def parse_value(
+        self, value: str, is_json: bool, type_hint: Optional[Type]
+    ) -> Any:
+        """Parse the value based on the configuration.
+
+        Parameters
+        ----------
+        value : str
+            The value to parse
+        is_json : bool, optional
+            If True, parse the value as JSON, by default False
+        allow_empty : bool, optional
+            If True, include empty strings in the result, by default False
+        type_hint : Type, optional
+            Type to convert the value to (if not JSON), by default str
+
+        Returns
+        -------
+        EnvVarBuilder
+            Returns self for method chaining
+
+        Raises
+        ------
+        ValueError
+            If the environment variable cannot be parsed according to specifications
+
+        Notes
+        -----
+        - JSON parsing is performed before type conversion if is_json is True
+
+        """
+        if is_json:
+            return json.loads(value)
+        if type_hint is not None:
+            return type_hint(value)
+
+    def parse_single_param(self, config: ParamConfig):
+        """Parse a single parameter from the environment variables."""
+        value = self.env.get(config.env_var)
+        if value is not None and (config.allow_empty or value.strip()):
+            parsed_value = self.parse_value(
+                value, config.is_json, config.type_val
+            )
+            self._update_params(config.key, parsed_value)
+
+    def _update_params(self, key: str, value: Any):
+        self.params = deepcopy(self.params)
+        self.params[key] = deepcopy(value)
+
+    def with_var(
+        self,
+        var_name: str,
+        key: str,
+        is_json: bool = False,
+        allow_empty: bool = False,
+        type_hint: Type = str,
+    ) -> "EnvVarBuilder":
+        """Build and return the final dictionary of parsed parameters.
+
+        Returns
+        -------
+        dict
+            Dictionary containing all parsed and transformed environment variables
+
+        Raises
+        ------
+        ValueError
+            If any configured environment variable parsing fails
+
+        Notes
+        -----
+        - Empty strings are ignored by default unless allow_empty is True
+        - JSON parsing is performed before type conversion if is_json is True
+
+        """
+        config = ParamConfig(var_name, key, is_json, allow_empty, type_hint)
+
+        self.parse_single_param(config)
+        return self
+
+    def build(self) -> dict:
+        """Build and return the final dictionary of parsed parameters.
+
+        Returns
+        -------
+        dict
+            Dictionary containing all parsed and transformed environment variables.
+
+        """
+        return self.params

src/gha_runner/helper/workflow_cmds.py

@@ -0,0 +1,14 @@
+import os
+
+
+def output(name: str, value: str):
+    with open(os.environ["GITHUB_OUTPUT"], "a") as output:
+        output.write(f"{name}={value}\n")
+
+
+def warning(title: str, message):
+    print(f"::warning title={title}::{message}")
+
+
+def error(title: str, message):
+    print(f"::error title={title}::{message}")

tests/test_helper.py

@@ -0,0 +1,44 @@
+import pytest
+from unittest.mock import mock_open, patch
+from gha_runner.helper.workflow_cmds import output, warning, error
+
+
+def test_output(monkeypatch):
+    # Setup mock environment variable
+    monkeypatch.setenv("GITHUB_OUTPUT", "mock_output_file")
+
+    mock_file = mock_open()
+    with patch("builtins.open", mock_file):
+        output("test_name", "test_value")
+
+    mock_file().write.assert_called_once_with("test_name=test_value\n")
+
+
+def test_warning(capsys):
+    warning("Test Title", "Test Message")
+    captured = capsys.readouterr()
+    assert captured.out == "::warning title=Test Title::Test Message\n"
+
+
+def test_warning_with_special_chars(capsys):
+    warning("Test:Title", "Test,Message")
+    captured = capsys.readouterr()
+    assert captured.out == "::warning title=Test:Title::Test,Message\n"
+
+
+def test_error(capsys):
+    error("Test Title", "Test Message")
+    captured = capsys.readouterr()
+    assert captured.out == "::error title=Test Title::Test Message\n"
+
+
+def test_error_with_special_chars(capsys):
+    error("Test:Title", "Test,Message")
+    captured = capsys.readouterr()
+    assert captured.out == "::error title=Test:Title::Test,Message\n"
+
+
+def test_output_missing_env_var(monkeypatch):
+    monkeypatch.delenv("GITHUB_OUTPUT", raising=False)
+    with pytest.raises(KeyError):
+        output("test_name", "test_value")

tests/test_input.py

@@ -0,0 +1,88 @@
+import re
+from gha_runner.helper.input import (
+    EnvVarBuilder,
+    check_required,
+)
+import pytest
+
+
+def test_parse_params_empty():
+    env = {}
+    env["INPUT_AWS_IMAGE_ID"] = ""
+    env["INPUT_AWS_INSTANCE_TYPE"] = ""
+    env["INPUT_AWS_SUBNET_ID"] = ""
+    env["INPUT_AWS_SECURITY_GROUP_ID"] = ""
+    env["INPUT_AWS_IAM_ROLE"] = ""
+    env["INPUT_AWS_TAGS"] = ""
+    env["INPUT_AWS_REGION_NAME"] = ""
+    env["INPUT_AWS_HOME_DIR"] = ""
+    env["INPUT_AWS_LABELS"] = ""
+    params = (
+        EnvVarBuilder(env)
+        .with_var("INPUT_AWS_IMAGE_ID", "image_id", allow_empty=True)
+        .with_var("INPUT_AWS_INSTANCE_TYPE", "instance_type", allow_empty=True)
+        .with_var("INPUT_AWS_SUBNET_ID", "subnet_id")
+        .with_var("INPUT_AWS_SECURITY_GROUP_ID", "security_group_id")
+        .with_var("INPUT_AWS_IAM_ROLE", "iam_role")
+        .with_var("INPUT_AWS_TAGS", "tags", is_json=True)
+        .with_var("INPUT_AWS_REGION_NAME", "region_name", allow_empty=True)
+        .with_var("INPUT_AWS_HOME_DIR", "home_dir", allow_empty=True)
+        .with_var("INPUT_AWS_LABELS", "labels")
+        .build()
+    )
+
+    assert params == {
+        "image_id": "",
+        "instance_type": "",
+        "home_dir": "",
+        "region_name": "",
+    }
+
+
+def test_env_builder():
+    env = {}
+    env["INPUT_AWS_IMAGE_ID"] = "ami-1234567890"
+    env["INPUT_AWS_INSTANCE_TYPE"] = "t2.micro"
+    env["INPUT_GH_REPO"] = "owner/test"
+    env["GITHUB_REPOSITORY"] = "owner/test_other"
+    env["INPUT_INSTANCE_COUNT"] = "1"
+    env["INPUT_AWS_TAGS"] = '{"Key": "Name", "Value": "test"}'
+    config = (
+        EnvVarBuilder(env)
+        .with_var("INPUT_AWS_IMAGE_ID", "image_id")
+        .with_var("INPUT_AWS_INSTANCE_TYPE", "instance_type")
+        .with_var("GITHUB_REPOSITORY", "repo")
+        .with_var("INPUT_GH_REPO", "repo")
+        .with_var("INPUT_INSTANCE_COUNT", "instance_count", type_hint=int)
+        .with_var("INPUT_AWS_TAGS", "tags", is_json=True)
+        .build()
+    )
+    assert config["image_id"] == "ami-1234567890"
+    assert config["instance_type"] == "t2.micro"
+    assert config["repo"] == "owner/test"
+    assert config["instance_count"] == 1
+    assert isinstance(config["instance_count"], int)
+    assert config["tags"] == {"Key": "Name", "Value": "test"}
+    assert isinstance(config["tags"], dict)
+
+
+def test_check_required():
+    env = {"GH_TOKEN": "123", "AWS_ACCESS_KEY": "123", "AWS_SECRET_KEY": "123"}
+    required = ["GH_TOKEN", "AWS_ACCESS_KEY", "AWS_SECRET_KEY"]
+    check_required(env, required)
+    env = {"GH_TOKEN": "123"}
+    with pytest.raises(
+        Exception,
+        match=re.escape(
+            "Missing required environment variables: ['AWS_ACCESS_KEY', 'AWS_SECRET_KEY']"
+        ),
+    ):
+        check_required(env, required)
+    env = {}
+    with pytest.raises(
+        Exception,
+        match=re.escape(
+            "Missing required environment variables: ['GH_TOKEN', 'AWS_ACCESS_KEY', 'AWS_SECRET_KEY']"
+        ),
+    ):
+        check_required(env, required)