feat(reservations): allow admins to bypass maintenance window check

Adds a warre:reservation:create:bypass_maintenance policy (admin-only by
default) so operators can create test reservations that overlap an active
maintenance window. Capacity from real reservations is still enforced.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Change-Id: Icc9ed9a09ee520cd33ead178f14d87a1b4138326

Author: sorrison

warre/api/v1/resources/reservation.py

@@ -97,9 +97,15 @@ def post(self, **kwargs):
         except limit_exceptions.ProjectOverLimit as e:
             return {"error_message": str(e)}, 413
 
+        bypass_maintenance = self.authorize(
+            "create:bypass_maintenance", do_raise=False
+        )
+
         try:
             reservation = self.manager.create_reservation(
-                self.context, reservation
+                self.context,
+                reservation,
+                bypass_maintenance=bypass_maintenance,
             )
         except exceptions.InvalidReservation as err:
             LOG.info("Failed to create reservation: %s", err)

warre/common/policies.py

@@ -230,6 +230,13 @@
         description="List all reservations.",
         operations=[{"path": "/v1/reservations/", "method": "GET"}],
     ),
+    policy.DocumentedRuleDefault(
+        name=RESERVATION_PREFIX % "create:bypass_maintenance",
+        check_str="rule:admin_required",
+        description="Create a reservation that overlaps a maintenance "
+        "window (intended for admin testing during maintenance).",
+        operations=[{"path": "/v1/reservations/", "method": "POST"}],
+    ),
     policy.DocumentedRuleDefault(
         name=RESERVATION_PREFIX % "update",
         check_str=f"rule:{ADMIN_OR_OWNER}",

warre/manager.py

@@ -31,7 +31,9 @@ def __init__(self):
         self.worker_api = worker_api.WorkerAPI()
         self.blazar = blazar.BlazarClient()
 
-    def create_reservation(self, context, reservation):
+    def create_reservation(
+        self, context, reservation, bypass_maintenance=False
+    ):
         flavor = db.session.query(models.Flavor).get(reservation.flavor_id)
         if not flavor.active:
             raise exceptions.InvalidReservation("Flavor is not available")
@@ -66,28 +68,30 @@ def create_reservation(self, context, reservation):
                 f"reservation end time of {reservation.end}"
             )
 
-        maintenance_windows = (
-            db.session.query(models.MaintenanceWindow)
-            .filter(models.MaintenanceWindow.end >= reservation.start)
-            .filter(models.MaintenanceWindow.start <= reservation.end)
-            .filter(
-                models.MaintenanceWindow.flavors.any(
-                    models.Flavor.id == flavor.id
+        if not bypass_maintenance:
+            maintenance_windows = (
+                db.session.query(models.MaintenanceWindow)
+                .filter(models.MaintenanceWindow.end >= reservation.start)
+                .filter(models.MaintenanceWindow.start <= reservation.end)
+                .filter(
+                    models.MaintenanceWindow.flavors.any(
+                        models.Flavor.id == flavor.id
+                    )
                 )
+                .count()
             )
-            .count()
-        )
-        if maintenance_windows > 0:
-            raise exceptions.InvalidReservation(
-                "Reservation conflicts with a maintenance window"
-            )
+            if maintenance_windows > 0:
+                raise exceptions.InvalidReservation(
+                    "Reservation conflicts with a maintenance window"
+                )
 
         free_slots = self.flavor_free_slots(
             context,
             flavor,
             reservation.start,
             reservation.end,
             reservation,
+            include_maintenance=not bypass_maintenance,
         )
 
         if free_slots:
@@ -168,14 +172,24 @@ def delete_flavor(self, context, flavor):
         db.session.delete(flavor)
         db.session.commit()
 
-    def flavor_free_slots(self, context, flavor, start, end, reservation=None):
+    def flavor_free_slots(
+        self,
+        context,
+        flavor,
+        start,
+        end,
+        reservation=None,
+        include_maintenance=True,
+    ):
         """Get the free slots of a flavor
         Algorithm:
         1. Get slots from flavor table as the total resource
         2. Get all reservations of current flavor
         3. Calculate the free slots
 
         reservation - used to exclude an existing reservation when extending
+        include_maintenance - when False, treat maintenance windows as not
+            occupying any slots (admin bypass for testing during a window)
         """
         if not flavor.active:
             return []
@@ -209,17 +223,20 @@ def flavor_free_slots(self, context, flavor, start, end, reservation=None):
             query = query.filter(models.Reservation.id != reservation.id)
         reservations = query.all()
 
-        maintenance_windows = (
-            db.session.query(models.MaintenanceWindow)
-            .filter(models.MaintenanceWindow.end >= start)
-            .filter(models.MaintenanceWindow.start <= end)
-            .filter(
-                models.MaintenanceWindow.flavors.any(
-                    models.Flavor.id == flavor.id
+        if include_maintenance:
+            maintenance_windows = (
+                db.session.query(models.MaintenanceWindow)
+                .filter(models.MaintenanceWindow.end >= start)
+                .filter(models.MaintenanceWindow.start <= end)
+                .filter(
+                    models.MaintenanceWindow.flavors.any(
+                        models.Flavor.id == flavor.id
+                    )
                 )
+                .all()
             )
-            .all()
-        )
+        else:
+            maintenance_windows = []
 
         # the real thing begins here
         # Pass 1: segmentation and marking

warre/tests/unit/api/v1/test_reservation.py

@@ -262,10 +262,43 @@ def test_update_reservation_no_capacity(self, mock_blazar):
             reservation_json.get("error_message"),
         )
 
+    def test_create_reservation_during_maintenance_window_denied(self):
+        self.create_maintenance_window(
+            start=datetime.datetime(2020, 1, 1),
+            end=datetime.datetime(2020, 1, 10),
+            flavors=[self.flavor],
+        )
+        data = {
+            "flavor_id": self.flavor.id,
+            "start": "2020-01-02T00:00:00+00:00",
+            "end": "2020-01-02T01:00:00+00:00",
+        }
+        response = self.client.post("/v1/reservations/", json=data)
+        self.assert400(response)
+        self.assertIn(
+            "maintenance window", response.get_json().get("error_message")
+        )
+
 
+@mock.patch("warre.quota.get_enforcer", new=mock.Mock())
 class TestAdminReservationAPI(TestReservationAPI):
     ROLES = ["admin"]
 
+    def test_create_reservation_during_maintenance_window_denied(self):
+        # Admins are allowed to bypass the maintenance window check.
+        self.create_maintenance_window(
+            start=datetime.datetime(2020, 1, 1),
+            end=datetime.datetime(2020, 1, 10),
+            flavors=[self.flavor],
+        )
+        data = {
+            "flavor_id": self.flavor.id,
+            "start": "2020-01-02T00:00:00+00:00",
+            "end": "2020-01-02T01:00:00+00:00",
+        }
+        response = self.client.post("/v1/reservations/", json=data)
+        self.assert200(response)
+
     def test_list_reservations_all_projects(self):
         self.create_reservation(
             flavor_id=self.flavor.id,

warre/tests/unit/test_manager.py

@@ -893,3 +893,49 @@ def test_create_reservation_maintenance_window_other_flavor(
         mgr = manager.Manager()
         result = mgr.create_reservation(self.context, reservation)
         self.assertEqual(self.flavor, result.flavor)
+
+    def test_create_reservation_bypass_maintenance_window(self, mock_worker):
+        self.create_maintenance_window(
+            start=datetime(2021, 1, 1),
+            end=datetime(2021, 1, 10),
+            flavors=[self.flavor],
+        )
+        reservation = models.Reservation(
+            flavor_id=self.flavor.id,
+            start=datetime(2021, 1, 3),
+            end=datetime(2021, 1, 5),
+        )
+        mgr = manager.Manager()
+        result = mgr.create_reservation(
+            self.context, reservation, bypass_maintenance=True
+        )
+        self.assertEqual(self.flavor, result.flavor)
+
+    def test_create_reservation_bypass_still_enforces_capacity(
+        self, mock_worker
+    ):
+        self.create_maintenance_window(
+            start=datetime(2021, 1, 1),
+            end=datetime(2021, 1, 10),
+            flavors=[self.flavor],
+        )
+        # Existing reservation already consumes the only slot during the
+        # maintenance window — bypass must not let us double-book.
+        self.create_reservation(
+            flavor_id=self.flavor.id,
+            start=datetime(2021, 1, 3),
+            end=datetime(2021, 1, 5),
+            status=models.Reservation.ALLOCATED,
+        )
+        reservation = models.Reservation(
+            flavor_id=self.flavor.id,
+            start=datetime(2021, 1, 3),
+            end=datetime(2021, 1, 5),
+        )
+        mgr = manager.Manager()
+        with self.assertRaisesRegex(
+            exceptions.InvalidReservation, "No capacity"
+        ):
+            mgr.create_reservation(
+                self.context, reservation, bypass_maintenance=True
+            )