fix: adjust the dry-run behavior and file parsing (#130)

Author: misohu

scripts/README.md

@@ -0,0 +1,37 @@
+# Scripts
+
+## `charms_promotions.py`
+
+Promotes charm revisions from a Juju status export to a higher risk channel (`beta` → `candidate` → `stable`).
+
+### What it reads
+
+- `--format text`: a `juju status` text output (full output is fine, not just the apps table)
+- `--format yaml`: a status YAML with `applications.*.charm-name/charm-rev/charm-channel`
+
+### Quick usage
+
+From the `scripts/` directory:
+
+```bash
+python charms_promotions.py --file status.txt --format text --promote-to candidate --dry-run
+```
+
+Run for real:
+
+```bash
+python charms_promotions.py --file status.txt --format text --promote-to candidate --apply
+```
+
+Exclude charms (space-separated):
+
+```bash
+python charms_promotions.py --file status.txt --format text --promote-to candidate --dry-run --exclude mysql-k8s
+```
+
+### Notes
+
+- Default mode is dry-run: it prints `charmcraft release ...` commands.
+- Dry-run still calls `charmcraft status` to resolve exact revisions/resources.
+- If you do not have access to a charm package, the script prints a warning and skips it.
+- Passing `--exclude` replaces the default exclude list.

scripts/charms_promotions.py

@@ -1,12 +1,13 @@
 import json
 import re
+import shlex
 import subprocess
+import sys
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Optional
 
 import yaml
-from IPython.core.oinspect import Bundle
 
 RISKS = {"edge": 1, "beta": 2, "candidate": 3, "stable": 4}
 
@@ -29,7 +30,8 @@ def get_status(self, channel: Optional[str] = None, revision: Optional[int] = No
 
         if not self._status:
             output = subprocess.check_output(
-                ["charmcraft", "status", self.name, "--format", "json"]
+                ["charmcraft", "status", self.name, "--format", "json"],
+                stderr=subprocess.STDOUT,
             )
 
             self._status = json.loads(output.decode("utf-8"))
@@ -80,7 +82,7 @@ def promote_version(self, risk: str, dry_run: bool = True):
         )
 
         if dry_run:
-            return cmds
+            return shlex.join(cmds)
 
         return subprocess.check_output(cmds).decode("utf-8")
 
@@ -97,13 +99,14 @@ class Bundle:
     charms: list[Charm]
 
     @classmethod
-    def from_status(cls, content: str, format: Format = "text"):
+    def from_status(cls, content: str, format: Format | str = Format.TEXT):
         parsers = {
             Format.TEXT: TextParser,
             Format.YAML: YAMLParser
         }
+        normalized_format = Format(format)
 
-        return parsers[format].parse(content)
+        return parsers[normalized_format].parse(content)
 
 
 class YAMLParser:
@@ -119,7 +122,7 @@ def parse(content: str):
 
 
 class TextParser:
-    word_with_leading_spaces = re.compile("^\s*[^\s]+")
+    word_with_leading_spaces = re.compile(r"^\s*[^\s]+")
 
     @staticmethod
     def extract_first_word(mystring):
@@ -134,13 +137,35 @@ def parse_line(line, indices):
 
     @staticmethod
     def parse(content: str):
+        lines = content.splitlines()
+
+        white_spaces = re.compile(r"\s+\s+")
+
+        app_header_index = next(
+            (
+                index
+                for index, line in enumerate(lines)
+                if line.strip().startswith("App")
+                and "Charm" in line
+                and "Channel" in line
+            ),
+            None,
+        )
+
+        if app_header_index is None:
+            raise ValueError("Could not locate applications table in status text")
 
-        lines = content.split("\n")
+        table_lines = []
+        for line in lines[app_header_index:]:
+            if table_lines and (not line.strip() or line.strip().startswith("Unit")):
+                break
+            table_lines.append(line)
 
-        white_spaces = re.compile("\s+\s+")
+        if len(table_lines) < 2:
+            raise ValueError("Applications table is empty in status text")
 
         # Get header
-        header = lines[0]
+        header = table_lines[0]
 
         # First guess of width based on headers
         ends=[s.end() for s in white_spaces.finditer(header)]
@@ -153,7 +178,7 @@ def parse(content: str):
         # This is due to the fact that some columns the text extends to before the start
         # of the columns header (text aligned right)
         widths = [len(column) for column in columns]
-        for line in lines[1:]:
+        for line in table_lines[1:]:
 
             widths = list(map(max,zip(
                 widths,
@@ -167,10 +192,10 @@ def parse(content: str):
 
         data = [
             dict(zip(columns, TextParser.parse_line(line, indices)))
-            for line in lines[1:]
+            for line in table_lines[1:]
+            if line.strip()
         ]
 
-        # pd.DataFrame([TextParser.parse_line(line, indices) for line in lines[1:]], columns= columns)
         return Bundle([
             Charm(item["Charm"], int(item["Rev"]), item["Channel"])
             for item in data
@@ -180,18 +205,25 @@ def parse(content: str):
 
 if __name__ == "__main__":
 
-    # with open("./scripts/status.txt") as fid:
-    #     bundle = Bundle.from_status(fid.read(), Format.TEXT)
-
-    # with open("./scripts/status.yaml") as fid:
-    #     bundle = Bundle.from_status(fid.read(), Format.YAML)
-
     import argparse
 
     parser = argparse.ArgumentParser()
 
     parser.add_argument("--file", required=True)
-    parser.add_argument("--apply", default=False, action="store_true")
+    action_group = parser.add_mutually_exclusive_group()
+    action_group.add_argument(
+        "--dry-run",
+        dest="dry_run",
+        action="store_true",
+        help="Print release commands without executing them (default)",
+    )
+    action_group.add_argument(
+        "--apply",
+        dest="dry_run",
+        action="store_false",
+        help="Execute charmcraft release commands",
+    )
+    parser.set_defaults(dry_run=True)
     parser.add_argument("--format", choices=("text", "yaml"), default="text")
     parser.add_argument("--promote-to", choices=("beta", "candidate", "stable"), default="beta")
     parser.add_argument("--exclude", nargs="*", default=["mysql-k8s"])
@@ -202,15 +234,17 @@ def parse(content: str):
 
     for charm in bundle.charms:
         if not charm.name in args.exclude:
-            print(charm.promote_version(args.promote_to, not args.apply))
-
-
-
-
-
-
-
-
-
-
-    
+            try:
+                print(charm.promote_version(args.promote_to, args.dry_run))
+            except subprocess.CalledProcessError as err:
+                output = (err.output or b"").decode("utf-8", errors="replace").strip()
+                if "permission-required" in output:
+                    print(
+                        f"WARNING: skipping '{charm.name}' due to missing permissions. Add it to --exclude to avoid this warning.",
+                        file=sys.stderr,
+                    )
+                else:
+                    print(
+                        f"WARNING: skipping '{charm.name}' after command failure: {shlex.join(err.cmd)}",
+                        file=sys.stderr,
+                    )