repo-cloner/tests/lib/test_repo_tool.py

import logging

import git
import pytest
from cloner_test_fixtures import support_data_path
from pathlib import Path
from repo_cloner.lib.repo_tool import RepoTool


def test_init(support_data_path: Path):
	test_repos_path = support_data_path.joinpath("tool_repos")
	# test on non-existent dir
	rt = RepoTool(test_repos_path.joinpath("nonexistent.git").as_posix())
	assert not rt.initialized

	# uninitialized, but existing repo
	rt = RepoTool(test_repos_path.joinpath("uninitialized.git").as_posix())
	assert not rt.initialized
	assert not rt.bare

	# initialized, existing repo
	rt = RepoTool(test_repos_path.joinpath("initialized.git").as_posix())
	assert rt.initialized
	assert rt.bare

	rt = RepoTool(test_repos_path.joinpath("non-bare-init").as_posix())
	assert rt.initialized
	assert not rt.bare


def test_initialized(tmp_path, monkeypatch):
	rt = RepoTool(tmp_path.as_posix())
	monkeypatch.setattr(rt, "_initialized", False)
	assert not rt.initialized
	monkeypatch.setattr(rt, "_initialized", True)
	assert rt.initialized


def test_bare(tmp_path, monkeypatch):
	rt = RepoTool(tmp_path.as_posix())
	monkeypatch.setattr(rt, "_bare", False)
	assert not rt.bare
	monkeypatch.setattr(rt, "_bare", True)
	assert rt.bare


def test_path(tmp_path, monkeypatch):
	rt = RepoTool(tmp_path)
	assert tmp_path.as_posix() == rt.path
	monkeypatch.setattr(rt, "_path", "/tmp")
	assert "/tmp" == rt.path


def test_clone_initialized_repo(tmp_path, caplog, support_data_path):
	from git import Repo
	# initialize repo
	Repo().init(tmp_path, bare = True)
	rt = RepoTool(tmp_path.as_posix())
	# check it
	assert rt.initialized
	assert rt.bare

	# try clone
	test_repo = support_data_path.joinpath("test-repo-base").as_uri()
	assert not rt.clone(test_repo)

	r = caplog.records[0]
	assert r.levelname == "WARNING"
	assert r.message == "Trying to clone to initialized repository!"


def test_check_initialized(tmp_path, monkeypatch, caplog):
	# hard testing directly, testing via fetch
	rt = RepoTool(tmp_path.as_posix())
	assert not rt.initialized
	assert not rt.fetch()
	assert caplog.records[1].levelname == "CRITICAL"
	assert caplog.records[1].message == f"Repo {tmp_path.as_posix()} is not initialized!"


def test_clone_okay(tmp_path, caplog, support_data_path):
	rt = RepoTool(tmp_path.as_posix())
	assert not rt.initialized

	caplog.set_level(logging.INFO)

	# try clone
	test_repo = support_data_path.joinpath("test-repo-base").as_uri()
	assert rt.clone(test_repo)

	# warning about uninit repo
	assert caplog.records[0].levelname == "WARNING"
	# info cloning
	assert caplog.records[1].levelname == "INFO"
	assert "Cloning repository from url: file:///" in caplog.records[1].message

	# progress states
	counting_cnt = 0
	compressing_cnt = 0
	receiving_cnt = 0
	resolving_cnt = 0

	for x in range(1, len(caplog.records)):
		rec = caplog.records[x]
		assert rec.levelname == "INFO"
		if "GIT COUNTING" in rec.message:
			counting_cnt += 1
		if "GIT COMPRESSING" in rec.message:
			compressing_cnt += 1
		if "GIT RECEIVING" in rec.message:
			receiving_cnt += 1
		if "GIT RESOLVING" in rec.message:
			resolving_cnt += 1

	assert counting_cnt >= 2
	assert compressing_cnt >= 2
	assert receiving_cnt >= 2
	assert resolving_cnt >= 2


def test_fetch_uninitialized(tmp_path, caplog):
	repo_a = tmp_path.joinpath("A.git")
	repo_b = tmp_path.joinpath("B.git")

	repo_a.mkdir()
	rt = RepoTool(repo_a.as_posix())
	assert not rt.initialized
	assert not rt.fetch()

	assert f"Repo {repo_a.as_posix()} is not initialized!"


def test_fetch_no_remotes(tmp_path, caplog):
	repo_a = tmp_path.joinpath("A.git")

	git.Repo().init(repo_a.as_posix())

	rt = RepoTool(repo_a.as_posix())
	assert not rt.fetch()
	assert caplog.records[0].levelname == "WARNING"
	assert caplog.records[0].message == f"Repo: {repo_a.as_posix()} does not contain any remotes!"


def test_fetch_ok(tmp_path):
	repo_a = tmp_path.joinpath("A.git")
	repo_b = tmp_path.joinpath("B.git")

	r = git.Repo().init(repo_a.as_posix())

	rt = RepoTool(repo_b.as_posix())
	assert rt.clone(repo_a.as_uri())

	# new file
	repo_a.joinpath(".gitignore").touch()
	# commit it
	r.git.add(".gitignore")
	r.git.commit(message = "Test Commit", author = "Tester <test@a.b>")
	hash = r.commit("master").hexsha

	assert rt.fetch()
	assert rt._repo.commit(hash)