Ability to override state file name for test runs

Author: Peter-Sh

src/redis_release/cli.py

@@ -107,6 +107,11 @@ def release(
         "--force-release-type",
         help="Force release type (public or internal)",
     ),
+    override_state_name: Optional[str] = typer.Option(
+        None,
+        "--override-state-name",
+        help="Custom state name to use instead of release tag, to be able to make test runs without affecting production state",
+    ),
 ) -> None:
     """Run release using behaviour tree implementation."""
     setup_logging()
@@ -119,6 +124,7 @@ def release(
         force_rebuild=force_rebuild or [],
         only_packages=only_packages or [],
         force_release_type=force_release_type,
+        override_state_name=override_state_name,
     )
 
     # Use context manager version with automatic lock management

src/redis_release/models.py

@@ -160,3 +160,4 @@ class ReleaseArgs(BaseModel):
     force_rebuild: List[str] = Field(default_factory=list)
     only_packages: List[str] = Field(default_factory=list)
     force_release_type: Optional[ReleaseType] = None
+    override_state_name: Optional[str] = None

src/redis_release/state_manager.py

@@ -167,24 +167,34 @@ def __init__(
         self._lock_acquired = False
         self.read_only = read_only
 
+        # allow custom name for state, to be able to make test runs with real tags
+        # without affecting production state
+        if args.override_state_name:
+            logger.info(
+                f"Using custom state name instead of release tag: {args.override_state_name}"
+            )
+            self.state_name = args.override_state_name
+        else:
+            self.state_name = self.tag
+
     def __enter__(self) -> "StateManager":
         if self.read_only:
             return self
         """Acquire lock when entering context."""
-        if not self.storage.acquire_lock(self.tag):
-            raise RuntimeError(f"Failed to acquire lock for tag: {self.tag}")
+        if not self.storage.acquire_lock(self.state_name):
+            raise RuntimeError(f"Failed to acquire lock for tag: {self.state_name}")
         self._lock_acquired = True
-        logger.info(f"Lock acquired for tag: {self.tag}")
+        logger.info(f"Lock acquired for tag: {self.state_name}")
         return self
 
     def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         if self.read_only:
             return
         """Release lock when exiting context."""
         if self._lock_acquired:
-            self.storage.release_lock(self.tag)
+            self.storage.release_lock(self.state_name)
             self._lock_acquired = False
-            logger.info(f"Lock released for tag: {self.tag}")
+            logger.info(f"Lock released for tag: {self.state_name}")
 
     @property
     def state(self) -> ReleaseState:
@@ -233,7 +243,7 @@ def apply_args(self, state: ReleaseState) -> None:
 
     def load(self) -> Optional[ReleaseState]:
         """Load state from storage backend."""
-        state_data = self.storage.get(self.tag)
+        state_data = self.storage.get(self.state_name)
         if state_data is None:
             return None
 
@@ -250,7 +260,7 @@ def sync(self) -> None:
         if current_dump != self.last_dump:
             self.last_dump = current_dump
             state_dict = json.loads(current_dump)
-            self.storage.put(self.tag, state_dict)
+            self.storage.put(self.state_name, state_dict)
             logger.debug("State saved")