diff --git a/.github/workflows/action_integration_test.yml b/.github/workflows/action_integration_test.yml index 08563dc..83c5ea8 100644 --- a/.github/workflows/action_integration_test.yml +++ b/.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}} diff --git a/.vscode/settings.json b/.vscode/settings.json index 14ad595..50d96e1 100644 --- a/.vscode/settings.json +++ b/.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 + } } \ No newline at end of file diff --git a/.vscode/tasks.json b/.vscode/tasks.json index 7bf227b..074dde1 100644 --- a/.vscode/tasks.json +++ b/.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 + } } ] } \ No newline at end of file diff --git a/README.md b/README.md index 114b385..876d224 100644 --- a/README.md +++ b/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 }} = ``` Each secret must be specified in a new line with the following format: `=` + +#### 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" +``` diff --git a/action.yml b/action.yml index f43adb2..86e03f0 100644 --- a/action.yml +++ b/action.yml @@ -22,21 +22,20 @@ inputs: description: "Git email to use for commits. Defaults to bot@infrapatch.ch" required: false default: "bot@infrapatch.ch" - 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 :. 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/action-create-branch@v2.2.0 - 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 }} diff --git a/infrapatch/action/__main__.py b/infrapatch/action/__main__.py index 27f0bd6..4954126 100644 --- a/infrapatch/action/__main__.py +++ b/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() diff --git a/infrapatch/action/config.py b/infrapatch/action/config.py new file mode 100644 index 0000000..4f589de --- /dev/null +++ b/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"] diff --git a/infrapatch/action/tests/test___main__.py b/infrapatch/action/tests/test___main__.py deleted file mode 100644 index 7a0c0ff..0000000 --- a/infrapatch/action/tests/test___main__.py +++ /dev/null @@ -1,25 +0,0 @@ -from infrapatch.action.__main__ import get_credentials_from_string - - -def test_get_credentials_from_string(): - # Test case 1: Empty credentials string - credentials_string = "" - expected_result = {} - assert get_credentials_from_string(credentials_string) == expected_result - - # Test case 2: Single line credentials string - credentials_string = "username=abc123" - expected_result = {"username": "abc123"} - assert get_credentials_from_string(credentials_string) == expected_result - - # Test case 3: Multiple line credentials string - credentials_string = "username=abc123\npassword=xyz789\ntoken=123456" - expected_result = {"username": "abc123", "password": "xyz789", "token": "123456"} - assert get_credentials_from_string(credentials_string) == expected_result - - # Test case 4: Invalid credentials string - credentials_string = "username=abc123\npassword" - try: - get_credentials_from_string(credentials_string) - except Exception as e: - assert str(e) == "Error processing secrets: 'not enough values to unpack (expected 2, got 1)'" diff --git a/infrapatch/action/tests/test_config.py b/infrapatch/action/tests/test_config.py new file mode 100644 index 0000000..c465f0d --- /dev/null +++ b/infrapatch/action/tests/test_config.py @@ -0,0 +1,101 @@ +import os +from pathlib import Path + +import pytest + +from infrapatch.action.config import ActionConfigProvider, MissingConfigException, _from_env_to_bool, _get_credentials_from_string, _get_value_from_env + + +def test_get_credentials_from_string(): + # Test case 1: Empty credentials string + credentials_string = "" + expected_result = {} + assert _get_credentials_from_string(credentials_string) == expected_result + + # Test case 2: Single line credentials string + credentials_string = "username=abc123" + expected_result = {"username": "abc123"} + assert _get_credentials_from_string(credentials_string) == expected_result + + # Test case 3: Multiple line credentials string + credentials_string = "username=abc123\npassword=xyz789\ntoken=123456" + expected_result = {"username": "abc123", "password": "xyz789", "token": "123456"} + assert _get_credentials_from_string(credentials_string) == expected_result + + # Test case 4: Invalid credentials string + credentials_string = "username=abc123\npassword" + try: + _get_credentials_from_string(credentials_string) + except Exception as e: + assert str(e) == "Error processing secrets: 'not enough values to unpack (expected 2, got 1)'" + + +def test_get_value_from_env(): + # Test case 1: Value exists in os.environ + os.environ["TEST_VALUE"] = "abc123" + assert _get_value_from_env("TEST_VALUE") == "abc123" + + # Test case 2: Value does not exist in os.environ + os.environ.clear() + try: + _get_value_from_env("TEST_VALUE") + except MissingConfigException as e: + assert str(e) == "Missing configuration for key: TEST_VALUE" + + # Test case 3: Value does not exist in os.environ, but default is provided + assert _get_value_from_env("TEST_VALUE", default="abc123") == "abc123" + + +@pytest.mark.parametrize("working_directory_relative_path", ["/working/directory", ""], ids=lambda d: f"x={d}") +def test_action_config_init(working_directory_relative_path): + # Test case 1: All values exist in os.environ + os.environ["GITHUB_TOKEN"] = "abc123" + os.environ["HEAD_BRANCH"] = "main" + os.environ["TARGET_BRANCH"] = "develop" + os.environ["REPOSITORY_NAME"] = "my-repo" + os.environ["WORKING_DIRECTORY_RELATIVE"] = working_directory_relative_path + os.environ["DEFAULT_REGISTRY_DOMAIN"] = "registry.example.com" + os.environ["REGISTRY_SECRET_STRING"] = "test_registry.ch=abc123" + os.environ["REPORT_ONLY"] = "False" + + config = ActionConfigProvider() + + assert config.github_token == "abc123" + assert config.head_branch == "main" + assert config.target_branch == "develop" + assert config.repository_name == "my-repo" + assert config.working_directory == Path(os.getcwd()).joinpath(working_directory_relative_path) + assert config.repository_root == Path(os.getcwd()) + assert config.default_registry_domain == "registry.example.com" + assert config.registry_secrets == {"test_registry.ch": "abc123"} + assert config.report_only is False + + # Test case 2: Missing values in os.environ + os.environ.clear() + try: + config = ActionConfigProvider() + except MissingConfigException as e: + assert str(e).__contains__("Missing configuration for key") + + +def test_env_to_bool(): + # Test case 1: True values + assert _from_env_to_bool("true") is True + assert _from_env_to_bool("1") is True + assert _from_env_to_bool("yes") is True + assert _from_env_to_bool("y") is True + assert _from_env_to_bool("t") is True + + # Test case 2: False values + assert _from_env_to_bool("false") is False + assert _from_env_to_bool("0") is False + assert _from_env_to_bool("no") is False + assert _from_env_to_bool("n") is False + assert _from_env_to_bool("f") is False + + # Test case 3: Case-insensitive + assert _from_env_to_bool("True") is True + assert _from_env_to_bool("FALSE") is False + assert _from_env_to_bool("YeS") is True + assert _from_env_to_bool("N") is False + assert _from_env_to_bool("T") is True diff --git a/infrapatch/cli/__main__.py b/infrapatch/cli/__main__.py index 5be3a36..446b897 100644 --- a/infrapatch/cli/__main__.py +++ b/infrapatch/cli/__main__.py @@ -1,8 +1,71 @@ -from infrapatch.cli import cli +from pathlib import Path +from typing import Union +import click -def main(): - cli.main() +from infrapatch.cli.__init__ import __version__ +from infrapatch.core.composition import MainHandler, build_main_handler +from infrapatch.core.log_helper import catch_exception, setup_logging + +main_handler: Union[MainHandler, None] = None + + +@click.group(invoke_without_command=True) +@click.option("--debug", is_flag=True, help="Enable debug logging.") +@click.option("--version", is_flag=True, help="Prints the version of the tool.") +@click.option("--credentials-file-path", default=None, help="Path to a file containing credentials for private registries.") +@click.option("--default_registry_domain", default="registry.terraform.io", help="Default registry domain for resources without a specified domain.") +@catch_exception(handle=Exception) +def main(debug: bool, version: bool, credentials_file_path: str, default_registry_domain: str): + if version: + print(f"You are running infrapatch version: {__version__}") + exit(0) + setup_logging(debug) + global main_handler + credentials_file = None + if credentials_file_path is not None: + credentials_file = Path(credentials_file_path) + main_handler = build_main_handler(default_registry_domain, credentials_file) + + +# noinspection PyUnresolvedReferences +@main.command() +@click.option("--project-root-path", default=None, help="Root directory of the project. If not specified, the current working directory is used.") +@click.option("--only-upgradable", is_flag=True, help="Only show providers and modules that can be upgraded.") +@click.option("--dump-json-statistics", is_flag=True, help="Creates a json file containing statistics about the found resources and there update status as json file in the cwd.") +@catch_exception(handle=Exception) +def report(project_root_path: str, only_upgradable: bool, dump_json_statistics: bool): + """Finds all modules and providers in the project_root and prints the newest version.""" + if project_root_path is None: + project_root = Path.cwd() + else: + project_root = Path(project_root_path) + global main_handler + if main_handler is None: + raise Exception("main_handler not initialized.") + resources = main_handler.get_all_terraform_resources(project_root) + main_handler.print_resource_table(resources, only_upgradable) + main_handler.dump_statistics(resources, dump_json_statistics) + + +@main.command() +@click.option("--project-root-path", default=None, help="Root directory of the project. If not specified, the current working directory is used.") +@click.option("--confirm", is_flag=True, help="Apply changes without confirmation.") +@click.option("--dump-json-statistics", is_flag=True, help="Creates a json file containing statistics about the updated resources in the cwd.") +@catch_exception(handle=Exception) +def update(project_root_path: str, confirm: bool, dump_json_statistics: bool): + """Finds all modules and providers in the project_root and updates them to the newest version.""" + if project_root_path is None: + project_root = Path.cwd() + else: + project_root = Path(project_root_path) + global main_handler + if main_handler is None: + raise Exception("main_handler not initialized.") + + resources = main_handler.get_all_terraform_resources(project_root) + main_handler.update_resources(resources, confirm, project_root, project_root) + main_handler.dump_statistics(resources, dump_json_statistics) if __name__ == "__main__": diff --git a/infrapatch/cli/cli.py b/infrapatch/cli/cli.py deleted file mode 100644 index 0d632b4..0000000 --- a/infrapatch/cli/cli.py +++ /dev/null @@ -1,67 +0,0 @@ -from pathlib import Path -from typing import Union - -import click - -from infrapatch.cli.__init__ import __version__ -from infrapatch.core.composition import MainHandler, build_main_handler -from infrapatch.core.log_helper import catch_exception, setup_logging - -main_handler: Union[MainHandler, None] = None - - -@click.group(invoke_without_command=True) -@click.option("--debug", is_flag=True, help="Enable debug logging.") -@click.option("--version", is_flag=True, help="Prints the version of the tool.") -@click.option("--credentials-file-path", default=None, help="Path to a file containing credentials for private registries.") -@click.option("--default_registry_domain", default="registry.terraform.io", help="Default registry domain for resources without a specified domain.") -@catch_exception(handle=Exception) -def main(debug: bool, version: bool, credentials_file_path: str, default_registry_domain: str): - if version: - print(f"You are running infrapatch version: {__version__}") - exit(0) - setup_logging(debug) - global main_handler - credentials_file = None - if credentials_file_path is not None: - credentials_file = Path(credentials_file_path) - main_handler = build_main_handler(default_registry_domain, credentials_file) - - -# noinspection PyUnresolvedReferences -@main.command() -@click.option("--project-root-path", default=None, help="Root directory of the project. If not specified, the current working directory is used.") -@click.option("--only-upgradable", is_flag=True, help="Only show providers and modules that can be upgraded.") -@click.option("--dump-json-statistics", is_flag=True, help="Creates a json file containing statistics about the found resources and there update status as json file in the cwd.") -@catch_exception(handle=Exception) -def report(project_root_path: str, only_upgradable: bool, dump_json_statistics: bool): - """Finds all modules and providers in the project_root and prints the newest version.""" - if project_root_path is None: - project_root = Path.cwd() - else: - project_root = Path(project_root_path) - global main_handler - if main_handler is None: - raise Exception("main_handler not initialized.") - resources = main_handler.get_all_terraform_resources(project_root) - main_handler.print_resource_table(resources, only_upgradable) - main_handler.dump_statistics(resources, dump_json_statistics) - - -@main.command() -@click.option("--project-root-path", default=None, help="Root directory of the project. If not specified, the current working directory is used.") -@click.option("--confirm", is_flag=True, help="Apply changes without confirmation.") -@click.option("--dump-json-statistics", is_flag=True, help="Creates a json file containing statistics about the updated resources in the cwd.") -@catch_exception(handle=Exception) -def update(project_root_path: str, confirm: bool, dump_json_statistics: bool): - """Finds all modules and providers in the project_root and updates them to the newest version.""" - if project_root_path is None: - project_root = Path.cwd() - else: - project_root = Path(project_root_path) - global main_handler - if main_handler is None: - raise Exception("main_handler not initialized.") - resources = main_handler.get_all_terraform_resources(project_root) - main_handler.update_resources(resources, confirm, Path(project_root)) - main_handler.dump_statistics(resources, dump_json_statistics) diff --git a/infrapatch/core/composition.py b/infrapatch/core/composition.py index 9bd410f..bdc20f9 100644 --- a/infrapatch/core/composition.py +++ b/infrapatch/core/composition.py @@ -86,7 +86,7 @@ def print_resource_table(self, resources: Sequence[VersionedTerraformResource], # noinspection PyUnboundLocalVariable def update_resources( - self, resources: Sequence[VersionedTerraformResource], confirm: bool, working_dir: Path, commit_changes: bool = False + self, resources: Sequence[VersionedTerraformResource], confirm: bool, working_dir: Path, repo_root: Path, commit_changes: bool = False ) -> Sequence[VersionedTerraformResource]: upgradable_resources = get_upgradable_resources(resources) if len(upgradable_resources) == 0: diff --git a/infrapatch/core/utils/git.py b/infrapatch/core/utils/git.py new file mode 100644 index 0000000..75d69ae --- /dev/null +++ b/infrapatch/core/utils/git.py @@ -0,0 +1,38 @@ +from pathlib import Path +import subprocess +from typing import Union +import logging as log + + +class GitException(Exception): + pass + + +class Git: + _repo_path: Path + + def __init__(self, repo_path: Path): + self._repo_path = repo_path + + def run_git_command(self, command: list[str]) -> tuple[str, Union[str, None]]: + command = ["git", *command] + command_string = " ".join(command) + log.debug(f"Executing git command: {command_string}") + try: + result = subprocess.run(command, capture_output=True, text=True, cwd=self._repo_path.absolute().as_posix()) + except Exception as e: + raise GitException(f"Error executing git command {command_string} with error: {e}") + if result.returncode != 0: + raise GitException(f"Git command {command_string} exited with non-zero exit code {result.returncode}: {result.stderr}") + return result.stdout, result.stderr + + def fetch_origin(self): + log.debug("Fetching origin") + self.run_git_command(["fetch", "origin"]) + + def checkout_branch(self, target: str, origin: str): + log.debug(f"Checking out branch {target} from {origin}") + self.run_git_command(["checkout", "-b", target, origin]) + + def push(self, additional_arguments: list[str] = []): + self.run_git_command(["push", *additional_arguments])