MGMT-19400 - create tag management flow to tag assisted installer components according to capbcoa versions

Author: omer-vishlitzky

.gitignore

@@ -25,3 +25,5 @@ go.work
 *.swp
 *.swo
 *~
+hack/__pycache__/*
+hack/test/__pycache__/*

hack/__init__.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+
+import os
+import sys
+import subprocess
+import logging
+from datetime import datetime
+
+script_dir = os.path.dirname(os.path.abspath(__file__))
+project_root = os.path.dirname(script_dir)
+if project_root not in sys.path:
+    sys.path.insert(0, project_root)
+
+
+from hack.release_candidates_repository import ReleaseCandidatesRepository
+from hack.shared_types import ReleaseCandidates, SnapshotInfo
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("test-runner")
+
+RELEASE_CANDIDATES_FILE = os.path.join(
+    os.path.dirname(os.path.dirname(__file__)), "release-candidates.yaml"
+)
+
+
+def find_pending_snapshot(data: ReleaseCandidates) -> tuple[str, SnapshotInfo] | None:
+    for snapshot in data["snapshots"]:
+        if "metadata" in snapshot and snapshot["metadata"].get("status") == "pending":
+            return snapshot["metadata"].get("id"), snapshot
+    return None, None
+
+
+def export_component_variables(snapshot: SnapshotInfo) -> None:
+    env_vars = {}
+    component_env_map = {
+        "kubernetes-sigs/cluster-api": "CAPI_VERSION",
+        "metal3-io/cluster-api-provider-metal3": "CAPM3_VERSION",
+        "openshift/assisted-service": "ASSISTED_SERVICE_IMAGE",
+        "openshift/assisted-image-service": "ASSISTED_IMAGE_SERVICE_IMAGE",
+        "openshift/assisted-installer-agent": "ASSISTED_INSTALLER_AGENT_IMAGE",
+        "openshift/assisted-installer-controller": "ASSISTED_INSTALLER_CONTROLLER_IMAGE",
+        "openshift/assisted-installer": "ASSISTED_INSTALLER_IMAGE",
+    }
+    for component in snapshot["components"]:
+        repo = component["repository"]
+        repo_name: str = repo.removeprefix("https://github.com/")
+        image_url: str = component.get("image_url")
+        env_var = component_env_map.get(repo_name)
+        if not env_var:
+            logger.warning(
+                f"No environment variable mapping for component: {component}"
+            )
+            continue
+
+        if (
+            repo == "https://github.com/kubernetes-sigs/cluster-api"
+            or repo == "https://github.com/metal3-io/cluster-api-provider-metal3"
+        ):
+            value = component["ref"]
+        else:
+            value = component["image_url"]
+
+        if repo_name == "openshift/assisted-installer":
+            if "controller" in image_url.lower():
+                env_var = "ASSISTED_INSTALLER_CONTROLLER_IMAGE"
+            else:
+                env_var = "ASSISTED_INSTALLER_IMAGE"
+
+        if value:
+            env_vars[env_var] = value
+            os.environ[env_var] = value
+            logger.info(f"Exported {env_var}={value}")
+
+
+def run_ansible_tests() -> bool:
+    try:
+        logger.info("Running Ansible tests...")
+        ansible_cmd = [
+            "ansible-playbook",
+            "test/ansible/run_test.yaml",
+            "-i",
+            "test/ansible/inventory.yaml",
+        ]
+        result = subprocess.run(
+            ansible_cmd, check=False, capture_output=False, text=True
+        )
+
+        if result.returncode == 0:
+            logger.info("Ansible tests passed successfully")
+            return True
+        else:
+            logger.error(f"Ansible tests failed with return code {result.returncode}")
+            return False
+
+    except Exception as e:
+        logger.error(f"Error running Ansible tests: {e}")
+        return False
+
+
+def update_snapshot_status(snapshot_id: str, success: bool) -> None:
+    try:
+        repo = ReleaseCandidatesRepository(RELEASE_CANDIDATES_FILE)
+        updated = repo.update_status(
+            snapshot_id,
+            "successful" if success else "failed",
+            datetime.now().isoformat(),
+        )
+
+        if updated:
+            logger.info(
+                f"Updated snapshot {snapshot_id} status to {'successful' if success else 'failed'}"
+            )
+        else:
+            logger.error(f"Snapshot with ID {snapshot_id} not found")
+    except Exception as e:
+        logger.error(f"Error updating snapshot status: {e}")
+
+
+def main() -> None:
+    try:
+        logger.info("Starting test runner")
+        repo = ReleaseCandidatesRepository(RELEASE_CANDIDATES_FILE)
+        pending_candidates = repo.find_by_status("pending")
+
+        if not pending_candidates:
+            logger.info("No pending snapshots to process")
+            return
+
+        candidate = pending_candidates[0]
+        snapshot_id = candidate.metadata.get("id")
+
+        logger.info(f"Processing pending snapshot {snapshot_id}")
+        export_component_variables(candidate.to_dict())
+
+        success = run_ansible_tests()
+        update_snapshot_status(snapshot_id, success)
+
+        if success:
+            logger.info("Snapshot processed successfully")
+        else:
+            logger.error("Snapshot testing failed")
+            sys.exit(1)
+
+    except Exception as e:
+        logger.error(f"Test runner failed: {e}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

hack/ansible_test_runner.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+
+import os
+import logging
+from github import Github, GithubIntegration
+
+logger = logging.getLogger("github-auth")
+
+def get_github_client():
+    """
+    Creates a PyGithub client using GitHub App authentication
+    
+    Environment variables:
+        GITHUB_APP_ID: The ID of your GitHub App
+        GITHUB_APP_INSTALLATION_ID: The installation ID for your GitHub App
+        GITHUB_APP_PRIVATE_KEY: Private key .pem file
+    """
+    try:
+        app_id = os.environ["GITHUB_APP_ID"]
+        installation_id = os.environ["GITHUB_APP_INSTALLATION_ID"]
+        private_key = os.environ["GITHUB_APP_PRIVATE_KEY"]
+        
+        if not all(bool(x) for x in [app_id, installation_id, private_key]):
+            logger.error("Missing GitHub App credentials in environment")
+            raise ValueError("""
+                Missing GitHub App credentials.
+                Ensure GITHUB_APP_ID, GITHUB_APP_INSTALLATION_ID, and GITHUB_APP_PRIVATE_KEY are set.
+            """)
+            
+        integration = GithubIntegration(
+            int(app_id),
+            private_key
+        )
+        access_token = integration.get_access_token(int(installation_id))
+        github_client = Github(access_token.token)
+        logger.debug("Created GitHub client with new token")
+        return github_client
+    except Exception as e:
+        logger.error(f"Error creating GitHub client: {e}")
+        raise

hack/github_auth.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+
+import logging
+import os
+import sys
+import uuid
+from datetime import datetime
+
+from ruamel.yaml import YAML
+
+script_dir = os.path.dirname(os.path.abspath(__file__))
+project_root = os.path.dirname(script_dir)
+if project_root not in sys.path:
+    sys.path.insert(0, project_root)
+
+from hack.shared_types import ReleaseCandidates, RepositoryInfo, SnapshotInfo
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("release-repository")
+
+# Configure YAML parser
+yaml = YAML(typ="rt")
+yaml.default_flow_style = False
+yaml.explicit_start = False
+yaml.preserve_quotes = True
+yaml.width = 4096
+yaml.indent(mapping=2, sequence=2, offset=0)
+
+
+class ReleaseCandidate:
+
+    def __init__(
+        self,
+        components: list[RepositoryInfo],
+        generated_at: str = None,
+        status: str = "pending",
+        tested_at: str = None,
+        id: str = None,
+    ):
+        self.metadata = {
+            "generated_at": generated_at or datetime.now().isoformat(),
+            "status": status,
+        }
+
+        if tested_at:
+            self.metadata["tested_at"] = tested_at
+
+        if id:
+            self.metadata["id"] = id
+        else:
+            self.metadata["id"] = str(uuid.uuid4())
+
+        self.components = components
+
+    def to_dict(self) -> SnapshotInfo:
+        return {"metadata": self.metadata, "components": self.components}
+
+    @classmethod
+    def from_dict(cls, data: SnapshotInfo) -> "ReleaseCandidate":
+        metadata = data.get("metadata", {})
+        components = data.get("components", [])
+
+        return cls(
+            components=components,
+            generated_at=metadata.get("generated_at"),
+            status=metadata.get("status", "pending"),
+            tested_at=metadata.get("tested_at"),
+            id=metadata.get("id"),
+        )
+
+    def equals(self, other: "ReleaseCandidate") -> bool:
+        set1 = {
+            (d.get("repository", ""), d.get("ref", ""), d.get("image_url", ""))
+            for d in self.components
+        }
+        set2 = {
+            (d.get("repository", ""), d.get("ref", ""), d.get("image_url", ""))
+            for d in other.components
+        }
+        return set1 == set2
+
+
+class ReleaseCandidatesRepository:
+
+    def __init__(self, file_path: str = "release-candidates.yaml"):
+        self.file_path = file_path
+
+    def as_list(self) -> list[ReleaseCandidate]:
+        data = self._read_file()
+        snapshots = data.get("snapshots", [])
+
+        return [ReleaseCandidate.from_dict(snapshot) for snapshot in snapshots]
+
+    def find_by_status(self, status: str) -> list[ReleaseCandidate]:
+        candidates = self.as_list()
+        return [rc for rc in candidates if rc.metadata.get("status") == status]
+
+    def find_by_id(self, id: str) -> ReleaseCandidate | None:
+        candidates = self.as_list()
+        for candidate in candidates:
+            if candidate.metadata.get("id") == id:
+                return candidate
+        return None
+
+    def save(self, candidate: ReleaseCandidate) -> bool:
+        data = self._read_file()
+
+        candidate_dict = candidate.to_dict()
+
+        candidates = data.get("snapshots", [])
+        for i, existing in enumerate(candidates):
+            existing_id = existing.get("metadata", {}).get("id")
+            if existing_id and existing_id == candidate.metadata.get("id"):
+                candidates[i] = candidate_dict
+                data["snapshots"] = candidates
+                self._write_file(data)
+                return True
+
+        if "snapshots" not in data:
+            data["snapshots"] = []
+
+        new_rc = ReleaseCandidate.from_dict(candidate_dict)
+        for existing in self.as_list():
+            if new_rc.equals(existing):
+                logger.info("Similar release candidate already exists, skipping")
+                return False
+
+        data["snapshots"].insert(0, candidate_dict)
+        self._write_file(data)
+        return True
+
+    def update_status(self, id: str, status: str, tested_at: str = None) -> bool:
+        candidate = self.find_by_id(id)
+        if not candidate:
+            return False
+
+        candidate.metadata["status"] = status
+        if tested_at or status in ["successful", "failed"]:
+            candidate.metadata["tested_at"] = tested_at or datetime.now().isoformat()
+
+        return self.save(candidate)
+
+    def _read_file(self) -> ReleaseCandidates:
+        if not os.path.exists(self.file_path):
+            return {"snapshots": []}
+
+        try:
+            with open(self.file_path, "r") as f:
+                data = yaml.load(f)
+                return data or {"snapshots": []}
+        except Exception as e:
+            logger.error(f"Error reading {self.file_path}: {e}")
+            return {"snapshots": []}
+
+    def _write_file(self, data: ReleaseCandidates) -> None:
+        try:
+            with open(self.file_path, "w") as f:
+                yaml.dump(data, f)
+        except Exception as e:
+            logger.error(f"Error writing to {self.file_path}: {e}")
+            raise

hack/release_candidates_repository.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+
+from typing import TypedDict
+
+class RepositoryInfo(TypedDict):
+    repository: str
+    ref: str
+    image_url: str | None
+
+class SnapshotMetadata(TypedDict):
+    generated_at: str
+    status: str
+    tested_at: str
+    id: str
+
+class SnapshotInfo(TypedDict):
+    metadata: SnapshotMetadata
+    components: list[RepositoryInfo]
+
+class ReleaseCandidates(TypedDict):
+    snapshots: list[SnapshotInfo]
+
+class VersionEntry(TypedDict):
+    name: str
+    components: list[RepositoryInfo]
+
+class VersionsFile(TypedDict):
+    versions: list[VersionEntry]
+
+class RepositoryConfig(TypedDict, total=False):
+    images: list[str] | None
+    version_prefix: str