diff --git a/README.md b/README.md index f3cf778..49cbbf4 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,12 @@ python sbg.py \ --- +## Logging + +The script uses Python's standard `logging` module for all console output. This provides users with more control over verbosity and output destinations (e.g., logging to a file). By default, it logs messages at the `INFO` level and above to standard output. + +--- + ## Script Reference ```text diff --git a/sbg.py b/sbg.py old mode 100755 new mode 100644 index f24788e..0ae9546 --- a/sbg.py +++ b/sbg.py @@ -3,10 +3,19 @@ import sys import argparse import requests +import logging import subprocess + +logger = logging.getLogger(__name__) from urllib.parse import urljoin def parse_args(): + """ + Parses command-line arguments for cloning or updating GitLab repositories. + + Returns: + argparse.Namespace: Parsed arguments including GitLab URL, access token, group IDs or paths, destination directory, and SSH usage flag. + """ p = argparse.ArgumentParser( description="Clone or pull all GitLab repos under one or more groups " "(including nested subgroups) into folders by namespace." @@ -56,7 +65,15 @@ def list_projects(self, group_id): params={"include_subgroups": False}) def gather_all_projects(self, group_id): - """Recursively gather all projects under group_id.""" + """ + Recursively collects all projects within a group and its nested subgroups. + + Args: + group_id: The ID of the root group to search. + + Returns: + A list of project metadata dictionaries for all projects found under the group and its subgroups. + """ projects = [] stack, seen = [group_id], set() while stack: @@ -68,7 +85,7 @@ def gather_all_projects(self, group_id): projs = self.list_projects(gid) subs = self.list_subgroups(gid) except requests.HTTPError as error: - print(f"Warning: could not fetch for group {gid}: {error}") + logger.warning("Could not fetch for group %s: %s", gid, error) continue projects.extend(projs) for sg in subs: @@ -77,31 +94,41 @@ def gather_all_projects(self, group_id): @staticmethod def clone_or_pull(repo_url, target_path): - """Clone if missing, or pull if already a Git repo.""" + """ + Clones a Git repository to the target path or updates it if already present. + + If the target path is an existing Git repository, performs a `git pull` to update it. Otherwise, clones the repository from the given URL into the target path. Logs errors if cloning or pulling fails. + """ if os.path.isdir(target_path) and os.path.isdir(os.path.join(target_path, ".git")): - print(f"Updating existing repo at {target_path}") + logger.info("Updating existing repo at %s", target_path) try: subprocess.check_call(["git", "-C", target_path, "pull"]) except subprocess.CalledProcessError as error: - print(f" ❌ pull failed for {target_path}: {error}") + logger.error("Pull failed for %s: %s", target_path, error) else: - print(f"Cloning into {target_path}") + logger.info("Cloning into %s", target_path) try: subprocess.check_call(["git", "clone", repo_url, target_path]) except subprocess.CalledProcessError as error: - print(f" ❌ clone failed for {repo_url} into {target_path}: {error}") + logger.error("Clone failed for %s into %s: %s", repo_url, target_path, error) def main(): + """ + Clones or updates all GitLab repositories under specified groups into a local directory. + + Parses command-line arguments, gathers all projects (including those in nested subgroups) for each provided group, deduplicates them, ensures the destination directory structure exists, and clones or pulls each repository into its corresponding namespace folder using either SSH or HTTP URLs. + """ + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True) args = parse_args() cloner = GitLabCloner(args.gitlab_url, args.token, args.use_ssh) # 1) Gather & dedupe all projects all_projects = {} for gid in args.group_ids: - print(f"Fetching projects under group '{gid}' …") + logger.info("Fetching projects under group '%s' …", gid) for proj in cloner.gather_all_projects(gid): all_projects[proj["id"]] = proj - print(f"Total unique projects to process: {len(all_projects)}") + logger.info("Total unique projects to process: %s", len(all_projects)) # 2) Ensure destination root exists dest_root = os.path.abspath(args.dest) @@ -126,6 +153,8 @@ def main(): try: main() except KeyboardInterrupt: - sys.exit("\nAborted by user") + logger.info("Aborted by user") + sys.exit(0) except Exception as e: - sys.exit(f"Fatal error: {e}") + logger.critical("Fatal error: %s", e) + sys.exit(1) diff --git a/test_sbg.py b/test_sbg.py index fb2e3ca..663e7e5 100644 --- a/test_sbg.py +++ b/test_sbg.py @@ -6,6 +6,7 @@ import argparse import requests from io import StringIO +import logging # Import the module to test import sbg @@ -39,6 +40,8 @@ def setUp(self): self.cloner = sbg.GitLabCloner('https://gitlab.com', 'token123', False) self.session_mock = MagicMock() self.cloner.session = self.session_mock + # Configure logging for sbg.py to be captured by assertLogs + logging.getLogger(sbg.__name__).setLevel(logging.INFO) # Or DEBUG for more verbosity def test_init(self): cloner = sbg.GitLabCloner('https://gitlab.com/', 'token123', True) @@ -131,18 +134,12 @@ def side_effect(path, params=None): self.cloner._get = MagicMock(side_effect=side_effect) - # Redirect stdout to capture warnings - saved_stdout = sys.stdout - try: - out = StringIO() - sys.stdout = out + with self.assertLogs(logging.getLogger(sbg.__name__), level='WARNING') as cm: result = self.cloner.gather_all_projects(123) - # Should have 2 projects and a warning - self.assertEqual(len(result), 2) - self.assertIn("Warning", out.getvalue()) - finally: - sys.stdout = saved_stdout + # Should have 2 projects and a warning + self.assertEqual(len(result), 2) + self.assertTrue(any("Could not fetch for group 456" in message for message in cm.output)) @patch('subprocess.check_call') @patch('os.path.isdir') @@ -172,15 +169,10 @@ def test_clone_error_handling(self, isdir_mock, subprocess_mock): isdir_mock.return_value = False subprocess_mock.side_effect = subprocess.CalledProcessError(1, 'git clone') - # Redirect stdout to capture error message - saved_stdout = sys.stdout - try: - out = StringIO() - sys.stdout = out + with self.assertLogs(logging.getLogger(sbg.__name__), level='ERROR') as cm: sbg.GitLabCloner.clone_or_pull('https://gitlab.com/user/repo.git', '/tmp/repo') - self.assertIn("❌ clone failed", out.getvalue()) - finally: - sys.stdout = saved_stdout + + self.assertTrue(any("Clone failed for https://gitlab.com/user/repo.git into /tmp/repo" in message for message in cm.output)) @patch('sbg.GitLabCloner')