repo-cloner/tests/lib/test_cloner.py

import git
import pytest
from repo_cloner.lib.checksum import gen_repo_hashed_name
import repo_cloner.lib.dir_not_found_error
from cloner_test_fixtures import *
from repo_cloner.lib.cloner import Cloner
from pathlib import Path
import logging


def mock_time() -> float:
	return 1658702099.4258687


class MockConfig:
	def __init__(self):
		self.cloner_repo_url = ""
		self.cloner_project_name = "Mocked Project"
		self.cloner_interval = 0


class MockDirStruct:
	raise_cache_exists = False

	def __init__(self, tmp: Path):
		self.config = MockConfig()
		self.cache_dir = tmp.joinpath("cache")
		self.config_dir = tmp.joinpath("config")
		self.repos_dir = tmp.joinpath("repos")
		self.raise_cache_exists = False

	@property
	def cache_dir_exists(self):
		if self.raise_cache_exists:
			raise repo_cloner.lib.dir_not_found_error.DirNotFoundError("mock_dir")
		return True


class MockRepoTool:
	path = None
	initialized = False

	def __init__(self, path: str):
		self.path = path
		self.initialized = False


def test_init_invalid_config(tmp_path, caplog):
	caplog.set_level(logging.INFO)
	mock = MockDirStruct(tmp_path)
	with pytest.raises(KeyError) as e:
		Cloner(mock)
	assert "KeyError: 'cloner_repo_url not defined in config!'" == e.exconly()
	assert "CRITICAL root:cloner.py:" in caplog.text
	assert "Undefined repo cloner URL in config!\n" in caplog.text

	# valid config
	caplog.clear()
	# set mocks
	mock.config.cloner_repo_url = "git@mocked:/path"
	cache_dir = tmp_path.joinpath("cache")
	# validate dir does not exist
	assert not cache_dir.exists()
	# set mock to raise error
	mock.raise_cache_exists = True
	# create new cloner class
	Cloner(mock)
	# dir should exist now
	assert cache_dir.exists()
	assert "Cache dir for project Mocked Project not found -> creating" in caplog.text


def test_check_interval(cloner_dir_struct: Path, monkeypatch):
	mock = MockDirStruct(cloner_dir_struct)
	mock.config.cloner_repo_url = "git@mocked:/path"
	timestamp_file = cloner_dir_struct.joinpath("cache", "last-check-time")

	with monkeypatch.context() as m:
		m.setattr("repo_cloner.lib.cloner.time", mock_time)
		cloner = Cloner(mock)
		# no timestamp file exists -> timer has to run
		assert cloner.check_interval()

		# timestamp file in future -> no run
		timestamp_file.touch()
		timestamp_file.write_text(str(int(mock_time() + 1)))
		assert not cloner.check_interval()

		# timestamp in history, but run everytime is set >> RUN
		timestamp_file.write_text(str(int(mock_time() - 1)))
		assert cloner.check_interval()

		# second ago, but with 5 minute interval
		mock.config.cloner_interval = 5
		assert not cloner.check_interval()

		# 6 minutes ago, with 5 min interval -> run!
		timestamp_file.write_text(str(int(mock_time() - 360)))
		assert cloner.check_interval()

		# corrupted file
		timestamp_file.write_text("1234WTF")
		assert cloner.check_interval()


def test__opened(cloner_dir_with_config, monkeypatch):
	ds = MockDirStruct(cloner_dir_with_config)
	ds.config.cloner_repo_url = cloner_dir_with_config.as_uri()
	c = Cloner(ds)
	# no repo defined
	assert not c._Cloner__opened
	# with mocked repo
	monkeypatch.setattr(c, "_repo", MockRepoTool(cloner_dir_with_config.as_posix()))
	c._repo.initialized = True
	assert c._Cloner__opened
	c._repo.initialized = False
	assert not c._Cloner__opened


def test_repo_path_by_url(cloner_dir_struct: Path, path_repo_base: Path):
	ds = MockDirStruct(cloner_dir_struct)
	ds.config.cloner_repo_url = path_repo_base.as_uri()
	c = Cloner(ds)
	x = c._repo_path_by_url("git@server:namespace/repo.git")
	assert x == ds.repos_dir.joinpath("namespace_repo_3375634822.git").as_posix()


def test__main_repo_path(cloner_dir_struct: Path, path_repo_base: Path):
	ds = MockDirStruct(cloner_dir_struct)
	ds.config.cloner_repo_url = path_repo_base.as_uri()
	hashed = gen_repo_hashed_name(path_repo_base.as_uri())
	c = Cloner(ds)
	x = c._Cloner__main_repo_path
	assert x == ds.repos_dir.joinpath(hashed).as_posix()


def test_sync(cloner_dir_struct, tmp_path, monkeypatch):
	called: bool = False

	def fetch():
		nonlocal called
		called = True
		return True

	class PatchCloner(Cloner):
		open = False

		@property
		def __opened(self) -> bool:
			return self.open

	ds = MockDirStruct(cloner_dir_struct)
	ds.config.cloner_repo_url = "file://wut!"
	repo = tmp_path.joinpath("repo.git")
	c = PatchCloner(ds)
	# invalid repo, closed
	c.open = False
	assert not c.sync()
	assert not called
	# mocked valid repo, opened
	c._repo = MockRepoTool(repo.as_posix())
	c._repo.initialized = True
	c._repo.fetch = fetch
	assert c.sync()
	assert called

	called = False
	# mocked repo, invalid, closed
	c._repo.initialized = False
	assert not called
	assert not c.sync()


def test_perform_check(cloner_dir_with_config, monkeypatch, caplog):
	call_counter: int = 0

	def ret_true():
		return True

	def ret_false():
		return False

	def call_counter_inc():
		nonlocal call_counter
		call_counter += 1

	# set logger
	caplog.set_level(logging.INFO)

	mock = MockDirStruct(cloner_dir_with_config)
	mock.config.cloner_repo_url = "git://repo"
	cloner = Cloner(mock)
	# timer did not pass
	monkeypatch.setattr(cloner, "check_interval", ret_false)
	monkeypatch.setattr(cloner, "sync", call_counter_inc)
	assert call_counter == 0
	cloner.perform_check()
	assert call_counter == 0
	# timer passed
	monkeypatch.setattr(cloner, "check_interval", ret_true)
	cloner.perform_check()
	assert call_counter == 1
	# right logs
	assert 2 == sum(
		r.message == "Started check for Mocked Project, url: git://repo"
		and r.levelname == "INFO"
		for r in caplog.records
	)
	assert 2 == sum(
		r.message == "Check finished"
		and r.levelname == "INFO"
		for r in caplog.records
	)


def test_open_uninitialized(cloner_dir_with_config, path_repo_base):
	url = path_repo_base.as_uri()
	mock = MockDirStruct(cloner_dir_with_config)
	mock.config.cloner_repo_url = path_repo_base.as_uri()
	c = Cloner(mock)
	assert not c.open(mock.config.cloner_repo_url)
	assert c._repo_url == path_repo_base.as_uri()


def test_open_initialized(cloner_dir_with_config, path_repo_base, caplog):
	mock = MockDirStruct(cloner_dir_with_config)
	hashed = gen_repo_hashed_name(path_repo_base.as_uri())
	path = cloner_dir_with_config.joinpath("repos", hashed).as_posix()
	mock.config.cloner_repo_url = path_repo_base.as_uri()
	r = git.Repo().clone_from(path_repo_base.as_uri(), path, bare = True)
	commit = r.head.commit.hexsha
	assert r.bare
	c = Cloner(mock)
	assert c.open(path_repo_base.as_uri())
	assert c._repo._repo.head.commit.hexsha == commit


def test_clone_from_url(tmp_path, path_repo_base):
	mock = MockDirStruct(tmp_path)
	mock.config.cloner_repo_url = "invalid"
	c = Cloner(mock)
	assert c.clone_from_url(path_repo_base.as_uri())
	assert "e0c7e2a72579e24657c05e875201011d2b48bf94" == c._repo._repo.head.commit.hexsha


def test_clone_from_url_initialized(tmp_path, path_repo_base, caplog):
	mock = MockDirStruct(tmp_path)
	hashed = gen_repo_hashed_name(path_repo_base.as_uri())
	path = tmp_path.joinpath("repos", hashed).as_posix()
	mock.config.cloner_repo_url = path_repo_base.as_uri()
	git.Repo().init(path)
	c = Cloner(mock)
	assert not c.clone_from_url(path_repo_base.as_uri())
	assert caplog.records[0].levelname == "CRITICAL"
	assert caplog.records[0].message == f"Repo path {path} is initialized... Refusing clone!"