diff --git a/src/quads/cli/cli.py b/src/quads/cli/cli.py index 4790882f..9c8d100e 100644 --- a/src/quads/cli/cli.py +++ b/src/quads/cli/cli.py @@ -1914,3 +1914,93 @@ def action_validate_env(self): validate_env(_args, _loop, self.logger) self.logger.info("Quads assignments validation executed.") + + def action_list_notifications(self): + cloud_name = self.cli_args.get("cloud") + try: + if not cloud_name: + assignments = self.quads.get_active_assignments() + else: + assignment = self.quads.get_active_cloud_assignment(cloud_name=cloud_name) + assignments = [] if not assignment else [assignment] + except (APIServerException, APIBadRequest) as ex: + raise CliException(str(ex)) + headers = [ + "fail", + "success", + "initial", + "pre_initial", + "pre", + "one_day", + "three_days", + "five_days", + "seven_days", + ] + rows = [] + if assignments: + for ass in assignments: + cloud_name = str(ass.cloud.name) + ticket = str(ass.ticket) + notification = ass.notification + row = [cloud_name, ticket] + for header in headers: + row.append(str(getattr(notification, header))) + rows.append(row) + headers.insert(0, "cloud") + headers.insert(1, "ticket") + + col_widths = [max(len(str(item)) for item in col) for col in zip(headers, *rows)] + + def format_row(data_row): + return " ".join(str(item).ljust(width) for item, width in zip(data_row, col_widths)) + + table = [format_row(headers), "=" * (sum(col_widths) + 2 * (len(headers) - 1))] + table.extend(format_row(row) for row in rows) + self.logger.info("\n".join(table)) + else: + message = "WARNING: there are no current or future schedules" + if not cloud_name: + self.logger.warning(f"{message}") + else: + self.logger.warning(f"{message} for {cloud_name}") + + def action_modify_notification(self): + cloud_name = self.cli_args.get("cloud") + if not cloud_name: + raise CliException("Missing parameter --cloud") + mod_notification_arg_names = [ + "fail", + "success", + "initial", + "pre_initial", + "pre", + "one_day", + "three_days", + "five_days", + "seven_days", + ] + modify_notifications = list( + filter(lambda item: self.cli_args.get(item) is not None, mod_notification_arg_names) + ) + if not modify_notifications: + err_message = "" + for arg in mod_notification_arg_names: + err_message += f"\n--{arg.replace('_', '-')} [true|false]" + raise CliException("At least one arg should be given: {}".format(err_message)) + try: + assignment = self.quads.get_active_cloud_assignment(cloud_name=cloud_name) + except (APIServerException, APIBadRequest) as ex: + raise CliException(str(ex)) + if assignment: + assignment_id = assignment.id + payload = {} + for arg in modify_notifications: + payload.update({arg: self.cli_args.get(arg)}) + response = self.quads.update_notification(notification_id=assignment_id, data=payload) + if response.status_code == 200: + self.logger.info(f"{cloud_name}, Notification updated successfully\n") + self.action_list_notifications() + else: + self.logger.error("Something went wrong while updating notification") + else: + self.logger.warning(f"{cloud_name}, No active cloud assignment found") diff --git a/src/quads/cli/parser.py b/src/quads/cli/parser.py index 8282b0f1..414a01f0 100644 --- a/src/quads/cli/parser.py +++ b/src/quads/cli/parser.py @@ -699,6 +699,43 @@ nargs="*", help="Skip specific hosts, when validating Quads assignments", ) +parser.add_argument( + "--ls-notifications", + dest="action", + action="store_const", + const="list_notifications", + help="List notifications", +) +parser.add_argument( + "--mod-notification", + dest="action", + action="store_const", + const="modify_notification", + help="Modify notification of a cloud", +) + +mod_notification_arg_names = [ + "fail", "success", "initial", "pre-initial", + "pre", "one-day", "three-days", "five-days", "seven-days"] + + +def str_to_bool(value): + """Convert string to boolean.""" + if value.lower() in {'true', 'yes', '1'}: + return True + elif value.lower() in {'false', 'no', '0'}: + return False + else: + raise argparse.ArgumentTypeError(f"Invalid value: {value}. Expected 'true' or 'false'.") + + +for arg in mod_notification_arg_names: + parser.add_argument(f"--{arg}", + type=str_to_bool, + choices=[True, False], + help=f"Set {arg} to true or false.") + + if __name__ == "__main__": # pragma: no cover diff --git a/tests/cli/test_notifications.py b/tests/cli/test_notifications.py new file mode 100644 index 00000000..a585617b --- /dev/null +++ b/tests/cli/test_notifications.py @@ -0,0 +1,77 @@ +import pytest + +from quads.server.dao.assignment import AssignmentDao +from quads.server.dao.cloud import CloudDao +from tests.cli.config import ( + DEFINE_CLOUD, + FREE_CLOUD, MOD_CLOUD, +) +from tests.cli.test_base import TestBase + + +def finalizer(): + cloud = CloudDao.get_cloud(DEFINE_CLOUD) + assignment = AssignmentDao.get_active_cloud_assignment(cloud) + if assignment: + AssignmentDao.remove_assignment(assignment.id) + if cloud: + CloudDao.remove_cloud(DEFINE_CLOUD) + + +def finalizer_free(): + cloud_free = CloudDao.get_cloud(FREE_CLOUD) + if cloud_free: + CloudDao.remove_cloud(FREE_CLOUD) + + +@pytest.fixture(autouse=True) +def remove_cloud(request): + finalizer() + request.addfinalizer(finalizer) + + +@pytest.fixture(autouse=True) +def define_cloud(request): + CloudDao.create_cloud(DEFINE_CLOUD) + request.addfinalizer(finalizer) + + +@pytest.fixture(autouse=True) +def define_free_cloud(request): + cloud = CloudDao.create_cloud(FREE_CLOUD) + assignment = AssignmentDao.get_active_cloud_assignment(cloud) + if assignment: + AssignmentDao.update_assignment(assignment.id, **{"active": False}) + request.addfinalizer(finalizer_free) + + +class TestNotifications(TestBase): + def test_ls_notifications(self): + self.quads_cli_call("list_notifications") + assert len(self._caplog.messages) == 1 + + def test_ls_cloud_notification(self): + self.cli_args["cloud"] = MOD_CLOUD + self.quads_cli_call("list_notifications") + messages = self._caplog.messages + assert MOD_CLOUD in messages[0] + + def test_ls_non_exists_cloud_notifications(self): + self.cli_args["cloud"] = FREE_CLOUD + self.quads_cli_call("list_notifications") + messages = self._caplog.messages + assert messages[0] == f"WARNING: there are no current or future schedules for {FREE_CLOUD}" + + def test_modify_notification(self): + self.cli_args["cloud"] = MOD_CLOUD + self.cli_args["fail"] = 'true' + self.quads_cli_call("modify_notification") + messages = self._caplog.messages + assert f"{MOD_CLOUD}, Notification updated successfully".strip() == messages[0].strip() + + def test_modify_non_exists_cloud_notification(self): + self.cli_args["cloud"] = FREE_CLOUD + self.cli_args["fail"] = 'true' + self.quads_cli_call("modify_notification") + messages = self._caplog.messages + assert f"{FREE_CLOUD}, No active cloud assignment found".strip() == messages[0].strip()