Feat/migrate git logic to python

.github/workflows/action_integration_test.yml

@@ -44,9 +44,10 @@ jobs:
         if: always()
         uses: dawidd6/action-delete-branch@v3
         with:
-          github_token: ${{github.token}}
           branches: ${{ steps.update.outputs.target_branch }}
           soft_fail: true
+        env:
+          GITHUB_TOKEN: ${{github.token}}
 
 
 

.vscode/settings.json

@@ -16,7 +16,7 @@
     },
     "editor.defaultFormatter": "charliermarsh.ruff"
   },
-  "files.autoSave": "afterDelay",
+  "files.autoSave": "onFocusChange",
   "python.analysis.inlayHints.functionReturnTypes": true,
   "python.analysis.inlayHints.pytestParameters": true,
   "python.analysis.inlayHints.variableTypes": true,
@@ -46,4 +46,7 @@
   ],
   "python.testing.unittestEnabled": false,
   "python.testing.pytestEnabled": true,
+  "files.exclude": {
+    "**/__pycache__": true
+  }
 }

.vscode/tasks.json

@@ -35,6 +35,19 @@
                 "reveal": "always"
             },
             "problemMatcher": []
+        },
+        {
+            "label": "run tests (pytest)",
+            "type": "shell",
+            "command": "pytest",
+            "presentation": {
+                "reveal": "always"
+            },
+            "problemMatcher": [],
+            "group": {
+                "kind": "test",
+                "isDefault": true
+            }
         }
     ]
 }

README.md

@@ -98,7 +98,8 @@ jobs:
       - name: Run in update mode
         uses: Noahnc/infrapatch@main
         with:
-          report_only: false 
+          report_only: false
+
 ```
 
 #### Report only Mode
@@ -114,10 +115,20 @@ If you use private registries in your Terraform project, you can specify credent
   - name: Run in update mode
     uses: Noahnc/infrapatch@main
     with:
-      report_only: false
       registry_secrets: |
         spacelift.io=${{ secrets.SPACELIFT_API_TOKEN }}
         <second_registry>=<registry_token>
 ```
 
 Each secret must be specified in a new line with the following format: `<registry_name>=<registry_token>`
+
+#### Working Directory
+
+By default, the Action will run in the root directory of the repository. If you want to only scan a subdirectory, you can specify a subdirectory with the `working_directory_relative` input:
+
+```yaml
+  - name: Run in update mode
+    uses: Noahnc/infrapatch@main
+    with:
+      working_directory: "path/to/terraform/code"
+```

action.yml

@@ -22,21 +22,20 @@ inputs:
     description: "Git email to use for commits. Defaults to [email protected]"
     required: false
     default: "[email protected]"
-  github_token:
-    description: "GitHub access token. Defaults to github.token."
-    default: ${{ github.token }}
   report_only:
     description: "Only report new versions. Do not update files. Defaults to false"
     default: "false"
     required: true
   registry_secrets:
-    description: "Registry secrets to use for private registries"
+    description: "Registry secrets to use for private registries. Needs to be a newline separated list of secrets in the format <registry_domain>:<secret_name>. Defaults to empty"
     required: false
     default: ""
-  working_directory:
-    description: "Working directory to run the command in. Defaults to the root of the repository"
+  working_directory_relative:
+    description: "Working directory to run the action in. Defaults to the root of the repository"
     required: false
-    default: ${{ github.workspace }}
+  github_token:
+    description: "GitHub access token. Defaults to github.token."
+    default: ${{ github.token }}
 outputs:
   target_branch:
     description: "Name of the branch where changes will be pushed to"
@@ -70,39 +69,13 @@ runs:
         pip install -r requirements.txt
       shell: bash
 
-    - name: Create target branch
-      id: create_branch
-      if: ${{ inputs.report_only == 'false' }}
-      uses: peterjgrainger/[email protected]
-      env:
-        GITHUB_TOKEN: ${{ inputs.github_token }}
-      with:
-        branch: "refs/heads/${{ inputs.target_branch_name }}"
-
     - name: Configure git
       if: ${{ inputs.report_only }} == 'false' }}
-      working-directory: ${{ inputs.working_directory }}
       shell: bash
       run: |
         git config --global user.name "${{ inputs.git_user }}"
         git config --global user.email "${{ inputs.git_email }}"
 
-    - name: Switch to target branch
-      if: ${{ inputs.report_only == 'false' }}
-      working-directory: ${{ inputs.working_directory }}
-      shell: bash
-      run: |
-        git fetch origin
-        git checkout -b "${{ steps.branch.outputs.target }}" "${{ steps.branch.outputs.target_origin }}"
-
-    - name: Rebase target branch
-      if: ${{ steps.create_branch.outputs.created == 'false' }}
-      working-directory: ${{ inputs.working_directory }}
-      shell: bash
-      run: |
-        echo "Rebasing ${{ steps.branch.outputs.target }} on ${{ steps.branch.outputs.head_origin }}"
-        git rebase -Xtheirs ${{ steps.branch.outputs.head_origin }}
-
     - name: Run InfraPatch Action
       shell: bash
       run: |
@@ -111,17 +84,17 @@ runs:
         if [ "${{ runner.debug }}" == "1" ]; then
             arguments+=("--debug")
         fi
-        if [ "${{ inputs.report_only }}" == "true" ]; then
-            arguments+=("--report-only")
-        fi
-        if [ "${{ inputs.registry_secrets }}" != "" ]; then
-            arguments+=("--registry-secrets-string" "\"${{ inputs.registry_secrets }}\"")
-        fi
-        arguments+=("--github-token" "${{ inputs.github_token }}")
-        arguments+=("--head-branch" "${{ steps.branch.outputs.head }}")
-        arguments+=("--target-branch" "${{ steps.branch.outputs.target }}")
-        arguments+=("--repository-name" "${{ inputs.repository_name }}")
-        arguments+=("--working-directory" "${{ inputs.working_directory }}")
-        arguments+=("--default-registry-domain" "${{ inputs.default_registry_domain }}")
         python -m "$module" "${arguments[@]}"
+      env:
+        # Config from inputs
+        GITHUB_TOKEN: ${{ inputs.github_token }}
+        DEFAULT_REGISTRY_DOMAIN: ${{ inputs.DEFAULT_REGISTRY_DOMAIN }}
+        REPOSITORY_NAME: ${{ inputs.repository_name }}
+        REPORT_ONLY: ${{ inputs.report_only }}
+        REGISTRY_SECRET_STRING: ${{ inputs.REGISTRY_SECRETS }}
+        WORKING_DIRECTORY_RELATIVE: ${{ inputs.working_directory_relative }}
+
+        # Calculated config from other steps
+        HEAD_BRANCH: ${{ steps.branch.outputs.head }}
+        TARGET_BRANCH: ${{ steps.branch.outputs.target }}
 

infrapatch/action/__main__.py

@@ -1,82 +1,71 @@
 import logging as log
-import subprocess
-from pathlib import Path
+
 
 import click
-from github import Auth, Github
+from github import Auth, Github, GithubException
 from github.PullRequest import PullRequest
+from infrapatch.action.config import ActionConfigProvider
 
 from infrapatch.core.composition import build_main_handler
 from infrapatch.core.log_helper import catch_exception, setup_logging
 from infrapatch.core.models.versioned_terraform_resources import get_upgradable_resources
+from infrapatch.core.utils.git import Git
 
 
 @click.group(invoke_without_command=True)
 @click.option("--debug", is_flag=True)
-@click.option("--default-registry-domain")
-@click.option("--registry-secrets-string", default=None)
-@click.option("--github-token")
-@click.option("--target-branch")
-@click.option("--head-branch")
-@click.option("--repository-name")
-@click.option("--report-only", is_flag=True)
-@click.option("--working-directory")
 @catch_exception(handle=Exception)
-def main(
-    debug: bool,
-    default_registry_domain: str,
-    registry_secrets_string: str,
-    github_token: str,
-    target_branch: str,
-    head_branch: str,
-    repository_name: str,
-    report_only: bool,
-    working_directory: Path,
-):
+def main(debug: bool):
     setup_logging(debug)
-    log.debug(
-        f"Running infrapatch with the following parameters: "
-        f"default_registry_domain={default_registry_domain}, "
-        f"registry_secrets_string={registry_secrets_string}, "
-        f"github_token={github_token}, "
-        f"report_only={report_only}, "
-        f"working_directory={working_directory}"
-    )
-    credentials = {}
-    working_directory = Path(working_directory)
-    if registry_secrets_string is not None:
-        credentials = get_credentials_from_string(registry_secrets_string)
-    main_handler = build_main_handler(default_registry_domain=default_registry_domain, credentials_dict=credentials)
-    resources = main_handler.get_all_terraform_resources(working_directory)
-
-    if report_only:
+
+    config = ActionConfigProvider()
+
+    git = Git(config.repository_root)
+    github = Github(auth=Auth.Token(config.github_token))
+    github_repo = github.get_repo(config.repository_name)
+    github_head_branch = github_repo.get_branch(config.head_branch)
+
+    main_handler = build_main_handler(default_registry_domain=config.default_registry_domain, credentials_dict=config.registry_secrets)
+
+    git.fetch_origin()
+
+    try:
+        github_target_branch = github_repo.get_branch(config.target_branch)
+    except GithubException:
+        github_target_branch = None
+
+    if github_target_branch is not None and config.report_only is False:
+        log.info(f"Branch {config.target_branch} already exists. Checking out...")
+        git.checkout_branch(config.target_branch, f"origin/{config.target_branch}")
+
+        log.info(f"Rebasing branch {config.target_branch} onto origin/{config.head_branch}")
+        git.run_git_command(["rebase", "-Xtheirs", f"origin/{config.head_branch}"])
+        git.push(["-f", "-u", "origin", config.target_branch])
+
+    resources = main_handler.get_all_terraform_resources(config.working_directory)
+
+    if config.report_only:
         main_handler.print_resource_table(resources)
         log.info("Report only mode is enabled. No changes will be applied.")
         return
 
     upgradable_resources = get_upgradable_resources(resources)
+
     if len(upgradable_resources) == 0:
         log.info("No upgradable resources found.")
         return
 
-    main_handler.update_resources(upgradable_resources, True, working_directory, True)
-    main_handler.dump_statistics(upgradable_resources, save_as_json_file=True)
-
-    push_changes(target_branch, working_directory)
+    if github_target_branch is None:
+        log.info(f"Branch {config.target_branch} does not exist. Creating and checking out...")
+        github_repo.create_git_ref(ref=f"refs/heads/{config.target_branch}", sha=github_head_branch.commit.sha)
+        git.checkout_branch(config.target_branch, f"origin/{config.head_branch}")
 
-    create_pr(github_token, head_branch, repository_name, target_branch)
+    main_handler.update_resources(upgradable_resources, True, config.working_directory, config.repository_root, True)
+    main_handler.dump_statistics(upgradable_resources, save_as_json_file=True)
 
+    git.push(["-f", "-u", "origin", config.target_branch])
 
-def push_changes(target_branch, working_directory):
-    command = ["git", "push", "-f", "-u", "origin", target_branch]
-    log.debug(f"Executing command: {' '.join(command)}")
-    try:
-        result = subprocess.run(command, capture_output=True, text=True, cwd=working_directory.absolute().as_posix())
-    except Exception as e:
-        raise Exception(f"Error pushing to remote: {e}")
-    if result.returncode != 0:
-        log.error(f"Stdout: {result.stdout}")
-        raise Exception(f"Error pushing to remote: {result.stderr}")
+    create_pr(config.github_token, config.head_branch, config.repository_name, config.target_branch)
 
 
 def create_pr(github_token, head_branch, repository_name, target_branch) -> PullRequest:
@@ -91,20 +80,5 @@ def create_pr(github_token, head_branch, repository_name, target_branch) -> Pull
     return repo.create_pull(title="InfraPatch Module and Provider Update", body="InfraPatch Module and Provider Update", base=head_branch, head=target_branch)
 
 
-def get_credentials_from_string(credentials_string: str) -> dict:
-    credentials = {}
-    if credentials_string == "":
-        return credentials
-    for line in credentials_string.splitlines():
-        try:
-            name, token = line.split("=", 1)
-        except ValueError as e:
-            log.debug(f"Secrets line '{line}' could not be split into name and token.")
-            raise Exception(f"Error processing secrets: '{e}'")
-        # add the name and token to the credentials dict
-        credentials[name] = token
-    return credentials
-
-
 if __name__ == "__main__":
     main()

infrapatch/action/config.py

@@ -0,0 +1,63 @@
+import os
+from pathlib import Path
+from typing import Any
+import logging as log
+
+
+class MissingConfigException(Exception):
+    pass
+
+
+class ActionConfigProvider:
+    github_token: str
+    head_branch: str
+    target_branch: str
+    repository_name: str
+    default_registry_domain: str
+    working_directory: Path
+    repository_root: Path
+    report_only: bool
+    registry_secrets: dict[str, str]
+
+    def __init__(self) -> None:
+        self.github_token = _get_value_from_env("GITHUB_TOKEN", secret=True)
+        self.head_branch = _get_value_from_env("HEAD_BRANCH")
+        self.target_branch = _get_value_from_env("TARGET_BRANCH")
+        self.repository_name = _get_value_from_env("REPOSITORY_NAME")
+        self.repository_root = Path(os.getcwd())
+        self.working_directory = self.repository_root.joinpath(_get_value_from_env("WORKING_DIRECTORY_RELATIVE", default=""))
+        self.default_registry_domain = _get_value_from_env("DEFAULT_REGISTRY_DOMAIN")
+        self.registry_secrets = _get_credentials_from_string(_get_value_from_env("REGISTRY_SECRET_STRING", secret=True, default=""))
+        self.report_only = _from_env_to_bool(_get_value_from_env("REPORT_ONLY", default="False").lower())
+
+
+def _get_value_from_env(key: str, secret: bool = False, default: Any = None) -> Any:
+    if key in os.environ:
+        log_value = os.environ[key]
+        if secret:
+            log_value = f"{log_value[:3]}***"
+        log.debug(f"Found the following value for {key}: {log_value}")
+        return os.environ[key]
+    if default is not None:
+        log.debug(f"Using default value for {key}: {default}")
+        return default
+    raise MissingConfigException(f"Missing configuration for key: {key}")
+
+
+def _get_credentials_from_string(credentials_string: str) -> dict[str, str]:
+    credentials = {}
+    if credentials_string == "":
+        return credentials
+    for line in credentials_string.splitlines():
+        try:
+            name, token = line.split("=", 1)
+        except ValueError as e:
+            log.debug(f"Secrets line '{line}' could not be split into name and token.")
+            raise Exception(f"Error processing secrets: '{e}'")
+        # add the name and token to the credentials dict
+        credentials[name] = token
+    return credentials
+
+
+def _from_env_to_bool(value: str) -> bool:
+    return value.lower() in ["true", "1", "yes", "y", "t"]