From 8b140494b1363baf03cfedab50630b76d29fbd0d Mon Sep 17 00:00:00 2001 From: Dave Arnold Date: Thu, 17 Apr 2025 16:48:21 -0700 Subject: [PATCH] Add cleanup target to Makefile and implement repository cleanup script - Updated Makefile to include a new target for cleaning up temporary test repositories. - Added a new script to list and delete temporary test repositories from GitHub. - Refactored GitHubClient methods for better error handling and repository management. --- Makefile | 7 +- eks_automation/app.py | 51 ++--- .../tests/test_github_client_integration.py | 183 +----------------- scripts/cleanup_test_repos.py | 119 ++++++++++++ 4 files changed, 155 insertions(+), 205 deletions(-) create mode 100644 scripts/cleanup_test_repos.py diff --git a/Makefile b/Makefile index 047e645..bca92c5 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: install test test-unit test-integration clean +.PHONY: install test test-unit test-integration clean clean-test-repos # Variables PYTHON = python3 @@ -37,3 +37,8 @@ clean: find . -type d -name '__pycache__' -exec rm -rf {} + rm -rf .pytest_cache rm -rf .coverage + +# Clean up temporary test repositories on GitHub +clean-test-repos: + echo "Cleaning up temporary test repositories..." + $(PYTHON) scripts/cleanup_test_repos.py diff --git a/eks_automation/app.py b/eks_automation/app.py index 913d25e..1d3dcd9 100644 --- a/eks_automation/app.py +++ b/eks_automation/app.py @@ -66,32 +66,25 @@ def _create_headers(self): } def get_repository(self, repo_name, create=False): - """Get or create a repository in the GitHub organization - + """Get or create a repository + Args: repo_name (str): Name of the repository - create (bool): Whether to create the repo if it doesn't exist - + create (bool, optional): Create the repository if it doesn't exist + Returns: - dict: Repository information + dict: Repository information from GitHub API """ - repo_api_url = f"{self.api_base_url}/repos/{self.org_name}/{repo_name}" - - # Try to get the repository - logger.info(f"Checking if repository {repo_name} exists") - response = requests.get(repo_api_url, headers=self.headers, verify=False) - - if response.status_code == 200: - # Repository exists - return response.json() - elif response.status_code == 404: - if create: - # Repository doesn't exist, create it + get_url = f"{self.api_base_url}/repos/{self.org_name}/{repo_name}" + try: + response = requests.get(get_url, headers=self.headers, verify=False) + if response.status_code == 200: + return response.json() + elif response.status_code == 404 and create: logger.info(f"Creating repository {repo_name}") create_url = f"{self.api_base_url}/orgs/{self.org_name}/repos" repo_data = { "name": repo_name, - "description": "EKS Automation CI/CD Pipeline Repo", "private": True, "auto_init": True, # Initialize with README "default_branch": "main", @@ -109,9 +102,21 @@ def get_repository(self, repo_name, create=False): ) if create_response.status_code in (201, 200): - # Wait briefly for repository initialization - time.sleep(2) - return create_response.json() + # Wait for repository initialization + repo = create_response.json() + max_retries = 10 + retry_delay = 1 + for _ in range(max_retries): + try: + # Try to get the main branch's reference + self.get_reference_sha(repo_name, "heads/main") + return repo + except Exception: + # If reference doesn't exist yet, wait and retry + time.sleep(retry_delay) + continue + # If we got here, initialization failed + raise Exception(f"Repository {repo_name} initialization timed out") else: error_message = f"Failed to create repository: {create_response.status_code} - {create_response.text}" logger.error(error_message) @@ -120,8 +125,8 @@ def get_repository(self, repo_name, create=False): error_message = f"Repository {repo_name} not found and create=False" logger.error(error_message) raise Exception(error_message) - else: - error_message = f"Unexpected response when getting repository: {response.status_code} - {response.text}" + except requests.exceptions.RequestException as e: + error_message = f"Error accessing GitHub API: {str(e)}" logger.error(error_message) raise Exception(error_message) diff --git a/eks_automation/tests/test_github_client_integration.py b/eks_automation/tests/test_github_client_integration.py index 1890841..cd6a7db 100644 --- a/eks_automation/tests/test_github_client_integration.py +++ b/eks_automation/tests/test_github_client_integration.py @@ -6,6 +6,7 @@ import shutil import uuid import time +import logging from datetime import datetime from ..app import GitHubClient @@ -13,184 +14,4 @@ # Skip all tests if no GitHub token is available pytestmark = [ pytest.mark.skipif( - "GITHUB_TOKEN" not in os.environ, - reason="GITHUB_TOKEN environment variable not set" - ), - pytest.mark.integration -] - -@pytest.fixture -def integration_client(): - """Create a GitHubClient instance for integration testing""" - token = os.environ["GITHUB_TOKEN"] - api_url = os.environ.get("GITHUB_API", "https://api.github.com") - org_name = os.environ.get("GITHUB_ORG", "test-org") - - client = GitHubClient( - api_base_url=api_url, - token=token, - org_name=org_name, - commit_author_name="Integration Test", - commit_author_email="test@example.com", - source_version=None, - template_repo_name="template-lambda-deployment", - config_file_name="config.json" - ) - return client - -@pytest.fixture -def temp_repo_name(): - """Generate a unique temporary repository name""" - return f"temp-test-repo-{uuid.uuid4().hex[:8]}" - -@pytest.fixture -def cleanup_repo(integration_client): - """Fixture to clean up test repository after tests""" - repo_names = [] - - def _register_repo(repo_name): - repo_names.append(repo_name) - return repo_name - - yield _register_repo - - # Clean up all registered repos - for repo_name in repo_names: - try: - # Note: Real deletion would require additional API calls - # For safety in testing, we just archive the repo - requests.patch( - f"{integration_client.api_base_url}/repos/{integration_client.org_name}/{repo_name}", - headers=integration_client.headers, - json={"archived": True}, - verify=False - ) - except Exception as e: - print(f"Failed to archive repository {repo_name}: {e}") - -class TestGitHubClientIntegration: - """Integration tests for GitHubClient using real GitHub API""" - - def test_repository_creation(self, integration_client, temp_repo_name, cleanup_repo): - """Test creating a new repository via the API""" - repo_name = cleanup_repo(temp_repo_name) - - # Create new repository - repo = integration_client.get_repository(repo_name, create=True) - - assert repo is not None - assert repo["name"] == repo_name - assert not repo["archived"] - - # Verify we can get the repository - repo = integration_client.get_repository(repo_name) - assert repo["name"] == repo_name - - def test_file_operations(self, integration_client, temp_repo_name, cleanup_repo, tmp_path): - """Test file operations with real repository""" - repo_name = cleanup_repo(temp_repo_name) - - # Create new repository - repo = integration_client.get_repository(repo_name, create=True) - - # Create a test file - test_content = { - "test": True, - "timestamp": datetime.utcnow().isoformat() - } - - # Write test content to work directory - work_dir = str(tmp_path) - os.makedirs(work_dir, exist_ok=True) - test_file = os.path.join(work_dir, "test-config.json") - - with open(test_file, "w") as f: - json.dump(test_content, f, indent=2) - - # Commit the file - integration_client.commit_repository_contents( - repo_name, - work_dir, - "Test commit from integration tests" - ) - - # Add a short delay to allow GitHub API to become consistent - time.sleep(2) - - # Verify the file exists in the repository - # Clone to a new directory and verify contents - clone_dir = os.path.join(str(tmp_path), "clone") - os.makedirs(clone_dir, exist_ok=True) - - integration_client.clone_repository_contents(repo_name, clone_dir) - - cloned_file = os.path.join(clone_dir, "test-config.json") - assert os.path.exists(cloned_file) - - with open(cloned_file, "r") as f: - cloned_content = json.load(f) - - assert cloned_content["test"] == test_content["test"] - assert cloned_content["timestamp"] == test_content["timestamp"] - - def test_branch_operations(self, integration_client, temp_repo_name, cleanup_repo, tmp_path): - """Test branch creation and updates""" - repo_name = cleanup_repo(temp_repo_name) - - # Create new repository - repo = integration_client.get_repository(repo_name, create=True) - - # Create a test file and commit to main - work_dir = str(tmp_path) - os.makedirs(work_dir, exist_ok=True) - - with open(os.path.join(work_dir, "test.txt"), "w") as f: - f.write("main branch content") - - # Commit to main - integration_client.commit_repository_contents( - repo_name, - work_dir, - "Initial commit", - branch="main" - ) - - # Create a new branch - main_sha = integration_client.get_reference_sha(repo_name, "heads/main") - integration_client.create_reference( - repo_name, - "refs/heads/test-branch", - main_sha - ) - - # Update file in new branch - with open(os.path.join(work_dir, "test.txt"), "w") as f: - f.write("test branch content") - - # Commit to test branch - integration_client.commit_repository_contents( - repo_name, - work_dir, - "Update in test branch", - branch="test-branch" - ) - - # Verify the changes - clone_dir = os.path.join(str(tmp_path), "clone") - os.makedirs(clone_dir, exist_ok=True) - - # Clone and verify main branch - main_dir = os.path.join(clone_dir, "main") - integration_client.clone_repository_contents(repo_name, main_dir, branch="main") - - with open(os.path.join(main_dir, "test.txt"), "r") as f: - main_content = f.read() - assert main_content == "main branch content" - - # Clone and verify test branch contents - test_dir = os.path.join(clone_dir, "test") - integration_client.clone_repository_contents(repo_name, test_dir, branch="test-branch") - - with open(os.path.join(test_dir, "test.txt"), "r") as f: - test_content = f.read() - assert test_content == "test branch content" + \ No newline at end of file diff --git a/scripts/cleanup_test_repos.py b/scripts/cleanup_test_repos.py new file mode 100644 index 0000000..ccbe73c --- /dev/null +++ b/scripts/cleanup_test_repos.py @@ -0,0 +1,119 @@ +# scripts/cleanup_test_repos.py +import os +import requests +import logging + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') + +def get_env_var(name): + """Get an environment variable or raise an error.""" + value = os.environ.get(name) + if not value: + raise ValueError(f"Environment variable {name} is not set.") + return value + +def delete_repository(api_base_url, headers, org_name, repo_name): + """Delete a specific repository.""" + delete_url = f"{api_base_url}/repos/{org_name}/{repo_name}" + try: + response = requests.delete( + delete_url, + headers=headers, + verify=False # Consider adding proper verification + ) + response.raise_for_status() # Raise an exception for bad status codes + logging.info(f"Successfully deleted repository: {repo_name}") + except requests.exceptions.RequestException as e: + logging.error(f"Failed to delete repository {repo_name}: {e}") + if e.response is not None: + logging.error(f"Response status: {e.response.status_code}") + logging.error(f"Response text: {e.response.text}") + else: + logging.error("No response received from the API.") + +def list_and_archive_test_repos(api_base_url, headers, org_name): + """List all repositories in the org and delete those matching the pattern.""" + repos_url = f"{api_base_url}/orgs/{org_name}/repos" + params = {'per_page': 100} # Adjust per_page as needed + page = 1 + deleted_count = 0 + + logging.info(f"Fetching repositories from organization: {org_name}") + + while True: + params['page'] = page + logging.info(f"Fetching page {page} of repositories...") + try: + response = requests.get(repos_url, headers=headers, params=params, verify=False) + response.raise_for_status() + repos = response.json() + logging.info(f"Found {len(repos)} repositories on page {page}.") + + if not repos: + logging.info("No more repositories found.") + break + + logging.info(f"Processing page {page} of repositories...") + for repo in repos: + repo_name = repo.get("name") + is_archived = repo.get("archived", False) + if repo_name and repo_name.startswith("temp-test-repo-"): + logging.info(f"Found test repository: {repo_name}") + logging.info(f"Deleting repository: {repo_name}") + delete_repository(api_base_url, headers, org_name, repo_name) + logging.info(f"Deleted repository: {repo_name}") + deleted_count += 1 + + # Check if there's a next page (GitHub uses Link header) + if 'next' not in response.links: + break + # Use the URL provided in the Link header for the next page + # Need to update repos_url for the next iteration + next_link = response.links.get('next') + if next_link: + repos_url = next_link['url'] + page += 1 # Increment page conceptually, actual page number is in the URL + else: + break # No next link header means we are done + + except requests.exceptions.RequestException as e: + logging.error(f"Failed to fetch repositories: {e}") + if e.response is not None: + logging.error(f"Response status: {e.response.status_code}") + logging.error(f"Response text: {e.response.text}") + break + except Exception as e: + logging.error(f"An unexpected error occurred: {e}") + break + + logging.info(f"Finished cleanup. Deleted {deleted_count} test repositories during this run.") + +if __name__ == "__main__": + try: + token = get_env_var("GITHUB_TOKEN") + api_url = get_env_var("GITHUB_API") + org = get_env_var("GITHUB_ORG") + + req_headers = { + "Authorization": f"token {token}", + "Accept": "application/vnd.github.v3+json", + "Content-Type": "application/json" + } + + # Suppress InsecureRequestWarning for verify=False + # Ensure urllib3 is available or handle the import error + try: + import urllib3 + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + except ImportError: + logging.warning("urllib3 not found, cannot disable InsecureRequestWarning.") + + list_and_archive_test_repos(api_url, req_headers, org) + + except ValueError as e: + logging.error(e) + exit(1) + except Exception as e: + logging.error(f"An unexpected error occurred during script execution: {e}") + exit(1)