repo-cloner/tests/lib/test_repo_tool.py

import logging

import git
import pytest
from cloner_test_fixtures import support_data_path, path_repo_base, path_repo_changed_branches, \
	path_repo_different_tags, path_repo_new_commits, path_repo_reduced
from pathlib import Path
from repo_cloner.lib.repo_tool import RepoTool

base_repo_branches = [
	('br0', '1fe320fe278a40bfd6813860d9952b03caf5e2ea'),
	('branch-a', '71ce433cd98f899c723184b9ea1e0bcdf97d5c1a'),
	('branch-d', '5551c9980c9ff2b94865596df35e38527d9fcfbd'),
	(
		'dependabot/npm_and_yarn/dot-github/actions/test-action/node-fetch-2.6.7',
		'8298c84342db5cb93972426188f06431fc498681'),
	('master', 'e0c7e2a72579e24657c05e875201011d2b48bf94'),
	('orph', '3de11d7640ed17f93ce38d31c9dea0fd65774077'),
	('pr-1', '378d1f9e27dbacfcad0ee93355f0305c6c4432f1'),
	('pr-3', 'e2bf0be1b64f345077fa96671d1c27f0c3476ac4'),
	('release/1.2.3', 'f5eee554aecbee6f2f8e18de089476c3bf957ddc'),
	('release/4.5.6', 'be3cb4b2590091dbaf5ce0d861da0f7979c5fc23'),
	('release/7.7.10', '23478a09ad05642d64f44920f3247f7708d4bcae'),
	('release/7.7.11', 'c014e918175915e07eef0150984a24ff8c33e154'),
	('release/7.7.12', 'a7c8068c7c0a45f842400ffb2c4a245ab2f7a388'),
	('release/7.7.13', 'c49c48572f1422e8b222db854d7ecbeba27b8020'),
	('release/7.7.7', 'da67fad037eea0c065a5ea52a894d5522eb9f248'),
	('release/7.7.8', '85003adc6bc0e91391edd97efb7f515cbd6d303e'),
	('release/7.7.9', '73aaf9a4377bab721ba716776e81dc76bec25beb'),
	('symlink-readme', '8a2844f55fc1adde2e346d0999ef3e12aafbe511'),
	('temp', '72cab056e6a95fbb4e7ea3f0e1644a7eef345a2d'),
	('v1/foo/bar', '6e3325f7dc37f67b48e719de1ce4a4c9d33d1042'),
	('v1/nix', 'c55e6953ff343f9eeb42dc04eb6aa5dfed6f0756')]

base_repo_tags = [
	('meh', '703de123eaa9832f4ce15baf6b702cd7f72fea5a'),
	('v', '85003adc6bc0e91391edd97efb7f515cbd6d303e'), ('v1.2.3', '8f3d75c010a30babe3ee8a4176bb16136c5949b3'),
	('v4.5.6', '8a2844f55fc1adde2e346d0999ef3e12aafbe511'), ('v7.7.10', '23478a09ad05642d64f44920f3247f7708d4bcae'),
	('v7.7.11', 'c014e918175915e07eef0150984a24ff8c33e154'), ('v7.7.12', 'a7c8068c7c0a45f842400ffb2c4a245ab2f7a388'),
	('v7.7.13', 'c49c48572f1422e8b222db854d7ecbeba27b8020'), ('v7.7.9', 'dd62ea5560a6944a04c1f8fbdbbb09c01f1b4a21'),
	('wip-001', 'adbb090758d7bc201a35b5a7a470a3c01a3e5a95'), ('wip-002', 'a05c54d8623558963b8774bb01962fe0683c46bf'),
	('wip-003', 'd3e906333be8a6ff09f96e827199d51694a16dc1'), ('wip-r/1', '030cc2b97d8ecbf07b7bd361105f07e631393f00')]


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_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)


@pytest.fixture
def cloned_base_repo_obj(tmp_path: Path, path_repo_base: Path) -> RepoTool:
	rt = RepoTool(tmp_path.joinpath("repo.git").as_posix())
	assert rt.clone(path_repo_base.as_uri())
	return rt


def prepare_collection_list(collection) -> list:
	result = []
	for item in collection:
		result.append((item.name, str(item.commit)))
	result.sort()
	return result


def test_fetch_changed_branches(cloned_base_repo_obj: RepoTool, path_repo_changed_branches: Path):
	cloned_base_repo_obj._repo.remotes[0].set_url(path_repo_changed_branches.as_uri())
	branches = prepare_collection_list(cloned_base_repo_obj._repo.branches)
	assert branches == base_repo_branches

	tags = prepare_collection_list(cloned_base_repo_obj._repo.tags)
	assert tags == base_repo_tags

	assert cloned_base_repo_obj.fetch()

	branches = prepare_collection_list(cloned_base_repo_obj._repo.branches)
	assert branches == [
		('branch-a', '71ce433cd98f899c723184b9ea1e0bcdf97d5c1a'),
		('branch-d', '5551c9980c9ff2b94865596df35e38527d9fcfbd'),
		(
			'dependabot/npm_and_yarn/dot-github/actions/test-action/node-fetch-2.6.7',
			'8298c84342db5cb93972426188f06431fc498681'),
		('master', 'e0c7e2a72579e24657c05e875201011d2b48bf94'),
		('orph', '3de11d7640ed17f93ce38d31c9dea0fd65774077'),
		('pr-1', '378d1f9e27dbacfcad0ee93355f0305c6c4432f1'),
		('pr-3', 'e2bf0be1b64f345077fa96671d1c27f0c3476ac4'),
		('release/1.2.3', 'f5eee554aecbee6f2f8e18de089476c3bf957ddc'),
		('release/4.5.6', 'be3cb4b2590091dbaf5ce0d861da0f7979c5fc23'),
		('release/7.7.10', '23478a09ad05642d64f44920f3247f7708d4bcae'),
		('release/7.7.11', 'c014e918175915e07eef0150984a24ff8c33e154'),
		('release/7.7.12', 'a7c8068c7c0a45f842400ffb2c4a245ab2f7a388'),
		('release/7.7.13', 'c49c48572f1422e8b222db854d7ecbeba27b8020'),
		('release/7.7.7', 'da67fad037eea0c065a5ea52a894d5522eb9f248'),
		('release/7.7.8', '85003adc6bc0e91391edd97efb7f515cbd6d303e'),
		('release/7.7.9', '73aaf9a4377bab721ba716776e81dc76bec25beb'),
		('v1/foo/bar', '6e3325f7dc37f67b48e719de1ce4a4c9d33d1042'),
		('v1/nix', '64ab804caa81c6a1c302cc03c3671d80f767c2c9'),
		('wut', '558f12143c6d6d60690acb291cf64a863e12c2d0')]

	tags = prepare_collection_list(cloned_base_repo_obj._repo.tags)
	assert tags == [
		('meh', '703de123eaa9832f4ce15baf6b702cd7f72fea5a'),
		('v', '85003adc6bc0e91391edd97efb7f515cbd6d303e'),
		('v1.2.3', '8f3d75c010a30babe3ee8a4176bb16136c5949b3'),
		('v4.5.6', '8a2844f55fc1adde2e346d0999ef3e12aafbe511'),
		('v7.7.10', '23478a09ad05642d64f44920f3247f7708d4bcae'),
		('v7.7.11', 'c014e918175915e07eef0150984a24ff8c33e154'),
		('v7.7.12', 'a7c8068c7c0a45f842400ffb2c4a245ab2f7a388'),
		('v7.7.13', 'c49c48572f1422e8b222db854d7ecbeba27b8020'),
		('v7.7.9', 'dd62ea5560a6944a04c1f8fbdbbb09c01f1b4a21'),
		('wip-001', 'adbb090758d7bc201a35b5a7a470a3c01a3e5a95'),
		('wip-002', 'a05c54d8623558963b8774bb01962fe0683c46bf'),
		('wip-003', 'd3e906333be8a6ff09f96e827199d51694a16dc1'),
		('wip-r/1', '030cc2b97d8ecbf07b7bd361105f07e631393f00')]


def test_fetch_diferent_tags(cloned_base_repo_obj: RepoTool, path_repo_different_tags: Path):
	cloned_base_repo_obj._repo.remotes[0].set_url(path_repo_different_tags.as_uri())
	branches = prepare_collection_list(cloned_base_repo_obj._repo.branches)
	assert branches == base_repo_branches

	tags = prepare_collection_list(cloned_base_repo_obj._repo.tags)
	assert tags == base_repo_tags

	assert cloned_base_repo_obj.fetch()

	branches = prepare_collection_list(cloned_base_repo_obj._repo.branches)
	assert branches == [
		('br0', '1fe320fe278a40bfd6813860d9952b03caf5e2ea'),
		('branch-a', '71ce433cd98f899c723184b9ea1e0bcdf97d5c1a'),
		('branch-d', '5551c9980c9ff2b94865596df35e38527d9fcfbd'), (
			'dependabot/npm_and_yarn/dot-github/actions/test-action/node-fetch-2.6.7',
			'8298c84342db5cb93972426188f06431fc498681'),
		('master', 'e0c7e2a72579e24657c05e875201011d2b48bf94'),
		('orph', '3de11d7640ed17f93ce38d31c9dea0fd65774077'),
		('pr-1', '378d1f9e27dbacfcad0ee93355f0305c6c4432f1'),
		('pr-3', 'e2bf0be1b64f345077fa96671d1c27f0c3476ac4'),
		('release/1.2.3', 'f5eee554aecbee6f2f8e18de089476c3bf957ddc'),
		('release/4.5.6', 'be3cb4b2590091dbaf5ce0d861da0f7979c5fc23'),
		('release/7.7.10', '23478a09ad05642d64f44920f3247f7708d4bcae'),
		('release/7.7.11', 'c014e918175915e07eef0150984a24ff8c33e154'),
		('release/7.7.12', 'a7c8068c7c0a45f842400ffb2c4a245ab2f7a388'),
		('release/7.7.13', 'c49c48572f1422e8b222db854d7ecbeba27b8020'),
		('release/7.7.7', 'da67fad037eea0c065a5ea52a894d5522eb9f248'),
		('release/7.7.8', '85003adc6bc0e91391edd97efb7f515cbd6d303e'),
		('release/7.7.9', '73aaf9a4377bab721ba716776e81dc76bec25beb'),
		('symlink-readme', '8a2844f55fc1adde2e346d0999ef3e12aafbe511'),
		('v1/foo/bar', '6e3325f7dc37f67b48e719de1ce4a4c9d33d1042'),
		('v1/nix', 'c55e6953ff343f9eeb42dc04eb6aa5dfed6f0756')
	]

	tags = prepare_collection_list(cloned_base_repo_obj._repo.tags)
	assert tags == [
		('meh', 'e0c7e2a72579e24657c05e875201011d2b48bf94'),
		('v1.2.3', '8f3d75c010a30babe3ee8a4176bb16136c5949b3'),
		('v4.5.6', '8a2844f55fc1adde2e346d0999ef3e12aafbe511'),
		('v7.7.10', '23478a09ad05642d64f44920f3247f7708d4bcae'),
		('v7.7.11', 'c014e918175915e07eef0150984a24ff8c33e154'),
		('v7.7.12', 'a7c8068c7c0a45f842400ffb2c4a245ab2f7a388'),
		('v7.7.13', 'c49c48572f1422e8b222db854d7ecbeba27b8020'),
		('v7.7.9', 'dd62ea5560a6944a04c1f8fbdbbb09c01f1b4a21'),
		('vv', '85003adc6bc0e91391edd97efb7f515cbd6d303e'),
		('wip-001', 'adbb090758d7bc201a35b5a7a470a3c01a3e5a95'),
		('wip-002', 'a05c54d8623558963b8774bb01962fe0683c46bf'),
		('wip-003', 'd3e906333be8a6ff09f96e827199d51694a16dc1'),
		('wip-r/1', '030cc2b97d8ecbf07b7bd361105f07e631393f00')
	]


def test_fetch_new_commits(cloned_base_repo_obj: RepoTool, path_repo_new_commits: Path):
	cloned_base_repo_obj._repo.remotes[0].set_url(path_repo_new_commits.as_uri())
	branches = prepare_collection_list(cloned_base_repo_obj._repo.branches)
	assert branches == base_repo_branches

	tags = prepare_collection_list(cloned_base_repo_obj._repo.tags)
	assert tags == base_repo_tags

	assert cloned_base_repo_obj.fetch()

	branches = prepare_collection_list(cloned_base_repo_obj._repo.branches)

	assert branches == [
		('br0', '1fe320fe278a40bfd6813860d9952b03caf5e2ea'),
		('branch-a', '442ea9bbc1d5c0fb48325801d3edcb966cc26db2'),
		('branch-d', '201e167782b51e363fe6c1b15f9cec1b28cd1b2b'),
		(
			'dependabot/npm_and_yarn/dot-github/actions/test-action/node-fetch-2.6.7',
			'8298c84342db5cb93972426188f06431fc498681'),
		('master', 'f110a678f5c5a71c8f61bd7d88eeb9c729fd2b38'),
		('orph', '3de11d7640ed17f93ce38d31c9dea0fd65774077'),
		('pr-1', '378d1f9e27dbacfcad0ee93355f0305c6c4432f1'),
		('pr-3', 'e2bf0be1b64f345077fa96671d1c27f0c3476ac4'),
		('release/1.2.3', 'f5eee554aecbee6f2f8e18de089476c3bf957ddc'),
		('release/4.5.6', 'be3cb4b2590091dbaf5ce0d861da0f7979c5fc23'),
		('release/7.7.10', '23478a09ad05642d64f44920f3247f7708d4bcae'),
		('release/7.7.11', 'c014e918175915e07eef0150984a24ff8c33e154'),
		('release/7.7.12', 'a7c8068c7c0a45f842400ffb2c4a245ab2f7a388'),
		('release/7.7.13', 'c49c48572f1422e8b222db854d7ecbeba27b8020'),
		('release/7.7.7', 'da67fad037eea0c065a5ea52a894d5522eb9f248'),
		('release/7.7.8', '85003adc6bc0e91391edd97efb7f515cbd6d303e'),
		('release/7.7.9', '73aaf9a4377bab721ba716776e81dc76bec25beb'),
		('symlink-readme', '8a2844f55fc1adde2e346d0999ef3e12aafbe511'),
		('v1/foo/bar', '6e3325f7dc37f67b48e719de1ce4a4c9d33d1042'),
		('v1/nix', 'c55e6953ff343f9eeb42dc04eb6aa5dfed6f0756')
	]

	tags = prepare_collection_list(cloned_base_repo_obj._repo.tags)
	assert tags == base_repo_tags


def test_fetch_reduced(cloned_base_repo_obj: RepoTool, path_repo_reduced: Path):
	cloned_base_repo_obj._repo.remotes[0].set_url(path_repo_reduced.as_uri())
	branches = prepare_collection_list(cloned_base_repo_obj._repo.branches)
	assert branches == base_repo_branches

	tags = prepare_collection_list(cloned_base_repo_obj._repo.tags)
	assert tags == base_repo_tags

	assert cloned_base_repo_obj.fetch()

	branches = prepare_collection_list(cloned_base_repo_obj._repo.branches)

	assert branches == [
		('master', 'e0c7e2a72579e24657c05e875201011d2b48bf94'),
		('pr-1', '378d1f9e27dbacfcad0ee93355f0305c6c4432f1'),
		('pr-3', 'e2bf0be1b64f345077fa96671d1c27f0c3476ac4'),
		('release/1.2.3', 'f5eee554aecbee6f2f8e18de089476c3bf957ddc'),
		('release/4.5.6', 'be3cb4b2590091dbaf5ce0d861da0f7979c5fc23'),
		('release/7.7.10', '23478a09ad05642d64f44920f3247f7708d4bcae'),
		('release/7.7.11', 'c014e918175915e07eef0150984a24ff8c33e154'),
		('release/7.7.12', 'a7c8068c7c0a45f842400ffb2c4a245ab2f7a388'),
		('release/7.7.13', 'c49c48572f1422e8b222db854d7ecbeba27b8020'),
		('release/7.7.7', 'da67fad037eea0c065a5ea52a894d5522eb9f248'),
		('release/7.7.8', '85003adc6bc0e91391edd97efb7f515cbd6d303e'),
		('release/7.7.9', '73aaf9a4377bab721ba716776e81dc76bec25beb'),
		('symlink-readme', '8a2844f55fc1adde2e346d0999ef3e12aafbe511'),
	]

	tags = prepare_collection_list(cloned_base_repo_obj._repo.tags)
	assert tags == [
		('wip-001', 'adbb090758d7bc201a35b5a7a470a3c01a3e5a95'),
		('wip-002', 'a05c54d8623558963b8774bb01962fe0683c46bf'),
		('wip-003', 'd3e906333be8a6ff09f96e827199d51694a16dc1'),
		('wip-r/1', '030cc2b97d8ecbf07b7bd361105f07e631393f00')
	]