Source code for ska_oso_scripting.functions.environment

"""
The ska_oso_scripting.functions.environment module contains code used to identify the
execution environment and to verify commands are appropriate to that environment.
"""
import json
import os


class ExecutionEnvironmentError(ValueError):
    """
    Error raised when execution environment does not match the targeted
    telescope.
    """

    def __init__(
        self,
        expected_env,
        msg="Telescope environment required by script does not match deployment",
    ):
        super().__init__(msg)
        self.msg = msg
        self.current_env = os.getenv("SKA_TELESCOPE")
        self.expected_env = expected_env

    def __str__(self):
        return f"{self.msg} (expected {self.expected_env}, got" f" {self.current_env})"


def check_environment_for_consistency(request_json: str) -> None:
    """
    Confirm that the request is correct for the current environment, raising an
    ExecutionEnvironmentError if there is a mismatch.

    :param request_json: The JSON control string to be tested
    """
    request = json.loads(request_json)

    # delta configurations cannot be validated for environment as omit identifying material
    try:
        partial_configuration = request["tmc"]["partial_configuration"]
    except KeyError:
        partial_configuration = False
    if partial_configuration:
        return

    if request.get("mccs") is None and is_ska_low_environment():
        raise ExecutionEnvironmentError(
            expected_env="SKA-mid",
            msg="CDM object content does not match the expected telescope",
        )

    if request.get("dish") is None and is_ska_mid_environment():
        raise ExecutionEnvironmentError(
            expected_env="SKA-low",
            msg="CDM object content does not match the expected telescope",
        )


def is_ska_low_environment() -> bool:
    """
    Return True if execution environment is identified as SKA LOW.

    :return: True if SKA LOW
    """
    # value of environment variable will always be a string so we don't need
    # to cast to str
    env_var = os.getenv("SKA_TELESCOPE", "SKA-mid")
    return "low" in env_var.lower()


def is_ska_mid_environment() -> bool:
    """
    Return True if execution environment is identified as SKA MID

    :return: True if SKA MID
    """
    # used to check ska-telescope type i.e. 'mid' or 'low' defined in the environment
    # variable. Default value set to 'mid'
    return not is_ska_low_environment()