167 lines
4.3 KiB
Python
167 lines
4.3 KiB
Python
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")
|
|
with r.config_writer("repository") as cw:
|
|
cw.add_section("user")
|
|
cw.add_value("user", "name", "Tester")
|
|
cw.add_value("user", "email", "test@a.b")
|
|
|
|
r.git.commit(message = "Test Commit")
|
|
hash = r.commit("master").hexsha
|
|
|
|
assert rt.fetch()
|
|
assert rt._repo.commit(hash)
|