Create charm tracks automatically during charmcraft release

cilib/ch.py

@@ -0,0 +1,96 @@
+import base64
+import logging
+import json
+import os
+import subprocess
+import re
+import urllib.request
+
+from typing import Optional
+
+log = logging.getLogger(__name__)
+
+
+def _charmcraft_auth_to_macaroon(charmcraft_auth: str) -> Optional[str]:
+    """Decode charmcraft auth into the macaroon."""
+    try:
+        bytes = base64.b64decode(charmcraft_auth.strip().encode())
+        return json.loads(bytes).get("v")
+    except (base64.binascii.Error, json.JSONDecodeError):
+        return None
+
+
+def _track_or_channel(channel: str):
+    """Get the track from a channel."""
+    return channel.split("/")[0] if "/" in channel else channel
+
+
+def macaroon():
+    """Get the charmhub macaroon."""
+    macaroon = os.getenv("CHARM_MACAROON", "")
+    if not macaroon and (charmcraft_auth := os.getenv("CHARMCRAFT_AUTH")):
+        macaroon = _charmcraft_auth_to_macaroon(charmcraft_auth)
+    if not macaroon:
+        out = subprocess.run(
+            ["charmcraft", "login", "--export", "/dev/fd/2"],
+            stderr=subprocess.PIPE,
+            text=True,
+            check=True,
+        )
+        macaroon = _charmcraft_auth_to_macaroon(out.stderr.splitlines()[-1])
+    if not macaroon:
+        raise ValueError("No charmhub macaroon found")
+    os.environ["CHARM_MACAROON"] = macaroon
+    return macaroon
+
+
+def request(url: str):
+    """Create a request with the appropriate macaroon."""
+    return urllib.request.Request(
+        url,
+        method="GET",
+        headers={
+            "Authorization": f"Macaroon {macaroon()}",
+            "Content-Type": "application/json",
+        },
+    )
+
+
+def info(charm: str):
+    """Get charm info."""
+    req = request(f"https://api.charmhub.io/v1/charm/{charm}")
+    with urllib.request.urlopen(req) as resp:
+        if 200 <= resp.status < 300:
+            log.debug(f"Got charm info for {charm}")
+            return json.loads(resp.read())
+    raise ValueError(f"Failed to get charm info for {charm}: {resp.status}")
+
+
+def create_track(charm: str, track_or_channel: str):
+    """Create a track for a charm."""
+    req = request(f"https://api.charmhub.io/v1/charm/{charm}/tracks")
+    req.method = "POST"
+    track = _track_or_channel(track_or_channel)
+    req.data = json.dumps([{"name": track}]).encode()
+    with urllib.request.urlopen(req) as resp:
+        if 200 <= resp.status < 300:
+            log.info(f"Track {track} created for charm {charm}")
+            return
+    raise ValueError(f"Failed to create track {track} for charm {charm}: {resp.read()}")
+
+
+def ensure_track(charm: str, track_or_channel: str):
+    """Ensure a track exists for a charm."""
+    charm_info = info(charm)
+    track = _track_or_channel(track_or_channel)
+    charm_tracks = [t["name"] for t in charm_info["metadata"]["tracks"]]
+    if track in charm_tracks:
+        log.info(f"Track {track} already exists for charm {charm}")
+        return
+    patterns = [t["pattern"] for t in charm_info["metadata"]["track-guardrails"]]
+    if not any(re.compile(f"^{pattern}$").match(track) for pattern in patterns):
+        raise ValueError(
+            f"Track {track} does not match any guardrails for charm {charm}"
+        )
+
+    return create_track(charm, track)

jobs/build-charms/builder_local.py

@@ -24,6 +24,7 @@
 import zipfile
 from pathlib import Path
 from collections import defaultdict
+from cilib.ch import ensure_track
 from cilib.github_api import Repository
 from enum import Enum, unique
 from sh.contrib import git
@@ -332,6 +333,8 @@ def _unpublished_revisions(self, charm_entity):
     def release(self, entity: str, artifact: "Artifact", to_channels: List[str]):
         self._echo(f"Releasing :: {entity:^35} :: to: {to_channels}")
         rev_args = f"--revision={artifact.rev}"
+        for channel in to_channels:
+            ensure_track(entity, channel)
         channel_args = [f"--channel={chan}" for chan in to_channels]
         resource_rev_args = [
             f"--resource={rsc.name}:{rsc.rev}" for rsc in artifact.resources

tests/unit/build_charms/test_charms.py

@@ -424,8 +424,19 @@ def test_build_entity_assemble_resources(
     charm_cmd.assert_not_called()
 
 
+@pytest.fixture
+def ensure_track(builder_local):
+    with patch.object(builder_local, "ensure_track") as mocked:
+        yield mocked
+
+
 def test_build_entity_promote(
-    charm_environment, charm_cmd, charmcraft_cmd, tmpdir, builder_local
+    charm_environment,
+    charm_cmd,
+    charmcraft_cmd,
+    tmpdir,
+    builder_local,
+    ensure_track,
 ):
     """Test that BuildEntity releases to appropriate store."""
     charms = charm_environment.job_list
@@ -444,6 +455,10 @@ def test_build_entity_promote(
     ]
     charm_entity.release(artifact, to_channels=("latest/edge", "0.15/edge"))
     charm_cmd.release.assert_not_called()
+    ensure_track.assert_has_calls(
+        [call("k8s-ci-charm", "latest/edge"), call("k8s-ci-charm", "0.15/edge")]
+    )
+    ensure_track.reset_mock()
     charmcraft_cmd.release.assert_called_once_with(
         "k8s-ci-charm",
         "--revision=6",
@@ -456,6 +471,10 @@ def test_build_entity_promote(
 
     charm_entity.release(artifact, to_channels=("latest/stable", "0.15/stable"))
     charm_cmd.release.assert_not_called()
+    ensure_track.assert_has_calls(
+        [call("k8s-ci-charm", "latest/stable"), call("k8s-ci-charm", "0.15/stable")]
+    )
+    ensure_track.reset_mock()
     charmcraft_cmd.release.assert_called_once_with(
         "k8s-ci-charm",
         "--revision=6",
@@ -465,8 +484,10 @@ def test_build_entity_promote(
         "--resource=test-image:4",
     )
     charmcraft_cmd.release.reset_mock()
+
     charm_entity.release(artifact, to_channels=("0.14/stable",))
     charm_cmd.release.assert_not_called()
+    ensure_track.assert_called_once_with("k8s-ci-charm", "0.14/stable")
     charmcraft_cmd.release.assert_called_once_with(
         "k8s-ci-charm",
         "--revision=6",