Added flag to command slcli account events --date-min

Author: edsonarios

SoftLayer/CLI/account/events.py

@@ -11,21 +11,25 @@
 
 @click.command(cls=SLCommand)
 @click.option('--ack-all', is_flag=True, default=False,
-              help="Acknowledge every upcoming event. Doing so will turn off the popup in the control portal")
+              help="Acknowledge every upcoming event. Doing so will turn off the popup in the control portal.")
 @click.option('--planned', is_flag=True, default=False,
-              help="Show only planned events")
+              help="Show only planned events.")
 @click.option('--unplanned', is_flag=True, default=False,
-              help="Show only unplanned events")
+              help="Show only unplanned events.")
 @click.option('--announcement', is_flag=True, default=False,
-              help="Show only announcement events")
+              help="Show only announcement events.")
+@click.option('--date-min', help="Earliest date to retrieve events for [MM/DD/YYYY]. Default: 2 days ago.")
 @environment.pass_env
-def cli(env, ack_all, planned, unplanned, announcement):
+def cli(env, ack_all, planned, unplanned, announcement, date_min):
     """Summary and acknowledgement of upcoming and ongoing maintenance events"""
 
+    if date_min:
+        utils.verify_date(date_min)
+
     manager = AccountManager(env.client)
-    planned_events = manager.get_upcoming_events("PLANNED")
-    unplanned_events = manager.get_upcoming_events("UNPLANNED_INCIDENT")
-    announcement_events = manager.get_upcoming_events("ANNOUNCEMENT")
+    planned_events = manager.get_upcoming_events("PLANNED", date_min)
+    unplanned_events = manager.get_upcoming_events("UNPLANNED_INCIDENT", date_min)
+    announcement_events = manager.get_upcoming_events("ANNOUNCEMENT", date_min)
 
     add_ack_flag(planned_events, manager, ack_all)
     add_ack_flag(unplanned_events, manager, ack_all)

SoftLayer/managers/account.py

@@ -53,10 +53,11 @@ def get_summary(self):
         """
         return self.client.call('Account', 'getObject', mask=mask)
 
-    def get_upcoming_events(self, event_type):
+    def get_upcoming_events(self, event_type, date_min):
         """Retrieves a list of Notification_Occurrence_Events that have not ended yet
 
         :param: String event_type: notification event type.
+        :param: String date_min: greater Than Date to data recovery, default is 2 days ago.
         :return: SoftLayer_Notification_Occurrence_Event
         """
         mask = "mask[id, subject, startDate, endDate, modifyDate, statusCode, acknowledgedFlag, " \
@@ -70,21 +71,32 @@ def get_upcoming_events(self, event_type):
             }
         }
 
-        self.add_event_filter(_filter, event_type)
+        self.add_event_filter(_filter, event_type, date_min)
 
         return self.client.call('Notification_Occurrence_Event', 'getAllObjects', filter=_filter, mask=mask, iter=True)
 
     @staticmethod
-    def add_event_filter(_filter, event_type):
+    def add_event_filter(_filter, event_type, date_min):
         """Add data to the object filter.
 
         :param: _filter: event filter.
         :param: string event_type: event type.
+        :param: string date_min: greater Than Date to data recovery, default is 2 days ago.
         """
+
         if event_type == 'PLANNED':
-            _filter['endDate'] = {
-                'operation': '> sysdate - 2'
-            }
+            if date_min:
+                _filter['endDate'] = {
+                    'operation': 'greaterThanDate',
+                    'options': [{
+                        'name': 'date',
+                        'value': [date_min]
+                    }]
+                }
+            else:
+                _filter['endDate'] = {
+                    'operation': '> sysdate - 2'
+                }
             _filter['startDate'] = {
                 'operation': 'orderBy',
                 'options': [{
@@ -94,9 +106,18 @@ def add_event_filter(_filter, event_type):
             }
 
         if event_type == 'UNPLANNED_INCIDENT':
-            _filter['modifyDate'] = {
-                'operation': '> sysdate - 2'
-            }
+            if date_min:
+                _filter['modifyDate'] = {
+                    'operation': 'greaterThanDate',
+                    'options': [{
+                        'name': 'date',
+                        'value': [date_min]
+                    }]
+                }
+            else:
+                _filter['modifyDate'] = {
+                    'operation': '> sysdate - 2'
+                }
 
         if event_type == 'ANNOUNCEMENT':
             _filter['statusCode'] = {

SoftLayer/utils.py

@@ -13,7 +13,7 @@
 
 from rich.console import Console
 from rich.theme import Theme
-
+from SoftLayer.CLI import exceptions
 # pylint: disable=no-member, invalid-name
 
 UUID_RE = re.compile(r'^[0-9A-F]{8}-[0-9A-F]{4}-4[0-9A-F]{3}-[89AB][0-9A-F]{3}-[0-9A-F]{12}$', re.I)
@@ -371,6 +371,18 @@ def clean_time(sltime, in_format='%Y-%m-%dT%H:%M:%S%z', out_format='%Y-%m-%d %H:
         return sltime
 
 
+def verify_date(date):
+    """Verify if the date format is correct
+
+    :param string date: A date in format string
+    :return a exception if the date is not the correct format
+    """
+    try:
+        date = datetime.datetime.strptime(date, '%m/%d/%Y')
+    except Exception as exc:
+        raise exceptions.CLIAbort('Date invalid, example date: mm/dd/yyyy, 01/15/2023.') from exc
+
+
 def timestamp(date):
     """Converts a datetime to timestamp
 

tests/managers/account_tests.py

@@ -21,15 +21,15 @@ def test_get_summary(self):
         self.assert_called_with('SoftLayer_Account', 'getObject')
 
     def test_get_planned_upcoming_events(self):
-        self.manager.get_upcoming_events("PLANNED")
+        self.manager.get_upcoming_events("PLANNED", "")
         self.assert_called_with(self.SLNOE, 'getAllObjects')
 
     def test_get_unplanned_upcoming_events(self):
-        self.manager.get_upcoming_events("UNPLANNED_INCIDENT")
+        self.manager.get_upcoming_events("UNPLANNED_INCIDENT", "")
         self.assert_called_with(self.SLNOE, 'getAllObjects')
 
     def test_get_announcement_upcoming_events(self):
-        self.manager.get_upcoming_events("ANNOUNCEMENT")
+        self.manager.get_upcoming_events("ANNOUNCEMENT", "")
         self.assert_called_with(self.SLNOE, 'getAllObjects')
 
     def test_add_planned_event_filter(self):
@@ -41,7 +41,7 @@ def test_add_planned_event_filter(self):
                 }
             }
         }
-        self.manager.add_event_filter(_filter, event_type)
+        self.manager.add_event_filter(_filter, event_type, "")
 
     def test_add_unplanned_event_filter(self):
         event_type = 'UNPLANNED_INCIDENT'
@@ -52,7 +52,7 @@ def test_add_unplanned_event_filter(self):
                 }
             }
         }
-        self.manager.add_event_filter(_filter, event_type)
+        self.manager.add_event_filter(_filter, event_type, "")
 
     def test_add_announcement_event_filter(self):
         event_type = 'ANNOUNCEMENT'
@@ -63,7 +63,7 @@ def test_add_announcement_event_filter(self):
                 }
             }
         }
-        self.manager.add_event_filter(_filter, event_type)
+        self.manager.add_event_filter(_filter, event_type, "")
 
     def test_ack_event(self):
         self.manager.ack_event(12345)