diff --git a/projects/vdk-control-cli/src/vdk/internal/control/job/job_config.py b/projects/vdk-control-cli/src/vdk/internal/control/job/job_config.py index 88660a282c..42c5874984 100644 --- a/projects/vdk-control-cli/src/vdk/internal/control/job/job_config.py +++ b/projects/vdk-control-cli/src/vdk/internal/control/job/job_config.py @@ -5,12 +5,14 @@ import fileinput import logging import os +import re import sys +from typing import Any +from typing import List from vdk.internal.control.exception.vdk_exception import VDKException from vdk.internal.control.utils.control_utils import read_config_ini_file - log = logging.getLogger(__name__) @@ -36,10 +38,10 @@ def __init__(self, data_job_path): config_parser=self._config_ini, configuration_file_path=self._config_file ) - def get_team(self): + def get_team(self) -> str: return self._get_value("owner", "team") - def set_team_if_exists(self, value): + def set_team_if_exists(self, value) -> None: """ If 'team' option exists in section 'owner' of config.ini, value param is assigned to it and config.ini file is overwritten with the new team value. @@ -47,35 +49,35 @@ def set_team_if_exists(self, value): """ return self._set_value("owner", "team", value) - def get_schedule_cron(self): + def get_schedule_cron(self) -> str: return self._get_value("job", "schedule_cron") - def get_enable_execution_notifications(self): + def get_enable_execution_notifications(self) -> bool: return self._get_boolean( "contacts", "enable_execution_notifications", fallback=True ) - def get_notification_delay_period_minutes(self): + def get_notification_delay_period_minutes(self) -> int: return self._get_positive_int( "contacts", "notification_delay_period_minutes", fallback=240 ) - def get_contacts_notified_on_job_failure_user_error(self): + def get_contacts_notified_on_job_failure_user_error(self) -> List[str]: return self._get_contacts("notified_on_job_failure_user_error") - def get_contacts_notified_on_job_failure_platform_error(self): + def get_contacts_notified_on_job_failure_platform_error(self) -> List[str]: return self._get_contacts("notified_on_job_failure_platform_error") - def get_contacts_notified_on_job_success(self): + def get_contacts_notified_on_job_success(self) -> List[str]: return self._get_contacts("notified_on_job_success") - def get_contacts_notified_on_job_deploy(self): + def get_contacts_notified_on_job_deploy(self) -> List[str]: return self._get_contacts("notified_on_job_deploy") - def _get_boolean(self, section, key, fallback=None): + def _get_boolean(self, section, key, fallback=None) -> bool: return self._config_ini.getboolean(section, key, fallback=fallback) - def _get_positive_int(self, section, key, fallback=None): + def _get_positive_int(self, section, key, fallback=None) -> int: try: value = self._config_ini.getint(section, key, fallback=fallback) if value <= 0: @@ -91,12 +93,12 @@ def _get_positive_int(self, section, key, fallback=None): f"a positive integer and redeploy the job.", ) - def _get_value(self, section, key): + def _get_value(self, section, key) -> Any: if self._config_ini.has_option(section, key): return self._config_ini.get(section, key) return "" - def _set_value(self, section, key, value): + def _set_value(self, section, key, value) -> bool: success = False if self._config_ini.has_option(section, key): for line in fileinput.input(self._config_file, inplace=1): @@ -106,8 +108,9 @@ def _set_value(self, section, key, value): sys.stdout.write(line) return success - def _get_contacts(self, key): + def _get_contacts(self, key) -> List[str]: contacts_str = self._get_value("contacts", key).strip() + contacts = [] if contacts_str: - return [x.strip() for x in contacts_str.split(";")] - return [] + contacts = [x.strip() for x in re.split("[;,]", contacts_str)] + return contacts diff --git a/projects/vdk-control-cli/tests/vdk/internal/control/command_groups/job/test_job_config.py b/projects/vdk-control-cli/tests/vdk/internal/control/command_groups/job/test_datajob_config.py similarity index 70% rename from projects/vdk-control-cli/tests/vdk/internal/control/command_groups/job/test_job_config.py rename to projects/vdk-control-cli/tests/vdk/internal/control/command_groups/job/test_datajob_config.py index 0794eab037..3b470699b5 100644 --- a/projects/vdk-control-cli/tests/vdk/internal/control/command_groups/job/test_job_config.py +++ b/projects/vdk-control-cli/tests/vdk/internal/control/command_groups/job/test_datajob_config.py @@ -86,6 +86,51 @@ def test_negative_notification_delay_period_minutes(self): with pytest.raises(VDKException): job_config.get_notification_delay_period_minutes() + def test_parse_notification_contacts(self): + self.__create_config_ini( + self.tmp_copy_job_test_config_ini_path, + [ + ("notified_on_job_deploy", "a@abv.bg; b@dir.bg, c@abv.bg ; d@dir.bg"), + ( + "notified_on_job_success", + "a@abv.bg; b@dir.bg, c@abv.bg ; d@dir.bg", + ), + ( + "notified_on_job_failure_user_error", + "a@abv.bg; b@dir.bg, c@abv.bg ; d@dir.bg", + ), + ( + "notified_on_job_failure_platform_error", + "a@abv.bg; b@dir.bg, c@abv.bg ; d@dir.bg", + ), + ], + ) + job_config = JobConfig(self.tmp_copy_job_test_path) + assert job_config.get_contacts_notified_on_job_deploy() == [ + "a@abv.bg", + "b@dir.bg", + "c@abv.bg", + "d@dir.bg", + ] + assert job_config.get_contacts_notified_on_job_success() == [ + "a@abv.bg", + "b@dir.bg", + "c@abv.bg", + "d@dir.bg", + ] + assert job_config.get_contacts_notified_on_job_failure_user_error() == [ + "a@abv.bg", + "b@dir.bg", + "c@abv.bg", + "d@dir.bg", + ] + assert job_config.get_contacts_notified_on_job_failure_platform_error() == [ + "a@abv.bg", + "b@dir.bg", + "c@abv.bg", + "d@dir.bg", + ] + @staticmethod def __create_config_ini(file, contacts): config = configparser.ConfigParser()