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!"