feat: add enforce support for pipeline schedule

docs/reference/pipeline_schedules.md

@@ -12,7 +12,13 @@ Additionally, under a `variables` key you can add the pipeline schedule variable
 
     * Do not set `description` attribute - see [#535](https://github.com/gitlabform/gitlabform/issues/535#issue-1678509984)
 
-Example:
+There are 2 gitlabform specific keys/configs that can be set under `schedules` or individual schedule:
+
+- `delete` - set this to `true` under a specific schedule to delete that particular schedule.
+- `enforce` - set this to `true` under `schedules` so that any schedules that are not in `schedules` section are deleted.
+
+
+Example 1:
 ```yaml
 projects_and_groups:
   group_1/project_1:
@@ -32,5 +38,18 @@ projects_and_groups:
           other_variable:
             value: another_value
       "Obsolete schedule":
-        delete: true
+        delete: true  # Delete this schedule
+```
+
+Example 2:
+```yaml
+projects_and_groups:
+  group_1/project_1:
+    schedules:
+      enforce: true  # Delete all other pipeline schedules that exists for this project
+      "Some schedule":
+        ref: main
+        cron: "0 * * * MON-FRI"
+        cron_timezone: "London"
+        active: false
 ```

gitlabform/processors/project/schedules_processor.py

@@ -10,14 +10,22 @@
         super().__init__("schedules", gitlab)
 
     def _process_configuration(self, project_and_group: str, configuration: dict):
+        configured_schedules = configuration.get("schedules", {})
+        enforce_schedules = configuration.get("schedules|enforce", False)
+
+        # Remove 'enforce' key from the config so that it's not treated as a "schedule"
+        if enforce_schedules:
+            configured_schedules.pop("enforce")
+
         existing_schedules = self.gitlab.get_all_pipeline_schedules(project_and_group)
         schedule_ids_by_description = self.__group_schedule_ids_by_description(
             existing_schedules
         )
 
-        for schedule_description in sorted(configuration["schedules"]):
+        for schedule_description in sorted(configured_schedules):
             schedule_ids = schedule_ids_by_description.get(schedule_description)
-            if configuration.get("schedules|" + schedule_description + "|delete"):
+
+            if configured_schedules[schedule_description].get("delete"):
                 if schedule_ids:
                     debug("Deleting pipeline schedules '%s'", schedule_description)
                     for schedule_id in schedule_ids:
@@ -66,6 +74,21 @@
                         configuration, project_and_group, schedule_description
                     )
 
+        if enforce_schedules:
+            debug("Delete unconfigured schedules because enforce is enabled")
+
+            for schedule in existing_schedules:
+                schedule_description = schedule["description"]
+                schedule_id = schedule["id"]
+
+                debug(f"processing {schedule_id}: {schedule_description}")
+                if schedule_description not in configured_schedules:
+                    debug(
+                        "Deleting pipeline schedule named '%s', because it is not in gitlabform configuration",
+                        schedule_description,
+                    )
+                    self.gitlab.delete_pipeline_schedule(project_and_group, schedule_id)
+
     def create_schedule_with_variables(
         self, configuration, project_and_group, schedule_description
     ):

tests/acceptance/standard/test_schedules.py

@@ -199,6 +199,95 @@ def test__delete_schedule(self, project, schedules):
         )
         assert schedule is None
 
+    def test__schedule_enforce_new_schedule(self, project, schedules):
+        new_schedule = f"""
+        projects_and_groups:
+          {project.path_with_namespace}:
+            schedules:
+              enforce: true
+              "New schedule to test enforce config":
+                ref: main
+                cron: "30 1 * * *"
+        """
+
+        schedules_before = project.pipelineschedules.list()
+
+        run_gitlabform(new_schedule, project)
+
+        schedules_after = project.pipelineschedules.list()
+
+        schedule = self.__find_pipeline_schedule_by_description_and_get_first(
+            project, "New schedule to test enforce config"
+        )
+        assert len(schedules_before) == 6
+        assert len(schedules_after) == 1
+        assert schedule is not None
+        assert schedule.description == "New schedule to test enforce config"
+        assert schedule.ref == "main"
+        assert schedule.cron == "30 1 * * *"
+        assert schedule.cron_timezone == "UTC"
+        assert schedule.active is True
+
+    def test__schedule_enforce_new_and_existing_schedule(self, project, schedules):
+        new_schedule = f"""
+        projects_and_groups:
+          {project.path_with_namespace}:
+            schedules:
+              enforce: true
+              "New schedule to test enforce config":
+                ref: main
+                cron: "30 1 * * *"
+              "New schedule with variables":
+                ref: main
+                cron: "30 1 * * *"
+                variables:
+                    var1:
+                        value: value123
+                    var2:
+                        value: value987
+                        variable_type: file
+        """
+
+        schedules_before = project.pipelineschedules.list()
+
+        run_gitlabform(new_schedule, project)
+
+        schedules_after = project.pipelineschedules.list()
+
+        schedule1 = self.__find_pipeline_schedule_by_description_and_get_first(
+            project, "New schedule to test enforce config"
+        )
+        schedule2 = self.__find_pipeline_schedule_by_description_and_get_first(
+            project, "New schedule with variables"
+        )
+        assert len(schedules_before) == 1
+        assert len(schedules_after) == 2
+
+        assert schedule1 is not None
+        assert schedule1.description == "New schedule to test enforce config"
+        assert schedule1.ref == "main"
+        assert schedule1.cron == "30 1 * * *"
+        assert schedule1.cron_timezone == "UTC"
+        assert schedule1.active is True
+
+        assert schedule2 is not None
+        assert schedule2.description == "New schedule with variables"
+        assert schedule2.ref == "main"
+        assert schedule2.cron == "30 1 * * *"
+        assert schedule2.cron_timezone == "UTC"
+        assert schedule2.active is True
+
+        variables = schedule2.attributes["variables"]
+        assert variables is not None
+        assert len(variables) == 2
+        assert variables[0]["variable_type"] == "env_var"
+        assert variables[0]["key"] == "var1"
+        assert variables[0]["value"] == "value123"
+
+        assert variables[1]["variable_type"] == "file"
+        assert variables[1]["key"] == "var2"
+        assert variables[1]["value"] == "value987"
+
     @classmethod
     def __find_pipeline_schedule_by_description_and_get_first(
         cls, project, description