🪲 BUG-#250: Fix _crontab_to_aws to convert */N to 0/N for EventBridge

docs/nav/development/release-notes.md

@@ -4,6 +4,7 @@
 
 - [📦 PyPI - Build 0.15.0](https://github.com/dotflow-io/dotflow/releases/tag/v0.15.0)
 - [🪲 Bug: _serialize_context crashes when storage is a list of non-Context objects](https://github.com/dotflow-io/dotflow/pull/249)
+- [🪲 Bug: _crontab_to_aws generates invalid EventBridge cron — */N not supported](https://github.com/dotflow-io/dotflow/pull/251)
 - [⚙️ Feature: Add Alibaba Cloud Function Compute deployer](https://github.com/dotflow-io/dotflow/pull/238)
 - [⚙️ Feature: Implement Server provider for remote API communication](https://github.com/dotflow-io/dotflow/pull/240)
 - [⚙️ Feature: dotflow deploy CLI — cross-cloud infrastructure generation](https://github.com/dotflow-io/dotflow/pull/183)

dotflow/cloud/aws/schedule.py

@@ -58,9 +58,22 @@ def _crontab_to_aws(expression: str) -> str:
 
     minute, hour, dom, month, dow = parts
 
-    if dow != "*" and dom == "*":
+    minute = _fix_step(minute)
+    hour = _fix_step(hour)
+    dom = _fix_step(dom)
+    month = _fix_step(month)
+    dow = _fix_step(dow)
+
+    if dow != "*" and dow != "?" and dom == "*":
         dom = "?"
     else:
         dow = "?"
 
     return f"cron({minute} {hour} {dom} {month} {dow} *)"
+
+
+def _fix_step(field: str) -> str:
+    """Convert */N to 0/N for AWS EventBridge compatibility."""
+    if field.startswith("*/"):
+        return "0/" + field[2:]
+    return field

tests/cloud/aws/test_schedule.py

@@ -4,40 +4,91 @@
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 
-from dotflow.cloud.aws.schedule import AWSSchedule
+from dotflow.cloud.aws.schedule import AWSSchedule, _fix_step
+
+
+class TestFixStep(unittest.TestCase):
+    def test_star_slash_converts_to_zero_slash(self):
+        self.assertEqual(_fix_step("*/5"), "0/5")
+
+    def test_star_slash_large_number(self):
+        self.assertEqual(_fix_step("*/30"), "0/30")
+
+    def test_plain_star_unchanged(self):
+        self.assertEqual(_fix_step("*"), "*")
+
+    def test_fixed_value_unchanged(self):
+        self.assertEqual(_fix_step("30"), "30")
+
+    def test_range_unchanged(self):
+        self.assertEqual(_fix_step("1-5"), "1-5")
+
+    def test_explicit_start_step_unchanged(self):
+        self.assertEqual(_fix_step("0/5"), "0/5")
+
+    def test_question_mark_unchanged(self):
+        self.assertEqual(_fix_step("?"), "?")
 
 
 class TestAWSScheduleConvert(unittest.TestCase):
     def test_crontab_every_hour(self):
         result = AWSSchedule.convert("0 * * * *")
+
         self.assertEqual(result, "cron(0 * * * ? *)")
 
     def test_crontab_every_5_minutes(self):
         result = AWSSchedule.convert("*/5 * * * *")
-        self.assertEqual(result, "cron(*/5 * * * ? *)")
+
+        self.assertEqual(result, "cron(0/5 * * * ? *)")
+
+    def test_crontab_every_30_minutes(self):
+        result = AWSSchedule.convert("*/30 * * * *")
+
+        self.assertEqual(result, "cron(0/30 * * * ? *)")
+
+    def test_crontab_step_in_hours(self):
+        result = AWSSchedule.convert("0 */6 * * *")
+
+        self.assertEqual(result, "cron(0 0/6 * * ? *)")
+
+    def test_crontab_step_in_minutes_and_hours(self):
+        result = AWSSchedule.convert("*/10 */2 * * *")
+
+        self.assertEqual(result, "cron(0/10 0/2 * * ? *)")
+
+    def test_crontab_no_step_unchanged(self):
+        result = AWSSchedule.convert("30 2 * * *")
+
+        self.assertEqual(result, "cron(30 2 * * ? *)")
 
     def test_crontab_specific_day_of_week(self):
         result = AWSSchedule.convert("0 6 * * 1")
+
         self.assertEqual(result, "cron(0 6 ? * 1 *)")
 
     def test_crontab_specific_day_of_month(self):
         result = AWSSchedule.convert("0 0 1 * *")
+
         self.assertEqual(result, "cron(0 0 1 * ? *)")
 
     def test_rate_passthrough(self):
         result = AWSSchedule.convert("rate(6 hours)")
+
         self.assertEqual(result, "rate(6 hours)")
 
     def test_cron_aws_passthrough(self):
         result = AWSSchedule.convert("cron(0 * * * ? *)")
+
         self.assertEqual(result, "cron(0 * * * ? *)")
 
     def test_invalid_returns_as_is(self):
         result = AWSSchedule.convert("not a cron")
+
         self.assertEqual(result, "not a cron")
 
     def test_strips_whitespace(self):
         result = AWSSchedule.convert("  rate(1 day)  ")
+
         self.assertEqual(result, "rate(1 day)")
 
 
@@ -63,6 +114,7 @@ def test_reads_schedule_from_sam_template(self):
 
     def test_returns_none_for_missing_file(self):
         result = AWSSchedule.read_from_template(Path("nonexistent.yaml"))
+
         self.assertIsNone(result)
 
     def test_returns_none_for_no_schedule(self):