diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
new file mode 100644
index 0000000..efc5255
--- /dev/null
+++ b/.github/workflows/tests.yml
@@ -0,0 +1,43 @@
+name: tests
+
+on:
+ push:
+ branches: [master, main]
+ paths:
+ - "gitcolombo.py"
+ - "test_gitcolombo.py"
+ - ".github/workflows/tests.yml"
+ pull_request:
+ paths:
+ - "gitcolombo.py"
+ - "test_gitcolombo.py"
+ - ".github/workflows/tests.yml"
+ workflow_dispatch:
+
+jobs:
+ test:
+ runs-on: ubuntu-latest
+ strategy:
+ fail-fast: false
+ matrix:
+ python-version: ["3.10", "3.11", "3.12", "3.13"]
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python ${{ matrix.python-version }}
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ matrix.python-version }}
+
+ - name: Configure git identity (for end-to-end test)
+ run: |
+ git config --global user.name "ci"
+ git config --global user.email "ci@example.com"
+ git config --global init.defaultBranch main
+ git config --global commit.gpgsign false
+
+ - name: Run tests
+ run: python -m unittest test_gitcolombo -v
+
+ - name: Smoke-check CLI
+ run: python gitcolombo.py --help
diff --git a/.gitignore b/.gitignore
index e69de29..1de2299 100644
--- a/.gitignore
+++ b/.gitignore
@@ -0,0 +1,3 @@
+.DS_Store
+repos/
+__pycache__/
diff --git a/README.md b/README.md
index b69de5a..3534eae 100644
--- a/README.md
+++ b/README.md
@@ -21,6 +21,11 @@ OSINT tool to extract info about persons from git repositories: common names, em
# from all GitHub personal/org repos by nickname
./gitcolombo.py --nickname LubyRuffy
+ # change where remote repos get cloned (default: ./repos)
+ ./gitcolombo.py -u https://github.com/Kalanchyovskaia16/newlps --repos-dir ./clones
+
+Remote repositories are cloned into `./repos/` by default; override with `--repos-dir`.
+
For batch cloning from Gitlab and Bitbucket group repos you can use [ghorg](https://github.com/gabrie30/ghorg).
Output:
@@ -35,6 +40,19 @@ Output:
- different names for the same person
- general statistics
+### Testing
+
+Stdlib-only test suite (`test_gitcolombo.py`), no third-party dependencies.
+Run from the repo root:
+
+ python3 -m unittest test_gitcolombo -v
+
+The end-to-end test creates a real git repository in a temp directory, so a
+working `git` binary is required (the test is skipped if `git` is missing).
+
+Tests run on every push and pull request via GitHub Actions
+(`.github/workflows/tests.yml`) across Python 3.10–3.13.
+
### Details
[RUS] https://telegra.ph/Gitcolombo---OSINT-v-GitHub-03-02
diff --git a/gitcolombo.py b/gitcolombo.py
index 44143a2..c226447 100755
--- a/gitcolombo.py
+++ b/gitcolombo.py
@@ -1,354 +1,442 @@
#!/usr/bin/env python3
+"""Gitcolombo — OSINT tool: extract account info from git repositories.
+
+Walks one or more git repositories and aggregates per-person stats
+(name, email, author/committer counts, alternate identities) and detects
+identity overlaps via shared emails or shared names. Optionally resolves
+GitHub logins by scraping commit pages.
+"""
+from __future__ import annotations
+
import argparse
import json
import logging
import os
import re
import subprocess
-from threading import Thread
+import urllib.error
import urllib.request
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
+from typing import Iterable
-DELIMITER = '---------------'
+DELIMITER = "-" * 15
-LOG_FORMAT = r'%H;"%an %ae";"%cn %ce"'
-LOG_REGEXP = r'(\w+);"(.*?)";"(.*?)"'
-LOG_NAME_REGEXP = r'^(.*?)\s+(\S+)$'
+# git log --pretty format: hash;"author_name author_email";"committer_name committer_email"
+GIT_LOG_FORMAT = r'%H;"%an %ae";"%cn %ce"'
+GIT_LOG_LINE_RE = re.compile(r'(\w+);"(.*?)";"(.*?)"')
+GIT_NAME_EMAIL_RE = re.compile(r"^(.*?)\s+(\S+)$")
+GITHUB_COMMIT_AUTHOR_RE = re.compile(r' 0)
- for page_num in range(1, last_page + 1):
- req_url = url.format(nickname, page_num)
- req = urllib.request.Request(req_url)
- try:
- response = urllib.request.urlopen(req)
- except Exception as e:
- logging.debug(e)
- else:
- repos = json.loads((response.read().decode('utf8')))
- result = [r['html_url'] for r in repos if not only_forks or not r['fork']]
- repos_links.update(set(result))
- return repos_links
+# ---------- HTTP helpers ----------
+
+def _http_get(url: str) -> bytes | None:
+ req = urllib.request.Request(url, headers={"User-Agent": HTTP_USER_AGENT})
+ try:
+ with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
+ return resp.read()
+ except (urllib.error.URLError, TimeoutError) as exc:
+ logger.debug("GET %s failed: %s", url, exc)
+ return None
-def find_all_repos_recursively(path):
- git_dirs = []
+def _http_get_json(url: str):
+ payload = _http_get(url)
+ if payload is None:
+ return None
+ try:
+ return json.loads(payload.decode("utf-8"))
+ except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+ logger.debug("Bad JSON from %s: %s", url, exc)
+ return None
+
+
+# ---------- GitHub API ----------
+
+def get_public_repos_count(nickname: str) -> int:
+ data = _http_get_json(GITHUB_USER_URL.format(nickname=nickname))
+ if not data:
+ return 0
+ return int(data.get("public_repos", 0))
+
+
+def get_github_repos(
+ nickname: str, repos_count: int, include_forks: bool = False,
+) -> set[str]:
+ if repos_count <= 0:
+ return set()
+ last_page = (repos_count + GITHUB_PER_PAGE - 1) // GITHUB_PER_PAGE
+ repos: set[str] = set()
+ for page in range(1, last_page + 1):
+ data = _http_get_json(
+ GITHUB_REPOS_URL.format(
+ nickname=nickname, per_page=GITHUB_PER_PAGE, page=page,
+ )
+ )
+ if not data:
+ continue
+ for repo in data:
+ if include_forks or not repo.get("fork"):
+ repos.add(repo["html_url"])
+ return repos
+
+
+def resolve_github_username(repo_url: str, commit_hash: str) -> str | None:
+ """Scrape commit page to find the GitHub login behind an email."""
+ if not repo_url.startswith("https://github.com/"):
+ return None
+ commit_url = f"{repo_url.rstrip('/')}/commit/{commit_hash}"
+ page = _http_get(commit_url)
+ if page is None:
+ return None
+ match = GITHUB_COMMIT_AUTHOR_RE.search(page.decode("utf-8", errors="replace"))
+ return match.group(1) if match else None
+
+
+# ---------- Filesystem helpers ----------
+
+def find_all_repos_recursively(path: str) -> list[str]:
+ """Return repo roots (directories that contain a .git subdir) under path."""
+ repos: list[str] = []
for current_dir, dirs, _ in os.walk(path):
- if current_dir.endswith('.git'):
- git_dirs.append(current_dir)
- while dirs:
- dirs.pop()
+ if ".git" in dirs:
+ repos.append(current_dir)
+ dirs[:] = [d for d in dirs if d != ".git"]
+ return repos
- return git_dirs
+# ---------- Git subprocess ----------
-class Commit:
- """
- Extract and store basic commit info
- """
- @staticmethod
- def _extract_name_email(log_str_part):
- extracted = re.search(LOG_NAME_REGEXP, log_str_part)
- if not extracted:
- logging.error('Could not extract name/email from "%s"', log_str_part)
- return ('', '')
-
- return extracted.groups()
-
-
- def __init__(self, log_str):
- extracted = re.search(LOG_REGEXP, log_str)
- if not extracted:
- logging.error('Could not commit info from "%s"', log_str)
- else:
- self.hash, self.author, self.committer = extracted.groups()
- self.author_name, self.author_email = Commit._extract_name_email(self.author)
- self.committer_name, self.committer_email = Commit._extract_name_email(self.committer)
-
- self.author_committer_names_same = self.author_name == self.committer_name
- self.author_committer_emails_same = self.author_email == self.committer_email
-
- self.author_committer_same = self.author_committer_names_same and self.author_committer_emails_same
-
-
- def __str__(self):
- return """Hash: {hash}
-Author name: {author_name}
-Author email: {author_email}
-Committer name: {committer_name}
-Committer email: {committer_email}
- """.format(
- hash=self.hash,
- author_name=self.author_name, author_email=self.author_email,
- committer_name=self.committer_name, committer_email=self.committer_email,
+def git_log(repo_dir: str) -> str:
+ try:
+ result = subprocess.run(
+ ["git", "log", f"--pretty={GIT_LOG_FORMAT}", "--all"],
+ cwd=repo_dir, check=False, capture_output=True, text=True,
)
+ except FileNotFoundError:
+ logger.error("'git' binary not found")
+ return ""
+ if result.returncode != 0:
+ logger.debug("git log failed in %s: %s", repo_dir, result.stderr.strip())
+ return result.stdout
-class Git:
- """
- Make external git work
- """
- @staticmethod
- def get_tree_info(git_dir):
- process = subprocess.Popen(GIT_EXTRACT_CMD, cwd=git_dir, shell=True, stdout=subprocess.PIPE)
- stat = process.stdout.read().decode()
- return stat
-
- @staticmethod
- def clone(link):
- process = subprocess.Popen(GIT_CLONE_CMD.format(link), shell=True, stdout=subprocess.PIPE)
- res = process.stdout.read().decode()
- return res
-
- @staticmethod
- def get_verified_username(repo_url, commit, person):
- if not repo_url.startswith('https://github.com/'):
- return
+def _clone_target_dir(url: str) -> str:
+ name = url.rstrip("/").split("/")[-1]
+ return name[:-4] if name.endswith(".git") else name
- commit_link = repo_url.rstrip('/') + '/commit/' + commit.hash
- req = urllib.request.Request(commit_link)
- try:
- response = urllib.request.urlopen(req)
- page_source = response.read()
- # TODO: authored and committed
- extracted = re.search(r' str | None:
+ """Clone *url* into *dest_dir*/. Returns the cloned path or None."""
+ os.makedirs(dest_dir, exist_ok=True)
+ target = os.path.join(dest_dir, _clone_target_dir(url))
+ try:
+ result = subprocess.run(
+ ["git", "clone", url, target],
+ check=False, capture_output=True, text=True,
+ )
+ except FileNotFoundError:
+ logger.error("'git' binary not found")
+ return None
+ if result.returncode != 0:
+ logger.debug("git clone failed for %s: %s", url, result.stderr.strip())
+ return None
+ return target
+
+
+# ---------- Data classes ----------
+
+def _split_name_email(raw: str) -> tuple[str, str]:
+ m = GIT_NAME_EMAIL_RE.match(raw)
+ if not m:
+ logger.error("Could not extract name/email from %r", raw)
+ return "", ""
+ return m.group(1), m.group(2)
+
+
+@dataclass
+class Commit:
+ hash: str
+ author: str
+ committer: str
+ author_name: str
+ author_email: str
+ committer_name: str
+ committer_email: str
- name = extracted.groups(0)[0]
- person.github_link = name
- logging.debug(commit_link + '\n' + name)
+ @property
+ def author_committer_same(self) -> bool:
+ return (
+ self.author_name == self.committer_name
+ and self.author_email == self.committer_email
+ )
- except Exception as e:
- logging.debug(e)
+ @classmethod
+ def parse(cls, line: str) -> "Commit | None":
+ m = GIT_LOG_LINE_RE.search(line)
+ if not m:
+ logger.error("Could not parse commit line %r", line)
+ return None
+ h, author, committer = m.groups()
+ a_name, a_email = _split_name_email(author)
+ c_name, c_email = _split_name_email(committer)
+ return cls(h, author, committer, a_name, a_email, c_name, c_email)
+
+ def __str__(self) -> str:
+ return (
+ f"Hash: {self.hash}\n"
+ f"Author name: {self.author_name}\n"
+ f"Author email: {self.author_email}\n"
+ f"Committer name: {self.committer_name}\n"
+ f"Committer email: {self.committer_email}\n"
+ )
+@dataclass
class Person:
- """
- Basic person info from commit
- """
- def __init__(self, desc):
- self.name = ''
- self.email = ''
- self.desc = desc
- self.as_author = 0
- self.as_committer = 0
- self.also_known = {}
- self.github_link = None
-
- def __str__(self):
- result = "Name:\t\t\t{name}\nEmail:\t\t\t{email}".format(name=self.name, email=self.email)
+ key: str
+ name: str = ""
+ email: str = ""
+ as_author: int = 0
+ as_committer: int = 0
+ also_known: dict[str, "Person"] = field(default_factory=dict)
+ github_login: str | None = None
+ repo_url: str | None = None
+ last_commit_hash: str | None = None
+
+ def __str__(self) -> str:
+ lines = [
+ f"Name:\t\t\t{self.name}",
+ f"Email:\t\t\t{self.email}",
+ ]
if self.as_author:
- result += "\nAppears as author:\t{} times".format(self.as_author)
+ lines.append(f"Appears as author:\t{self.as_author} times")
if self.as_committer:
- result += "\nAppears as committer:\t{} times".format(self.as_committer)
- if self.github_link:
- result += "\nVerified account:\n\t\t\thttps://github.com/{}".format(self.github_link)
+ lines.append(f"Appears as committer:\t{self.as_committer} times")
+ if self.github_login:
+ lines.append(
+ f"Verified account:\n\t\t\thttps://github.com/{self.github_login}"
+ )
if self.also_known:
- result += '\nAlso appears with:{}'.format(
- '\n\t\t\t'.join(['']+list(self.also_known.keys()))
+ lines.append(
+ "Also appears with:" + "".join(f"\n\t\t\t{k}" for k in self.also_known)
)
+ return "\n".join(lines)
- return result
+# ---------- Analyst ----------
class GitAnalyst:
- """
- Git analysis
- """
- def __init__(self):
- self.git = Git()
-
- self.commits = []
- self.persons = {}
- self.names = {}
- self.emails = {}
- self.repos = []
- self.same_emails_persons = {}
-
- def append(self, source=None):
- if not source:
- return
-
- if not '://' in source:
- git_dir = source
+ def __init__(self, repos_dir: str = DEFAULT_REPOS_DIR) -> None:
+ self.repos_dir = repos_dir
+ self.commits: list[Commit] = []
+ self.persons: dict[str, Person] = {}
+ self.name_to_emails: dict[str, set[str]] = defaultdict(set)
+ self.repos: list[str] = []
+ self.same_emails_persons: dict[str, tuple[list[str], set[str]]] = {}
+
+ def append(self, source: str) -> None:
+ if "://" in source:
+ repo_dir = git_clone(source, self.repos_dir)
+ if repo_dir is None:
+ return
else:
- self.git.clone(source)
- git_dir = source.split('/')[-1]
-
- self.repos.append(git_dir)
- git_info = self.git.get_tree_info(git_dir)
- text_commits = filter(lambda x: x, git_info.split('\n'))
- new_commits = list(map(Commit, text_commits))
- self.commits += new_commits
+ repo_dir = source
- self.analyze(new_commits, source)
+ self.repos.append(repo_dir)
+ log_output = git_log(repo_dir)
+ new_commits = [
+ c for c in (Commit.parse(line) for line in log_output.splitlines() if line)
+ if c is not None
+ ]
+ self.commits.extend(new_commits)
+ self._analyze(new_commits, source)
@property
- def sorted_persons(self):
- return sorted(self.persons.items(), key=lambda p: p[1].as_author + p[1].as_committer)
-
- def resolve_persons(self):
- threads = []
- for _, person in self.persons.items():
- if person.email in SYSTEM_EMAILS:
- continue
- # TODO: optimize
- thread = Thread(target=self.git.get_verified_username, args=(person.repo_url, person.commit, person))
- thread.start()
- threads.append(thread)
-
- for thread in threads:
- thread.join()
-
- def analyze(self, new_commits, repo_url):
- # save all author and committers as unique persons
- for commit in new_commits:
- # author saving
- person = self.persons.get(commit.author, Person(commit.author))
- person.name = commit.author_name
- person.email = commit.author_email
- person.as_author += 1
- person.repo_url = repo_url
- person.commit = commit
- self.persons[commit.author] = person
-
- # committer saving
- person = self.persons.get(commit.committer, Person(commit.committer))
- person.name = commit.committer_name
- person.email = commit.committer_email
- person.as_committer += 1
- person.repo_url = repo_url
- person.commit = commit
- self.persons[commit.committer] = person
-
- # make persons graph links based on author/committer mismatch
- for commit in new_commits:
- if not commit.author_committer_same:
- self.persons[commit.author].also_known[commit.committer] = self.persons[commit.committer]
- self.persons[commit.committer].also_known[commit.author] = self.persons[commit.author]
-
- # TODO: probabilistic graph links based on same names/emails and Levenshtein distance
- # just checking same names now
+ def sorted_persons(self) -> list[tuple[str, Person]]:
+ return sorted(
+ self.persons.items(),
+ key=lambda item: item[1].as_author + item[1].as_committer,
+ )
+ def resolve_persons(self) -> None:
+ targets = [
+ p for p in self.persons.values()
+ if p.email not in SYSTEM_EMAILS and p.repo_url and p.last_commit_hash
+ ]
+ if not targets:
+ return
+ with ThreadPoolExecutor(max_workers=RESOLVE_WORKERS) as pool:
+ futures = {
+ pool.submit(resolve_github_username, p.repo_url, p.last_commit_hash): p
+ for p in targets
+ }
+ for fut, person in futures.items():
+ login = fut.result()
+ if login:
+ person.github_login = login
+
+ def _upsert(
+ self, key: str, name: str, email: str, repo_url: str, commit_hash: str,
+ ) -> Person:
+ person = self.persons.get(key) or Person(key=key)
+ person.name = name
+ person.email = email
+ person.repo_url = repo_url
+ person.last_commit_hash = commit_hash
+ self.persons[key] = person
+ return person
+
+ def _analyze(self, new_commits: Iterable[Commit], repo_url: str) -> None:
for commit in new_commits:
- author_emails = self.names.get(commit.author_name, set())
- author_emails.add(commit.author_email)
- self.names[commit.author_name] = author_emails
-
- committer_emails = self.names.get(commit.committer_name, set())
- committer_emails.add(commit.committer_email)
- self.names[commit.committer_name] = committer_emails
-
- for emails_set in self.names.values():
- names = [name for name, v in self.names.items() if v == emails_set]
- key = ','.join(sorted(names))
- if len(names) > 1 and key not in self.same_emails_persons:
- self.same_emails_persons[key] = (names, emails_set)
-
- return self.sorted_persons
-
- def __str__(self):
- result = 'Analyze of the git repo(s) "{}"'.format(', '.join(self.repos))
+ author = self._upsert(
+ commit.author, commit.author_name, commit.author_email,
+ repo_url, commit.hash,
+ )
+ author.as_author += 1
- result += '\nVerbose persons info:\n'
- for name, person in self.sorted_persons:
- result += ("{}\n{}\n".format(DELIMITER, person))
+ committer = self._upsert(
+ commit.committer, commit.committer_name, commit.committer_email,
+ repo_url, commit.hash,
+ )
+ committer.as_committer += 1
- matching_result = ''
- for name, emails in self.names.items():
+ if not commit.author_committer_same:
+ author.also_known[commit.committer] = committer
+ committer.also_known[commit.author] = author
+
+ self.name_to_emails[commit.author_name].add(commit.author_email)
+ self.name_to_emails[commit.committer_name].add(commit.committer_email)
+
+ # Group names that share the exact same set of emails — these are
+ # treated as the same person. O(n) instead of the previous O(n²).
+ emails_to_names: dict[frozenset[str], list[str]] = defaultdict(list)
+ for name, emails in self.name_to_emails.items():
+ emails_to_names[frozenset(emails)].append(name)
+ self.same_emails_persons = {
+ ",".join(sorted(names)): (sorted(names), set(emails))
+ for emails, names in emails_to_names.items()
+ if len(names) > 1
+ }
+
+ def __str__(self) -> str:
+ parts: list[str] = [
+ f'Analyze of the git repo(s) "{", ".join(self.repos)}"',
+ "",
+ "Verbose persons info:",
+ ]
+ for _, person in self.sorted_persons:
+ parts.append(DELIMITER)
+ parts.append(str(person))
+
+ matching: list[str] = []
+ for name, emails in self.name_to_emails.items():
if len(emails) > 1:
- matching_result += '\n{} is the owner of emails:\n\t\t\t{}\n'.format(name, '\n\t\t\t'.join(emails))
-
- if matching_result:
- result += '\nMatching info:\n{}{}'.format(DELIMITER, matching_result)
-
- for names, emails in self.same_emails_persons.values():
- result += '\n{} are the same person\n'.format(' and '.join(names))
-
- result += '\nStatistics info:\n{}'.format(DELIMITER)
- result += '\nTotal persons: {}'.format(len(self.persons))
-
- return result
-
-
-def main():
- parser = argparse.ArgumentParser(description='Extract accounts\' information from git repo and make some researches.')
- parser.add_argument('-d', '--dir', help='directory with git project(s)')
- parser.add_argument('-u', '--url', help='url of git repo')
- parser.add_argument('--github', action='store_true', help='try to extract extended info from GitHub')
- parser.add_argument('--nickname', type=str, help='try to download repos from all platforms by nickname')
- parser.add_argument('-r', '--recursive', action='store_true', help='recursive directory processing')
- parser.add_argument('--debug', action='store_true', help='print debug information')
- # TODO: clone repos as bare
- # TODO: allow forks
-
- args = parser.parse_args()
- log_level = logging.INFO if not args.debug else logging.DEBUG
- logging.basicConfig(level=log_level, format='-'*40 + '\n%(levelname)s: %(message)s')
-
- analyst = None
-
- analyst = GitAnalyst()
- repos = []
-
- repos.append(args.url)
- repos.append(args.dir and args.dir.rstrip('/'))
-
- if args.recursive and args.dir:
- dirs = find_all_repos_recursively(args.dir)
- repos += dirs
-
+ emails_block = "\n\t\t\t".join(sorted(emails))
+ matching.append(
+ f"\n{name} is the owner of emails:\n\t\t\t{emails_block}"
+ )
+ if matching:
+ parts.append("")
+ parts.append("Matching info:")
+ parts.append(DELIMITER + "".join(matching))
+
+ for names, _ in self.same_emails_persons.values():
+ parts.append(f"\n{' and '.join(names)} are the same person")
+
+ parts.append("")
+ parts.append("Statistics info:")
+ parts.append(DELIMITER)
+ parts.append(f"Total persons: {len(self.persons)}")
+ return "\n".join(parts)
+
+
+# ---------- CLI ----------
+
+def _parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser(
+ description="Extract accounts' information from git repo and make some researches.",
+ )
+ parser.add_argument("-d", "--dir", help="directory with git project(s)")
+ parser.add_argument("-u", "--url", help="url of git repo")
+ parser.add_argument(
+ "--github", action="store_true",
+ help="try to extract extended info from GitHub",
+ )
+ parser.add_argument(
+ "--nickname", type=str,
+ help="download repos from GitHub by nickname",
+ )
+ parser.add_argument(
+ "-r", "--recursive", action="store_true",
+ help="recursive directory processing",
+ )
+ parser.add_argument(
+ "--repos-dir", default=DEFAULT_REPOS_DIR,
+ help=f"directory to clone remote repositories into (default: {DEFAULT_REPOS_DIR})",
+ )
+ parser.add_argument("--debug", action="store_true", help="print debug information")
+ return parser.parse_args()
+
+
+def _collect_sources(args: argparse.Namespace) -> list[str]:
+ sources: list[str] = []
+ if args.url:
+ sources.append(args.url)
+ if args.dir:
+ sources.append(args.dir.rstrip("/"))
+ if args.recursive:
+ sources.extend(find_all_repos_recursively(args.dir))
if args.nickname:
- repos_count = get_public_repos_count(args.nickname)
- if repos_count:
- print('found', repos_count, 'repos')
- repos += get_github_repos(args.nickname, repos_count=repos_count)
-
- for repo in repos:
- analyst.append(source=repo)
-
- logging.info('Resolving GitHub usernames, please wait...')
+ count = get_public_repos_count(args.nickname)
+ if count:
+ print(f"found {count} repos")
+ sources.extend(get_github_repos(args.nickname, repos_count=count))
+ return sources
+
+
+def main() -> None:
+ args = _parse_args()
+ logging.basicConfig(
+ level=logging.DEBUG if args.debug else logging.INFO,
+ format="-" * 40 + "\n%(levelname)s: %(message)s",
+ )
+
+ sources = _collect_sources(args)
+ if not sources:
+ print("Run me with git repo link or path!")
+ return
+
+ analyst = GitAnalyst(repos_dir=args.repos_dir)
+ for source in sources:
+ analyst.append(source)
+
+ logger.info("Resolving GitHub usernames, please wait...")
analyst.resolve_persons()
if analyst.repos:
print(analyst)
else:
- print('Run me with git repo link or path!')
+ print("Run me with git repo link or path!")
-if __name__ == '__main__':
+if __name__ == "__main__":
main()
diff --git a/test_gitcolombo.py b/test_gitcolombo.py
new file mode 100644
index 0000000..3ddc696
--- /dev/null
+++ b/test_gitcolombo.py
@@ -0,0 +1,460 @@
+"""Tests for gitcolombo.
+
+Each test class is wired to one of the bugs / regressions documented in the
+refactor: invalid commit lines, missing Person attributes, unsafe subprocess
+calls, .git-suffix handling, recursive repo lookup, HTTP timeouts, O(n^2)
+identity matching, None sources, system-email skipping, and pagination math.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+import tempfile
+import types
+import unittest
+from io import BytesIO
+from unittest import mock
+
+import gitcolombo as gc
+
+
+# ---------- Commit parsing ----------
+
+class TestCommitParse(unittest.TestCase):
+ def test_valid_line(self):
+ line = 'abc123;"Alice alice@example.com";"Bob bob@example.com"'
+ c = gc.Commit.parse(line)
+ self.assertIsNotNone(c)
+ self.assertEqual(c.hash, "abc123")
+ self.assertEqual(c.author_name, "Alice")
+ self.assertEqual(c.author_email, "alice@example.com")
+ self.assertEqual(c.committer_name, "Bob")
+ self.assertEqual(c.committer_email, "bob@example.com")
+
+ def test_invalid_line_returns_none(self):
+ # Regression: original Commit.__init__ left attributes unset on bad
+ # input and crashed downstream with AttributeError.
+ self.assertIsNone(gc.Commit.parse("garbage"))
+ self.assertIsNone(gc.Commit.parse(""))
+
+ def test_author_committer_same(self):
+ line = 'h;"Alice alice@x.io";"Alice alice@x.io"'
+ c = gc.Commit.parse(line)
+ self.assertTrue(c.author_committer_same)
+
+ def test_author_committer_different(self):
+ line = 'h;"Alice alice@x.io";"Alice bob@x.io"'
+ c = gc.Commit.parse(line)
+ self.assertFalse(c.author_committer_same)
+
+ def test_split_name_email_invalid(self):
+ # Garbage in must not crash, just return empty strings.
+ self.assertEqual(gc._split_name_email("no-space-here"), ("", ""))
+
+
+# ---------- Person dataclass ----------
+
+class TestPerson(unittest.TestCase):
+ def test_default_attributes_present(self):
+ # Regression: original Person did not initialize repo_url / commit /
+ # github_link in __init__, so accessing them raised AttributeError.
+ p = gc.Person(key="Alice alice@x.io")
+ self.assertIsNone(p.repo_url)
+ self.assertIsNone(p.last_commit_hash)
+ self.assertIsNone(p.github_login)
+ self.assertEqual(p.also_known, {})
+ self.assertEqual(p.as_author, 0)
+ self.assertEqual(p.as_committer, 0)
+
+ def test_str_minimal(self):
+ p = gc.Person(key="x", name="Alice", email="a@x.io")
+ s = str(p)
+ self.assertIn("Alice", s)
+ self.assertIn("a@x.io", s)
+
+
+# ---------- Clone target directory normalization ----------
+
+class TestCloneTargetDir(unittest.TestCase):
+ def test_plain_url(self):
+ self.assertEqual(
+ gc._clone_target_dir("https://github.com/user/repo"), "repo",
+ )
+
+ def test_dot_git_suffix(self):
+ # Regression: original code used source.split('/')[-1] verbatim and
+ # ended up looking for a directory named "repo.git" that git clone
+ # never created.
+ self.assertEqual(
+ gc._clone_target_dir("https://github.com/user/repo.git"), "repo",
+ )
+
+ def test_trailing_slash(self):
+ self.assertEqual(
+ gc._clone_target_dir("https://github.com/user/repo/"), "repo",
+ )
+
+ def test_dot_git_with_trailing_slash(self):
+ self.assertEqual(
+ gc._clone_target_dir("https://github.com/user/repo.git/"), "repo",
+ )
+
+
+# ---------- Subprocess safety (no shell=True) ----------
+
+class TestSubprocessSafety(unittest.TestCase):
+ def test_git_log_uses_argv_not_shell(self):
+ # Regression: original used Popen(cmd, shell=True) — command
+ # injection if a malicious path/URL were ever passed.
+ with mock.patch("gitcolombo.subprocess.run") as run:
+ run.return_value = types.SimpleNamespace(
+ returncode=0, stdout="", stderr="",
+ )
+ gc.git_log("/tmp/repo")
+ args, kwargs = run.call_args
+ cmd = args[0]
+ self.assertIsInstance(cmd, list)
+ self.assertEqual(cmd[0], "git")
+ self.assertEqual(cmd[1], "log")
+ self.assertNotIn("shell", kwargs) # default is False
+
+ def test_git_clone_uses_argv_not_shell(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ with mock.patch("gitcolombo.subprocess.run") as run:
+ run.return_value = types.SimpleNamespace(
+ returncode=0, stdout="", stderr="",
+ )
+ url = "https://example.com/repo.git"
+ gc.git_clone(url, tmp)
+ args, kwargs = run.call_args
+ self.assertEqual(
+ args[0],
+ ["git", "clone", url, os.path.join(tmp, "repo")],
+ )
+ self.assertNotIn("shell", kwargs)
+
+ def test_git_log_returns_empty_when_binary_missing(self):
+ with mock.patch("gitcolombo.subprocess.run", side_effect=FileNotFoundError):
+ self.assertEqual(gc.git_log("/tmp/whatever"), "")
+
+ def test_git_clone_returns_none_on_failure(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ with mock.patch("gitcolombo.subprocess.run") as run:
+ run.return_value = types.SimpleNamespace(
+ returncode=128, stdout="", stderr="fatal",
+ )
+ self.assertIsNone(gc.git_clone("https://x/y", tmp))
+
+ def test_git_clone_creates_dest_dir(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ dest = os.path.join(tmp, "deep", "nested")
+ with mock.patch("gitcolombo.subprocess.run") as run:
+ run.return_value = types.SimpleNamespace(
+ returncode=0, stdout="", stderr="",
+ )
+ gc.git_clone("https://x/repo", dest)
+ self.assertTrue(os.path.isdir(dest))
+
+
+# ---------- Recursive repo discovery ----------
+
+class TestFindAllReposRecursively(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp()
+ # /tmp/root/a/.git, /tmp/root/b/c/.git, /tmp/root/d (not a repo)
+ os.makedirs(os.path.join(self.tmp, "a", ".git"))
+ os.makedirs(os.path.join(self.tmp, "b", "c", ".git"))
+ os.makedirs(os.path.join(self.tmp, "d"))
+
+ def tearDown(self):
+ shutil.rmtree(self.tmp, ignore_errors=True)
+
+ def test_returns_repo_roots_not_dot_git(self):
+ # Regression: original returned `.git` dirs themselves.
+ repos = gc.find_all_repos_recursively(self.tmp)
+ self.assertIn(os.path.join(self.tmp, "a"), repos)
+ self.assertIn(os.path.join(self.tmp, "b", "c"), repos)
+ for r in repos:
+ self.assertFalse(r.endswith(".git"))
+
+ def test_does_not_descend_into_dot_git(self):
+ # If .git itself contained another .git (it doesn't normally), we
+ # shouldn't report it. Create a decoy.
+ os.makedirs(os.path.join(self.tmp, "a", ".git", "nested", ".git"))
+ repos = gc.find_all_repos_recursively(self.tmp)
+ for r in repos:
+ self.assertNotIn(os.sep + ".git" + os.sep, r + os.sep)
+
+
+# ---------- HTTP timeout ----------
+
+class TestHttpTimeout(unittest.TestCase):
+ def test_http_get_passes_timeout(self):
+ # Regression: original urlopen calls had no timeout — could hang.
+ fake_resp = mock.MagicMock()
+ fake_resp.__enter__.return_value.read.return_value = b"ok"
+ fake_resp.__exit__.return_value = False
+ with mock.patch("gitcolombo.urllib.request.urlopen", return_value=fake_resp) as up:
+ gc._http_get("https://example.com/x")
+ _, kwargs = up.call_args
+ self.assertIn("timeout", kwargs)
+ self.assertEqual(kwargs["timeout"], gc.HTTP_TIMEOUT)
+
+ def test_http_get_returns_none_on_error(self):
+ import urllib.error
+ with mock.patch(
+ "gitcolombo.urllib.request.urlopen",
+ side_effect=urllib.error.URLError("boom"),
+ ):
+ self.assertIsNone(gc._http_get("https://example.com/x"))
+
+
+# ---------- GitHub pagination math ----------
+
+class TestGitHubPagination(unittest.TestCase):
+ def test_zero_repos(self):
+ # Regression: get_github_repos used to compute pagination even for
+ # count=0/None, doing one wasted request.
+ with mock.patch("gitcolombo._http_get_json") as get:
+ result = gc.get_github_repos("nobody", repos_count=0)
+ self.assertEqual(result, set())
+ get.assert_not_called()
+
+ def test_exact_page_boundary(self):
+ # 100 repos == exactly one page, not two.
+ with mock.patch("gitcolombo._http_get_json", return_value=[]) as get:
+ gc.get_github_repos("u", repos_count=100)
+ self.assertEqual(get.call_count, 1)
+
+ def test_two_pages(self):
+ with mock.patch("gitcolombo._http_get_json", return_value=[]) as get:
+ gc.get_github_repos("u", repos_count=150)
+ self.assertEqual(get.call_count, 2)
+
+ def test_forks_excluded_by_default(self):
+ page = [
+ {"html_url": "https://github.com/u/r1", "fork": False},
+ {"html_url": "https://github.com/u/r2", "fork": True},
+ ]
+ with mock.patch("gitcolombo._http_get_json", return_value=page):
+ result = gc.get_github_repos("u", repos_count=2)
+ self.assertEqual(result, {"https://github.com/u/r1"})
+
+ def test_forks_included_when_requested(self):
+ page = [
+ {"html_url": "https://github.com/u/r1", "fork": False},
+ {"html_url": "https://github.com/u/r2", "fork": True},
+ ]
+ with mock.patch("gitcolombo._http_get_json", return_value=page):
+ result = gc.get_github_repos("u", repos_count=2, include_forks=True)
+ self.assertEqual(result, {"https://github.com/u/r1", "https://github.com/u/r2"})
+
+ def test_public_repos_count_handles_missing(self):
+ with mock.patch("gitcolombo._http_get_json", return_value=None):
+ self.assertEqual(gc.get_public_repos_count("ghost"), 0)
+
+
+# ---------- Identity matching (same_emails_persons) ----------
+
+class TestSameEmailsPersons(unittest.TestCase):
+ def _commit(self, h, a_name, a_email, c_name=None, c_email=None):
+ c_name = c_name or a_name
+ c_email = c_email or a_email
+ line = f'{h};"{a_name} {a_email}";"{c_name} {c_email}"'
+ return gc.Commit.parse(line)
+
+ def test_two_names_share_email_set(self):
+ # Two names with the EXACT same email set must be linked as the
+ # same person. This used to be O(n^2) and rebuilt every call.
+ analyst = gc.GitAnalyst()
+ commits = [
+ self._commit("h1", "Alice", "shared@x.io"),
+ self._commit("h2", "AliceB", "shared@x.io"),
+ ]
+ analyst._analyze(commits, "https://example.com/r")
+ self.assertEqual(len(analyst.same_emails_persons), 1)
+ names, emails = next(iter(analyst.same_emails_persons.values()))
+ self.assertEqual(sorted(names), ["Alice", "AliceB"])
+ self.assertEqual(emails, {"shared@x.io"})
+
+ def test_disjoint_email_sets_not_linked(self):
+ analyst = gc.GitAnalyst()
+ commits = [
+ self._commit("h1", "Alice", "a@x.io"),
+ self._commit("h2", "Bob", "b@x.io"),
+ ]
+ analyst._analyze(commits, "r")
+ self.assertEqual(analyst.same_emails_persons, {})
+
+ def test_one_name_many_emails_reported_in_matching(self):
+ analyst = gc.GitAnalyst()
+ commits = [
+ self._commit("h1", "Alice", "a1@x.io"),
+ self._commit("h2", "Alice", "a2@x.io"),
+ ]
+ analyst._analyze(commits, "r")
+ self.assertEqual(analyst.name_to_emails["Alice"], {"a1@x.io", "a2@x.io"})
+
+ def test_author_committer_link(self):
+ # author != committer → must show up in each other's also_known.
+ analyst = gc.GitAnalyst()
+ commits = [self._commit("h1", "Alice", "a@x.io", "Bob", "b@x.io")]
+ analyst._analyze(commits, "r")
+ alice_key = "Alice a@x.io"
+ bob_key = "Bob b@x.io"
+ self.assertIn(bob_key, analyst.persons[alice_key].also_known)
+ self.assertIn(alice_key, analyst.persons[bob_key].also_known)
+
+
+# ---------- CLI source collection ----------
+
+class TestCollectSources(unittest.TestCase):
+ def _args(self, **kw):
+ defaults = dict(url=None, dir=None, recursive=False, nickname=None, github=False, debug=False)
+ defaults.update(kw)
+ return types.SimpleNamespace(**defaults)
+
+ def test_no_inputs_returns_empty(self):
+ # Regression: original main appended None into repos when -u/-d were
+ # absent and silently swallowed them inside append().
+ self.assertEqual(gc._collect_sources(self._args()), [])
+
+ def test_url_only(self):
+ self.assertEqual(
+ gc._collect_sources(self._args(url="https://x/y")),
+ ["https://x/y"],
+ )
+
+ def test_dir_strips_trailing_slash(self):
+ self.assertEqual(
+ gc._collect_sources(self._args(dir="/tmp/r/")),
+ ["/tmp/r"],
+ )
+
+ def test_default_repos_dir(self):
+ self.assertEqual(gc.GitAnalyst().repos_dir, gc.DEFAULT_REPOS_DIR)
+ self.assertEqual(gc.DEFAULT_REPOS_DIR, "repos")
+
+ def test_custom_repos_dir(self):
+ analyst = gc.GitAnalyst(repos_dir="/tmp/custom")
+ self.assertEqual(analyst.repos_dir, "/tmp/custom")
+
+ def test_append_uses_configured_repos_dir(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ analyst = gc.GitAnalyst(repos_dir=tmp)
+ with mock.patch("gitcolombo.subprocess.run") as run:
+ run.return_value = types.SimpleNamespace(
+ returncode=0, stdout="", stderr="",
+ )
+ analyst.append("https://example.com/u/repo.git")
+ clone_call = run.call_args_list[0]
+ self.assertEqual(
+ clone_call.args[0],
+ ["git", "clone", "https://example.com/u/repo.git",
+ os.path.join(tmp, "repo")],
+ )
+
+ def test_recursive_includes_discovered(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ os.makedirs(os.path.join(tmp, "a", ".git"))
+ sources = gc._collect_sources(self._args(dir=tmp, recursive=True))
+ self.assertIn(tmp, sources)
+ self.assertIn(os.path.join(tmp, "a"), sources)
+
+
+# ---------- GitHub login resolution ----------
+
+class TestResolvePersons(unittest.TestCase):
+ def test_system_emails_skipped(self):
+ analyst = gc.GitAnalyst()
+ p = gc.Person(
+ key="bot noreply@github.com",
+ name="bot", email="noreply@github.com",
+ repo_url="https://github.com/u/r", last_commit_hash="abc",
+ )
+ analyst.persons[p.key] = p
+ with mock.patch("gitcolombo.resolve_github_username") as rg:
+ analyst.resolve_persons()
+ rg.assert_not_called()
+
+ def test_non_github_url_skipped(self):
+ with mock.patch("gitcolombo._http_get") as g:
+ result = gc.resolve_github_username("https://gitlab.com/u/r", "abc")
+ self.assertIsNone(result)
+ g.assert_not_called()
+
+ def test_github_login_extracted(self):
+ html = b'Alice'
+ with mock.patch("gitcolombo._http_get", return_value=html):
+ self.assertEqual(
+ gc.resolve_github_username("https://github.com/u/r", "abc"),
+ "alice42",
+ )
+
+ def test_assigns_login_to_person(self):
+ analyst = gc.GitAnalyst()
+ p = gc.Person(
+ key="Alice a@x.io", name="Alice", email="a@x.io",
+ repo_url="https://github.com/u/r", last_commit_hash="abc",
+ )
+ analyst.persons[p.key] = p
+ with mock.patch("gitcolombo.resolve_github_username", return_value="alice42"):
+ analyst.resolve_persons()
+ self.assertEqual(p.github_login, "alice42")
+
+
+# ---------- End-to-end smoke test with a real repo ----------
+
+@unittest.skipIf(shutil.which("git") is None, "git binary not available")
+class TestEndToEnd(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp()
+ env = os.environ.copy()
+ env.update({
+ "GIT_AUTHOR_NAME": "Alice",
+ "GIT_AUTHOR_EMAIL": "alice@example.com",
+ "GIT_COMMITTER_NAME": "Bob",
+ "GIT_COMMITTER_EMAIL": "bob@example.com",
+ })
+ subprocess.run(["git", "init", "-q", self.tmp], check=True)
+ subprocess.run(
+ ["git", "-C", self.tmp, "config", "commit.gpgsign", "false"],
+ check=True,
+ )
+ with open(os.path.join(self.tmp, "f"), "w") as f:
+ f.write("hi")
+ subprocess.run(["git", "-C", self.tmp, "add", "f"], check=True, env=env)
+ subprocess.run(
+ ["git", "-C", self.tmp, "commit", "-q", "-m", "first"],
+ check=True, env=env,
+ )
+
+ def tearDown(self):
+ shutil.rmtree(self.tmp, ignore_errors=True)
+
+ def test_analyst_collects_author_and_committer(self):
+ analyst = gc.GitAnalyst()
+ analyst.append(self.tmp)
+ self.assertEqual(len(analyst.commits), 1)
+ # Two distinct persons: author and committer.
+ names = {p.name for p in analyst.persons.values()}
+ self.assertEqual(names, {"Alice", "Bob"})
+ # author != committer → mutual also_known link.
+ alice = next(p for p in analyst.persons.values() if p.name == "Alice")
+ bob = next(p for p in analyst.persons.values() if p.name == "Bob")
+ self.assertIn(bob.key, alice.also_known)
+ self.assertIn(alice.key, bob.also_known)
+
+ def test_str_render_contains_summary(self):
+ analyst = gc.GitAnalyst()
+ analyst.append(self.tmp)
+ rendered = str(analyst)
+ self.assertIn("Verbose persons info", rendered)
+ self.assertIn("Total persons: 2", rendered)
+ self.assertIn("alice@example.com", rendered)
+ self.assertIn("bob@example.com", rendered)
+
+
+if __name__ == "__main__":
+ unittest.main()