Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
51 changes: 40 additions & 11 deletions sbg.py
100755 → 100644
Original file line number Diff line number Diff line change
Expand Up @@ -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."
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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)
Expand All @@ -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)
28 changes: 10 additions & 18 deletions test_sbg.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import argparse
import requests
from io import StringIO
import logging

# Import the module to test
import sbg
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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')
Expand Down Expand Up @@ -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')
Expand Down