feat(cssc): 31005729 support cron expressions for '--schedule' parameter

src/acrcssc/azext_acrcssc/_params.py

@@ -20,13 +20,13 @@ def load_arguments(self: AzCommandsLoader, _):
 
     with self.argument_context("acr supply-chain workflow create") as c:
         c.argument("config", options_list=["--config"], help="Configuration file path containing the json schema for the list of repositories and tags to filter within the registry. Schema example:{\"repositories\":[{\"repository\":\"alpine\",\"tags\":[\"tag1\",\"tag2\"],\"enabled\":true},{\"repository\":\"python\",\"tags\":[\"*\"],\"enabled\":false}], \"version\": \"v1\", \"tag-convention\": \"floating\"}. \"tag-convention\" is an optional property, values can be \"incremental\" (the default behavior, will increase the patch version of the tag, for example \"{repository}:{original-tag}-1\", \"{repository}:{original-tag}-2\", etc), or \"floating\" (will reuse the tag \"{repository}:{original-tag}-patched\" for patching)", required=True)
-        c.argument("schedule", options_list=["--schedule"], help="schedule to run the scan and patching task. E.g. `<n>d` where <n> is the number of days between each run. Max value is 30d.", required=True)
+        c.argument("schedule", options_list=["--schedule"], help="Schedule to run the scan and patching task. It can be a timespan in the format `<n>d` where <n> is the number of days between each run (e.g., `5d` for every 5 days, max value is 30d), or a cron expression (e.g., `0 0 * * *` for daily at midnight).", required=True)
         c.argument("run_immediately", options_list=["--run-immediately"], help="Set this flag to trigger the immediate run of the selected workflow task. Default value: false.", arg_type=get_three_state_flag(), required=False)
         c.argument("dryrun", options_list=["--dry-run"], help="Use this flag to see the qualifying repositories and tags that would be affected by the workflow. Default value: false. 'config' parameter is mandatory to provide with dry-run", arg_type=get_three_state_flag(), required=False)
 
     with self.argument_context("acr supply-chain workflow update") as c:
         c.argument("config", options_list=["--config"], help="Configuration file path containing the json schema for the list of repositories and tags to filter within the registry. Schema example:{\"repositories\":[{\"repository\":\"alpine\",\"tags\":[\"tag1\",\"tag2\"],\"enabled\":true},{\"repository\":\"python\",\"tags\":[\"*\"],\"enabled\":false}], \"version\": \"v1\", \"tag-convention\": \"floating\"}}. \"tag-convention\" is an optional property, values can be \"incremental\" (the default behavior, will increase the patch version of the tag, for example \"{repository}:{original-tag}-1\", \"{repository}:{original-tag}-2\", etc), or \"floating\" (will reuse the tag \"{repository}:{original-tag}-patched\" for patching)", required=False)
-        c.argument("schedule", options_list=["--schedule"], help="schedule to run the scan and patching task. E.g. `<n>d` where n is the number of days between each run. Max value is 30d.", required=False)
+        c.argument("schedule", options_list=["--schedule"], help="Schedule to run the scan and patching task. It can be a timespan in the format `<n>d` where <n> is the number of days between each run (e.g., `5d` for every 5 days, max value is 30d), or a cron expression (e.g., `0 0 * * *` for daily at midnight).", required=False)
         c.argument("run_immediately", options_list=["--run-immediately"], help="Set this flag to trigger the immediate run of the selected workflow task. Default value: false.", arg_type=get_three_state_flag(), required=False)
         c.argument("dryrun", options_list=["--dry-run"], help="Use this flag to see the qualifying repositories and tags that would be affected by the workflow. Default value: false. 'config' parameter is mandatory to provide with dry-run", arg_type=get_three_state_flag(), required=False)
 

src/acrcssc/azext_acrcssc/_validators.py

@@ -9,6 +9,7 @@
 import os
 import re
 from knack.log import get_logger
+from croniter import croniter, CroniterBadCronError
 from azure.cli.command_modules.acr.repository import acr_repository_show
 from .helper._constants import (
     BEARER_TOKEN_USERNAME,
@@ -20,14 +21,15 @@
     CONTINUOUSPATCH_ALL_TASK_NAMES,
     ERROR_MESSAGE_INVALID_TIMESPAN_FORMAT,
     ERROR_MESSAGE_INVALID_TIMESPAN_VALUE,
-    RESOURCE_GROUP,
+    SCHEDULE_MIN_DAYS,
+    SCHEDULE_MAX_DAYS,
     SUBSCRIPTION)
 from .helper._constants import CSSCTaskTypes, ERROR_MESSAGE_INVALID_TASK, RECOMMENDATION_SCHEDULE
 from .helper._ociartifactoperations import _get_acr_token
 from azure.mgmt.core.tools import (parse_resource_id)
 from azure.cli.core.azclierror import InvalidArgumentValueError
 from ._client_factory import cf_acr_tasks
-from .helper._utility import get_task
+from .helper._utility import get_task, schedule_timespan_format
 
 logger = get_logger(__name__)
 
@@ -120,19 +122,27 @@ def _check_task_exists(cmd, registry, task_name=""):
     return False
 
 
+# schedule can be both a timespan (1d) or a cron expression (0 0 * * *), need to validate both
 def _validate_schedule(schedule):
     # during update, schedule can be null if we are only updating the config
     if schedule is None:
         return
+
+    # try to match a timespan first
     # Extract the numeric value and unit from the timespan expression
-    match = re.match(r'(\d+)(d)$', schedule)
-    if not match:
-        raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_FORMAT, recommendation=RECOMMENDATION_SCHEDULE)
+    match = schedule_timespan_format(schedule)
     if match is not None:
-        value = int(match.group(1))
-        unit = match.group(2)
-    if unit == 'd' and (value < 1 or value > 30):  # day of the month
-        raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_VALUE, recommendation=RECOMMENDATION_SCHEDULE)
+        if match < SCHEDULE_MIN_DAYS or match > SCHEDULE_MAX_DAYS:  # day of the month
+            raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_VALUE,
+                                            recommendation=RECOMMENDATION_SCHEDULE)
+    else:
+        try:
+            if not croniter(schedule).is_valid:
+                raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_FORMAT,
+                                                recommendation=RECOMMENDATION_SCHEDULE)
+        except CroniterBadCronError:
+            raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_FORMAT,
+                                            recommendation=RECOMMENDATION_SCHEDULE)
 
 
 def validate_inputs(schedule, config_file_path=None):

src/acrcssc/azext_acrcssc/helper/_constants.py

@@ -36,6 +36,8 @@ class TaskRunStatus(Enum):
 RESOURCE_GROUP = "resource_group"
 SUBSCRIPTION = "subscription"
 TMP_DRY_RUN_FILE_NAME = "tmp_dry_run_template.yaml"
+SCHEDULE_MIN_DAYS = 1
+SCHEDULE_MAX_DAYS = 30
 
 
 # Continuous Patch Constants
@@ -68,7 +70,7 @@ class TaskRunStatus(Enum):
 ERROR_MESSAGE_INVALID_TASK = "Workflow type is invalid"
 ERROR_MESSAGE_INVALID_TIMESPAN_VALUE = "Schedule value is invalid. "
 ERROR_MESSAGE_INVALID_TIMESPAN_FORMAT = "Schedule format is invalid. "
-RECOMMENDATION_SCHEDULE = "Schedule must be in the format of <number><unit> where unit is d for days. Example: 1d. Max value for d is 30d."
+RECOMMENDATION_SCHEDULE = "Schedule must be in the format of <number><unit> where unit is d for days. Example: 1d. Max value for d is 30d.\nOr a valid cron expression. Example: 0 0 * * *"
 # this dictionary can be expanded to handle more configuration of the tasks regarding continuous patching
 # if this gets out of hand, or more types of tasks are supported, this should be a class on its own
 CONTINUOUSPATCH_TASK_DEFINITION = {

src/acrcssc/azext_acrcssc/helper/_utility.py

@@ -10,29 +10,42 @@
 from azure.cli.core.azclierror import InvalidArgumentValueError
 from ._constants import ERROR_MESSAGE_INVALID_TIMESPAN_VALUE, TMP_DRY_RUN_FILE_NAME
 from azure.mgmt.core.tools import parse_resource_id
-from ._constants import RESOURCE_GROUP
+from ._constants import (
+    ERROR_MESSAGE_INVALID_TIMESPAN_FORMAT,
+    RESOURCE_GROUP,
+    SCHEDULE_MIN_DAYS,
+    SCHEDULE_MAX_DAYS
+)
 from .._client_factory import cf_acr_tasks
 
 logger = get_logger(__name__)
 # pylint: disable=logging-fstring-interpolation
 
 
+# this is a cheaper regex to match than the cron expression
+# Regex to look for pattern 1d, 2d, 3d, etc
+def schedule_timespan_format(schedule):
+    match = re.match(r'(\d+)d$', schedule)
+    if match is not None:
+        return int(match.group(1))
+    return None
+
+
 def convert_timespan_to_cron(schedule, date_time=None):
-    # Regex to look for pattern 1d, 2d, 3d, etc.
-    match = re.match(r'(\d+)([d])', schedule)
-    value = int(match.group(1))
-    unit = match.group(2)
+    # only timespan and cron formats are supported, and 'schedule' has already been validated
+    match = schedule_timespan_format(schedule)
+    if not match:
+        return schedule
 
     if date_time is None:
         date_time = datetime.now(timezone.utc)
 
     cron_hour = date_time.hour
     cron_minute = date_time.minute
 
-    if unit == 'd':  # day of the month
-        if value < 1 or value > 30:
-            raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_VALUE)
-        cron_expression = f'{cron_minute} {cron_hour} */{value} * *'
+    if match < SCHEDULE_MIN_DAYS or match > SCHEDULE_MAX_DAYS:
+        raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_VALUE)
+    cron_expression = f'{cron_minute} {cron_hour} */{match} * *'
 
     return cron_expression
 
@@ -45,10 +58,19 @@ def transform_cron_to_schedule(cron_expression, just_days=False):
     match = re.search(r'\*/(\d+)', third_part)
 
     if match:
+        days = int(match.group(1))
+
+        # cron expressions like "0 0 */99 * *" are valid (it will only trigger on the 1st of every month), but displaying it as days makes no sense.
+        # Display the full cron expression so the user can see what's going on.
+        if days < 1 or days > 31:
+            return cron_expression
+
         if just_days:
-            return match.group(1)
-        return match.group(1) + 'd'
-    return None
+            return days
+        return f"{days}d"
+
+    # if the cron expression is not in the format */n, return the cron expression as is
+    return cron_expression
 
 
 def create_temporary_dry_run_file(file_location, tmp_folder):

src/acrcssc/azext_acrcssc/tests/latest/test_validators.py

@@ -21,17 +21,30 @@ class AcrCsscCommandsTests(unittest.TestCase):
 
     def test_validate_schedule_valid(self):
         test_cases = [
-            ('1d' ),
+            ('1d'),
             ('5d'),
-            ('10d')
+            ('10d'),
+            ('* * * * *'),
+            ('0 0 * * *'),
+            ('20 4 */7 * *'),
+            ('20 4 2-30/7 * *')
         ]
 
         for timespan in test_cases:
             with self.subTest(timespan=timespan):
-               _validate_schedule(timespan)
-    
+                _validate_schedule(timespan)
+
     def test_validate_schedule_invalid(self):
-        test_cases = [('df'),('12'),('dd'),('41d'), ('21dd')]
+        test_cases = [
+            ('df'),
+            ('12'),
+            ('dd'),
+            ('41d'),
+            ('21dd'),
+            ('* * *'),
+            ('99 * * * *'),
+            ('* 9-100 * * *')
+        ]
 
         for timespan in test_cases:
             self.assertRaises(InvalidArgumentValueError, _validate_schedule, timespan)