test(live): assert weighted-capacity contract + fix ruff I001

- Multi-* termination tests conflated capacity_to_request (units) with
  physical machine count.  Weighted templates legitimately fulfil target
  with fewer machines (e.g. one t3.medium weight=4 covers target=2).
  Replace strict len(machines) == capacity_to_request with the actual
  contract: fulfilled_units >= target_units.  Preserve len(machines) >= 1
  as observable side-effect check for downstream EC2 state assertions.
- Drop stray blank line after imports in _capacity_helpers.py that
  tripped ruff I001 in CI-enforced lint.
- Apply ruff format wraps to _target/_fallback ternaries and terminal
  status set literal.

Author: fgogolli

tests/providers/aws/live/_capacity_helpers.py

@@ -40,7 +40,6 @@
 
 import pytest
 
-
 # Match the canonical "X/Y" partial-fulfilment pattern that ORB stamps on
 # partial messages (e.g. "Partially fulfilled: 2/4 instances",
 # "Instant fleet: 2/4 instance(s) running").

tests/providers/aws/live/test_multi_asg_termination.py

@@ -526,14 +526,21 @@ def test_multi_asg_termination(setup_multi_asg_templates):
 
             time.sleep(5)
 
-        _final_status = status_response["requests"][0]["status"]
-        machines = status_response["requests"][0]["machines"]
+        # Verify desired capacity delivered. capacity_to_request is in *capacity
+        # units*, not machine count: weighted templates (e.g. t3.medium weight=4)
+        # legitimately fulfil target with fewer physical machines than units.
+        # Contract: fulfilled_units >= target_units. machines>=1 is the
+        # observable side-effect we still need for downstream EC2 state checks.
+        _req = status_response["requests"][0]
+        _final_status = _req["status"]
+        machines = _req["machines"]
+        _target = int(_req.get("target_units") or capacity_to_request)
+        _fulfilled = int(_req.get("fulfilled_units") or len(machines))
         if _final_status == "complete":
-            assert len(machines) == capacity_to_request, (
-                f"status=complete but got {len(machines)}/{capacity_to_request} machines"
+            assert _fulfilled >= _target, (
+                f"status=complete but fulfilled_units={_fulfilled} < target_units={_target}"
             )
-        else:
-            assert len(machines) >= 1, f"status={_final_status!r} with zero machines"
+        assert len(machines) >= 1, f"status={_final_status!r} with zero machines"
 
         instance_ids = [
             machine.get("machineId") or machine.get("machine_id") for machine in machines

tests/providers/aws/live/test_multi_ec2_fleet_termination.py

@@ -548,16 +548,21 @@ def test_multi_ec2_fleet_termination(setup_multi_ec2_fleet_templates):
 
             time.sleep(5)
 
-        # Verify instances are provisioned (tolerant of AWS capacity shortfall;
-        # assert_terminal_ok above already enforced "some capacity delivered").
-        _final_status = status_response["requests"][0]["status"]
-        machines = status_response["requests"][0]["machines"]
+        # Verify desired capacity delivered. capacity_to_request is in *capacity
+        # units*, not machine count: weighted templates (e.g. t3.medium weight=4)
+        # legitimately fulfil target with fewer physical machines than units.
+        # Contract: fulfilled_units >= target_units. machines>=1 is the
+        # observable side-effect we still need for downstream EC2 state checks.
+        _req = status_response["requests"][0]
+        _final_status = _req["status"]
+        machines = _req["machines"]
+        _target = int(_req.get("target_units") or capacity_to_request)
+        _fulfilled = int(_req.get("fulfilled_units") or len(machines))
         if _final_status == "complete":
-            assert len(machines) == capacity_to_request, (
-                f"status=complete but got {len(machines)}/{capacity_to_request} machines"
+            assert _fulfilled >= _target, (
+                f"status=complete but fulfilled_units={_fulfilled} < target_units={_target}"
             )
-        else:
-            assert len(machines) >= 1, f"status={_final_status!r} with zero machines"
+        assert len(machines) >= 1, f"status={_final_status!r} with zero machines"
 
         instance_ids = [
             machine.get("machineId") or machine.get("machine_id") for machine in machines

tests/providers/aws/live/test_multi_resource_termination.py

@@ -563,14 +563,21 @@ def test_multi_resource_termination(setup_multi_resource_templates):
 
             time.sleep(5)
 
-        _final_status = status_response["requests"][0]["status"]
-        machines = status_response["requests"][0]["machines"]
+        # Verify desired capacity delivered. capacity_to_request is in *capacity
+        # units*, not machine count: weighted templates (e.g. t3.medium weight=4)
+        # legitimately fulfil target with fewer physical machines than units.
+        # Contract: fulfilled_units >= target_units. machines>=1 is the
+        # observable side-effect we still need for downstream EC2 state checks.
+        _req = status_response["requests"][0]
+        _final_status = _req["status"]
+        machines = _req["machines"]
+        _target = int(_req.get("target_units") or capacity_to_request)
+        _fulfilled = int(_req.get("fulfilled_units") or len(machines))
         if _final_status == "complete":
-            assert len(machines) == capacity_to_request, (
-                f"status=complete but got {len(machines)}/{capacity_to_request} machines"
+            assert _fulfilled >= _target, (
+                f"status=complete but fulfilled_units={_fulfilled} < target_units={_target}"
             )
-        else:
-            assert len(machines) >= 1, f"status={_final_status!r} with zero machines"
+        assert len(machines) >= 1, f"status={_final_status!r} with zero machines"
 
         instance_ids = [
             machine.get("machineId") or machine.get("machine_id") for machine in machines

tests/providers/aws/live/test_multi_spot_fleet_termination.py

@@ -532,14 +532,21 @@ def test_multi_spot_fleet_termination(setup_multi_spot_fleet_templates):
 
             time.sleep(5)
 
-        _final_status = status_response["requests"][0]["status"]
-        machines = status_response["requests"][0]["machines"]
+        # Verify desired capacity delivered. capacity_to_request is in *capacity
+        # units*, not machine count: weighted templates (e.g. t3.medium weight=4)
+        # legitimately fulfil target with fewer physical machines than units.
+        # Contract: fulfilled_units >= target_units. machines>=1 is the
+        # observable side-effect we still need for downstream EC2 state checks.
+        _req = status_response["requests"][0]
+        _final_status = _req["status"]
+        machines = _req["machines"]
+        _target = int(_req.get("target_units") or capacity_to_request)
+        _fulfilled = int(_req.get("fulfilled_units") or len(machines))
         if _final_status == "complete":
-            assert len(machines) == capacity_to_request, (
-                f"status=complete but got {len(machines)}/{capacity_to_request} machines"
+            assert _fulfilled >= _target, (
+                f"status=complete but fulfilled_units={_fulfilled} < target_units={_target}"
             )
-        else:
-            assert len(machines) >= 1, f"status={_final_status!r} with zero machines"
+        assert len(machines) >= 1, f"status={_final_status!r} with zero machines"
 
         instance_ids = [
             machine.get("machineId") or machine.get("machine_id") for machine in machines