Chore: [AEA-0000] - new script to setup repos

Makefile

@@ -13,6 +13,7 @@ deep-clean: clean
 
 test:
 	cd packages/get_repo_status && COVERAGE_FILE=coverage/.coverage poetry run python -m pytest
+	COVERAGE_FILE=packages/setup_github_repo/coverage/.coverage poetry run python -m pytest -c packages/setup_github_repo/pytest.ini packages/setup_github_repo
 
 lint: lint-black lint-flake8
 

packages/setup_github_repo/.coveragerc

@@ -0,0 +1,3 @@
+[run]
+data_file = packages/setup_github_repo/coverage/.coverage
+omit = */__init__.py

packages/setup_github_repo/__init__.py

@@ -0,0 +1 @@
+"""Utilities for setting up GitHub repositories for EPS projects."""

packages/setup_github_repo/__main__.py

@@ -0,0 +1,6 @@
+"""Module entrypoint for running setup_github_repo with python -m."""
+
+from .app.cli import main
+
+if __name__ == "__main__":
+    main()

packages/setup_github_repo/app/__init__.py

@@ -0,0 +1 @@
+"""Application modules for setup_github_repo."""

packages/setup_github_repo/app/aws_exports.py

@@ -0,0 +1,81 @@
+"""AWS CloudFormation export helpers used to resolve role values by environment."""
+
+from typing import Any
+
+import boto3
+
+from .models import Roles
+
+
+class AwsExportsService:
+    """Load CloudFormation exports and map them to the roles used by repo setup."""
+
+    def get_named_export(self, all_exports: list[dict[str, Any]], export_name: str, required: bool) -> str | None:
+        export_value = None
+
+        for export in all_exports:
+            if export["Name"] == export_name:
+                export_value = export["Value"]
+                break
+
+        if required and export_value is None:
+            raise ValueError(f"export {export_name} is required but not found")
+        return export_value
+
+    def get_all_exports(self, profile_name: str) -> list[dict[str, Any]]:
+        print(f"Getting exports for profile {profile_name}")
+        session = boto3.Session(profile_name=profile_name)
+        cloudformation_client = session.client("cloudformation")
+
+        all_exports: list[dict[str, Any]] = []
+        next_token = None
+
+        while True:
+            if next_token:
+                response = cloudformation_client.list_exports(NextToken=next_token)
+            else:
+                response = cloudformation_client.list_exports()
+
+            all_exports.extend(response.get("Exports", []))
+
+            next_token = response.get("NextToken")
+            if not next_token:
+                break
+        return all_exports
+
+    def get_role_exports(self, all_exports: list[dict[str, Any]]) -> Roles:
+        role_exports = [
+            {
+                "variable_name": "cloud_formation_deploy_role",
+                "export_name": "ci-resources:CloudFormationDeployRole",
+                "required": True,
+            },
+            {
+                "variable_name": "cloud_formation_check_version_role",
+                "export_name": "ci-resources:CloudFormationCheckVersionRole",
+                "required": True,
+            },
+            {
+                "variable_name": "cloud_formation_prepare_changeset_role",
+                "export_name": "ci-resources:CloudFormationPrepareChangesetRole",
+                "required": True,
+            },
+            {
+                "variable_name": "release_notes_execute_lambda_role",
+                "export_name": "ci-resources:ReleaseNotesExecuteLambdaRole",
+                "required": False,
+            },
+            {
+                "variable_name": "artillery_runner_role",
+                "export_name": "ci-resources:ArtilleryRunnerRole",
+                "required": False,
+            },
+        ]
+        all_roles: dict[str, str | None] = {}
+        for role_export in role_exports:
+            all_roles[role_export["variable_name"]] = self.get_named_export(
+                all_exports,
+                export_name=role_export["export_name"],
+                required=role_export["required"],
+            )
+        return Roles(**all_roles)

packages/setup_github_repo/app/cli.py

@@ -0,0 +1,111 @@
+"""CLI entrypoint that validates auth prerequisites and runs repository setup."""
+
+import argparse
+import subprocess
+
+from .constants import AWS_PROFILE_BY_ENV
+from .runner import SetupGithubRepoRunner
+
+
+def _read_gh_auth_token() -> str | None:
+    try:
+        result = subprocess.run(
+            ["gh", "auth", "token"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except FileNotFoundError as exc:
+        raise RuntimeError("GitHub CLI (gh) is not installed or not available on PATH.") from exc
+
+    if result.returncode != 0:
+        return None
+
+    token = result.stdout.strip()
+    if not token:
+        return None
+    return token
+
+
+def _get_or_create_gh_auth_token() -> str:
+    existing_token = _read_gh_auth_token()
+    if existing_token:
+        return existing_token
+
+    print("No GitHub token found. Running gh auth login to obtain one...")
+    subprocess.run(["gh", "auth", "login"], check=True)
+
+    token_after_login = _read_gh_auth_token()
+    if token_after_login:
+        return token_after_login
+
+    raise RuntimeError("Unable to retrieve GitHub token after running 'gh auth login'.")
+
+
+def resolve_gh_auth_token(explicit_token: str | None) -> str:
+    if explicit_token:
+        return explicit_token
+    return _get_or_create_gh_auth_token()
+
+
+def _has_valid_aws_credentials_for_profile(profile_name: str) -> bool:
+    try:
+        result = subprocess.run(
+            ["aws", "sts", "get-caller-identity", "--profile", profile_name],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except FileNotFoundError as exc:
+        raise RuntimeError("AWS CLI (aws) is not installed or not available on PATH.") from exc
+
+    return result.returncode == 0
+
+
+def _get_invalid_aws_profiles() -> list[str]:
+    required_profiles = sorted(set(AWS_PROFILE_BY_ENV.values()))
+    return [
+        profile_name for profile_name in required_profiles if not _has_valid_aws_credentials_for_profile(profile_name)
+    ]
+
+
+def ensure_aws_credentials() -> None:
+    invalid_profiles = _get_invalid_aws_profiles()
+    if not invalid_profiles:
+        return
+
+    invalid_profiles_text = ", ".join(invalid_profiles)
+    print(
+        f"AWS credentials missing or expired for profiles: {invalid_profiles_text}. "
+        "Running make aws-login to refresh credentials..."
+    )
+    try:
+        subprocess.run(["make", "aws-login"], check=True)
+    except FileNotFoundError as exc:
+        raise RuntimeError("make is not installed or not available on PATH.") from exc
+
+    remaining_invalid_profiles = _get_invalid_aws_profiles()
+    if remaining_invalid_profiles:
+        remaining_profiles_text = ", ".join(remaining_invalid_profiles)
+        raise RuntimeError(
+            "AWS credentials are still missing or expired after running make aws-login for profiles: "
+            f"{remaining_profiles_text}"
+        )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--gh_auth_token",
+        required=False,
+        help=(
+            "Please provide a github auth token. If authenticated with github cli this can be "
+            "retrieved using 'gh auth token'. If omitted, this script will try to retrieve one automatically."
+        ),
+    )
+
+    arguments = parser.parse_args()
+    ensure_aws_credentials()
+    github_auth_token = resolve_gh_auth_token(arguments.gh_auth_token)
+    runner = SetupGithubRepoRunner(gh_auth_token=github_auth_token)
+    runner.run()

packages/setup_github_repo/app/constants.py

@@ -0,0 +1,31 @@
+"""Static constants for AWS profiles, app IDs, and target service hostnames."""
+
+AWS_PROFILE_BY_ENV: dict[str, str] = {
+    "dev": "prescription-dev",
+    "qa": "prescription-qa",
+    "ref": "prescription-ref",
+    "int": "prescription-int",
+    "prod": "prescription-prod-readonly",
+    "recovery": "prescription-recovery",
+}
+
+AUTOMERGE_APP_ID = "420347"
+CREATE_PULL_REQUEST_APP_ID = "3182106"
+
+TARGET_SPINE_SERVERS: dict[str, str] = {
+    "dev": "msg.veit07.devspineservices.nhs.uk",
+    "int": "msg.intspineservices.nhs.uk",
+    "prod": "prescriptions.spineservices.nhs.uk",
+    "qa": "msg.intspineservices.nhs.uk",
+    "ref": "prescriptions.refspineservices.nhs.uk",
+    "recovery": "msg.veit07.devspineservices.nhs.uk",
+}
+
+TARGET_SERVICE_SEARCH_SERVERS: dict[str, str] = {
+    "dev": "int.api.service.nhs.uk",
+    "int": "api.service.nhs.uk",
+    "prod": "api.service.nhs.uk",
+    "qa": "int.api.service.nhs.uk",
+    "ref": "api.service.nhs.uk",
+    "recovery": "api.service.nhs.uk",
+}

packages/setup_github_repo/app/github_access.py

@@ -0,0 +1,26 @@
+"""Repository access permission management for standard EPS GitHub teams."""
+
+from .github_base import GithubOperationBase
+from .models import RepoConfig
+
+
+class GithubAccessManager(GithubOperationBase):
+    """Manage repository team access permissions."""
+
+    def setup_access(self, repo_config: RepoConfig) -> None:
+        repo_url = repo_config.repoUrl
+        if not self._confirm_action(f"Setting access in repo {repo_url}. Do you want to continue? (y/N): "):
+            return
+
+        org = self._github.get_organization("NHSDigital")
+        repo = self._github.get_repo(repo_url)
+        team_permissions = [
+            (self._github_teams.eps_team, "Write_View_Dependabot_Alerts"),
+            (self._github_teams.eps_administrator_team, "admin"),
+        ]
+
+        for team_id, permission in team_permissions:
+            team = org.get_team(int(team_id))
+            print(f"Granting team {team.slug} access to repo {repo_url} with role {permission}")
+            team.update_team_repository(repo, permission)
+            self._sleep_for_rate_limit()

packages/setup_github_repo/app/github_base.py

@@ -0,0 +1,38 @@
+"""Shared utilities for prompting and API pacing in GitHub setup operations."""
+
+import time
+
+from github import Github
+
+from .models import GithubTeams
+
+
+class GithubOperationBase:
+    """Shared behavior for interactive prompts and API rate-limit pacing."""
+
+    def __init__(
+        self,
+        github: Github,
+        github_teams: GithubTeams,
+        interactive: bool = True,
+        rate_limit_delay_seconds: float = 1.0,
+    ):
+        self._github = github
+        self._github_teams = github_teams
+        self._interactive = interactive
+        self._rate_limit_delay_seconds = rate_limit_delay_seconds
+
+    def _confirm_action(self, prompt: str) -> bool:
+        if not self._interactive:
+            return True
+
+        response = input(prompt)
+        if response.lower() == "y":
+            print("Continuing...")
+            return True
+
+        print("Returning.")
+        return False
+
+    def _sleep_for_rate_limit(self) -> None:
+        time.sleep(self._rate_limit_delay_seconds)