diff --git a/cli/cli/dynamic_configuration/__init__.py b/cli/cli/dynamic_configuration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/cli/cli/dynamic_configuration/predefined_funcs.py b/cli/cli/dynamic_configuration/predefined_funcs.py new file mode 100644 index 0000000..0bf05e0 --- /dev/null +++ b/cli/cli/dynamic_configuration/predefined_funcs.py @@ -0,0 +1,82 @@ +import re +from typing import Any, Tuple + +from cli.utils import run_script + + +def regex(pattern, text) -> bool: + """ + Check if the provided text matches the given regex pattern. + + :param pattern: The regex pattern to match against. + :param text: The text to be matched. + :return: True if the text matches the pattern, False otherwise. + """ + return bool(re.match(pattern, text)) + + +def lowercase(val) -> str: + """ + Convert the given string to lowercase. + + :param val: The string to be converted. + :return: The lowercase version of the input string. + """ + return val.lower() + + +def kubectlget(deployment_dir, *args) -> Tuple[bool, str]: + """ + Placeholder function to execute a kubectl get command. + + :param deployments_dir: The deployments_dir for a specific deployment that holds. + :param args: The arguments for the kubectl get command. + :return: The output of the kubectl get command. + """ + length = len(args) + args_with_spaces = [f"{arg}{"" if i == length - 1 else " "}" for i, arg in enumerate(args)] + params = concatenate(*args_with_spaces) + scripts_dir, _, _ = deployment_dir.rsplit("/", 2) + result = run_script(scripts_dir, deployment_dir, "kubectlget.sh", capture_output=True, **{"PARAMS": params}) + return result.exitcode == 0, result.output + + +def returnasis(val) -> Any: + """ + Return the input value as is. + + :param val: The value to be returned. + :return: The input value, unchanged. + """ + return val + + +def concatenate(*args) -> str: + """ + Concatenate the given arguments into a single string. + + :param args: The arguments to be concatenated. + :return: A single string formed by concatenating the input arguments. + """ + return "".join([str(arg) for arg in args]) + + +def divide(dividend, divisor): + """ + Divide the dividend by the divisor and return the result. + + :param dividend: The value to be divided. + :param divisor: The value to divide by. + :return: The result of the division. + :raises ValueError: If the dividend or divisor cannot be converted to an integer or if the divisor is zero. + """ + if divisor == 0: + raise ValueError("The divisor cannot be zero.") + + try: + dividend = int(dividend) + divisor = int(divisor) + except Exception as exc: + raise ValueError("Conversion error") from exc + + return dividend / divisor diff --git a/cli/cli/scripts/kubectlget.sh b/cli/cli/scripts/kubectlget.sh new file mode 100755 index 0000000..d705c4e --- /dev/null +++ b/cli/cli/scripts/kubectlget.sh @@ -0,0 +1,34 @@ +#!/bin/bash + +if [ -z "$DEPLOYMENT_DIR" ]; then + of echo "Error: DEPLOYMENT_DIR is not set" + exit 1 +fi + +if [ -z "$PARAMS" ]; then + echo "Error: PARAMS is not set" + exit 1 +fi + +source $DEPLOYMENT_DIR/.env --source-only + +if [ -z "$KUBECONFIG" ]; then + echo "Error: KUBECONFIG is not set" + exit 1 +fi + +output=$(kubectl --kubeconfig "$KUBECONFIG" $PARAMS 2>&1) + +if [ $? -eq 0 ]; then + output="${output%\"}" + output="${output#\"}" + echo "$output" +else + if [[ $output == "Error from server (NotFound):"*"not found" ]]; then + exit 0 + else + # If the error message does not match, print the error message and return exit code 1 + >&2 echo "$output" + exit 1 + fi +fi diff --git a/cli/tests/test_dynamic_configuration_predefined_funcs.py b/cli/tests/test_dynamic_configuration_predefined_funcs.py new file mode 100644 index 0000000..38b3f09 --- /dev/null +++ b/cli/tests/test_dynamic_configuration_predefined_funcs.py @@ -0,0 +1,127 @@ +from unittest import TestCase, mock + +from cli.dynamic_configuration.predefined_funcs import concatenate, divide, kubectlget, lowercase, regex, returnasis +from cli.utils import SubprocessResult + + +class TestRegex(TestCase): + def setUp(self): + pass + + def tearDown(self): + pass + + def validate(self, pattern, valid_inputs, invalid_inputs): + for valid_input in valid_inputs: + is_valid = regex(pattern, valid_input) + self.assertTrue(is_valid, f"Valid input '{valid_input}' failed the regex test") + + for invalid_input in invalid_inputs: + is_valid = regex(pattern, invalid_input) + self.assertFalse(is_valid, f"Invalid input '{invalid_input}' passed the regex test") + + def test_only_y_and_n(self): + pattern = "^[yYnN]$" + valid_inputs = ["y", "Y", "n", "N"] + invalid_inputs = ["1", "yes", "no"] + + self.validate(pattern, valid_inputs, invalid_inputs) + + def test_any_input(self): + pattern = ".+" + valid_inputs = ["any", "random", "1000", "storage-class"] + + self.validate(pattern, valid_inputs, []) + + def test_schema(self): + pattern = "^[Hh][Tt][Tt][Pp]([Ss])?$" + valid_inputs = ["http", "HTTP", "https", "HTTPS"] + invalid_inputs = ["1", "yes", "no", "cancel", "htt", "HTPS"] + + self.validate(pattern, valid_inputs, invalid_inputs) + + def test_divisible_by_1000(self): + pattern = "^[1-9][0-9]*000$" + valid_inputs = ["1000", "2000", "4000", "10000"] + invalid_inputs = ["750", "1250", "9999", "foobar"] + + self.validate(pattern, valid_inputs, invalid_inputs) + + def test_divisible_by_4(self): + pattern = "^.*([048]|([02468][048])|([13579][26]))$" + valid_inputs = ["4", "8", "32", "128"] + invalid_inputs = ["2", "6", "9", "1001"] + + self.validate(pattern, valid_inputs, invalid_inputs) + + def test_email(self): + pattern = "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$" + valid_inputs = ["foo@bar.com"] + invalid_inputs = ["any", "1000", "foobar.com"] + + self.validate(pattern, valid_inputs, invalid_inputs) + + def validate_password(self): + pattern = "^.{8,}$" + valid_inputs = ["12345678"] + invalid_inputs = ["1234567"] + + self.validate(pattern, valid_inputs, invalid_inputs) + + +def test_lowercase(): + inp = "CamelCase" + out = "camelcase" + + assert lowercase(inp) == out + + +def test_returnasis(): + inp = "foobar" + out = "foobar" + + assert returnasis(inp) == out + + +def test_concatenate(): + args = ["this", "is", 1, "foo", "bar"] + out = "thisis1foobar" + + assert concatenate(*args) == out + + +class TestDivide(TestCase): + def test_divide(self): + self.assertTrue(divide(4, 2) == 2) + self.assertTrue(divide(100, 25) == 4) + + def test_divide_conversion_error(self): + with self.assertRaises(ValueError) as context: + divide("string", 2) + self.assertTrue("Conversion error" in str(context.exception)) + + def test_divide_by_zero(self): + with self.assertRaises(ValueError) as context: + divide(10, 0) + self.assertTrue("The divisor cannot be zero." in str(context.exception)) + + +def test_kubectlget(): + with ( + mock.patch("cli.dynamic_configuration.predefined_funcs.run_script") as mock_run_script, + ): + mock_run_script.return_value = SubprocessResult(succeeded=True, output="local-path", exitcode=0) + deployment_dir = "foo/bar/scripts/deployments/a-deployment-id" + args = ["get", "pv", "-l", "pv-label-key=mylabel", "-o", 'jsonpath="{.items[*].spec.storageClassName}"'] + is_success, value = kubectlget(deployment_dir, *args) + + assert is_success is True + assert value == "local-path" + + mock_run_script.assert_called_with( + "foo/bar/scripts", + "foo/bar/scripts/deployments/a-deployment-id", + "kubectlget.sh", + capture_output=True, + **{"PARAMS": 'get pv -l pv-label-key=mylabel -o jsonpath="{.items[*].spec.storageClassName}"'}, + )