Merge "Added the ls-notification" into latest

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")

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

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()