import os import random from pathlib import Path import git import numpy as np import pytest from cadetrdm import initialize_repo, ProjectRepo, clone from cadetrdm.initialize_repo import init_lfs from cadetrdm.io_utils import delete_path from cadetrdm.repositories import OutputRepo, BaseRepo @pytest.fixture(scope="module") def path_to_repo(): # a "fixture" serves up shared, ready variables to test functions that should use the fixture as a kwarg return Path("test_repo") def modify_code(path_to_repo): # Add changes to the project code random_number = random.randint(0, 265) filepath = path_to_repo / f"print_random_number.py" with open(filepath, "w") as file: file.write(f"print({random_number})\n") def count_commit_number(repo): commit_log = repo._git.log("--oneline").split("\n") current_commit_number = len(commit_log) return current_commit_number def example_generate_results_array(path_to_repo, output_folder): results_array = np.random.random((500, 3)) np.savetxt(path_to_repo / output_folder / "result.csv", results_array, delimiter=",") return results_array def try_init_gitpython_repo(repo_path): Path(repo_path).exists() git.Repo(repo_path) return True def try_initialize_git_repo(path_to_repo): if path_to_repo.exists(): delete_path(path_to_repo) initialize_repo(path_to_repo, "results") assert try_init_gitpython_repo(path_to_repo) assert try_init_gitpython_repo(path_to_repo / "results") def try_commit_code(path_to_repo): repo = ProjectRepo(path_to_repo) current_commit_number = count_commit_number(repo) modify_code(path_to_repo) repo.commit("add code to print random number", add_all=True) updated_commit_number = count_commit_number(repo) assert current_commit_number + 1 == updated_commit_number def try_commit_code_without_code_changes(path_to_repo): repo = ProjectRepo(path_to_repo) current_commit_number = count_commit_number(repo) repo.commit("This commit will not be made", add_all=True) updated_commit_number = count_commit_number(repo) assert current_commit_number == updated_commit_number def try_commit_results_data(path_to_repo): repo = ProjectRepo(path_to_repo) current_commit_number = count_commit_number(repo.output_repo) with repo.track_results(results_commit_message="Add array"): example_generate_results_array(path_to_repo, output_folder=repo.output_path) updated_commit_number = count_commit_number(repo.output_repo) assert current_commit_number <= updated_commit_number return str(repo.output_repo.active_branch) def try_output_function(path_to_repo): repo = ProjectRepo(path_to_repo) filepath = repo.output_path / "foo" / "bar" assert isinstance(filepath, Path) filepath = repo.output_data("foo/bar") assert isinstance(filepath, Path) def try_print_log(path_to_repo): repo = ProjectRepo(path_to_repo) repo.print_output_log() def try_commit_results_with_uncommitted_code_changes(path_to_repo): repo = ProjectRepo(path_to_repo) modify_code(path_to_repo) with pytest.raises(Exception): with repo.track_results(results_commit_message="Add array"): example_generate_results_array(path_to_repo, output_folder=repo.output_path) repo.commit("add code to print random number", add_all=True) def try_load_previous_output(path_to_repo, branch_name): repo = ProjectRepo(path_to_repo) with repo.track_results(results_commit_message="Load array and extend"): cached_array_path = repo.input_data(branch_name=branch_name, file_path="result.csv") previous_array = np.loadtxt(cached_array_path, delimiter=",") extended_array = np.concatenate([previous_array, previous_array]) extended_array_file_path = path_to_repo / repo.output_path / "extended_result.csv" np.savetxt(extended_array_file_path, extended_array, delimiter=",") assert cached_array_path.exists() assert extended_array_file_path.exists() def try_add_remote(path_to_repo): repo = ProjectRepo(path_to_repo) repo.add_remote("git@jugit.fz-juelich.de:IBG-1/ModSim/cadet/CADET-RDM.git") assert "origin" in repo._git_repo.remotes def try_initialize_from_remote(): if Path("test_repo_from_remote").exists(): delete_path("test_repo_from_remote") clone("https://jugit.fz-juelich.de/IBG-1/ModSim/cadet/rdm-examples-fraunhofer-ime-aachen", "test_repo_from_remote") assert try_init_gitpython_repo("test_repo_from_remote") def test_init_over_existing_repo(monkeypatch): path_to_repo = Path("test_repo_2") if path_to_repo.exists(): delete_path(path_to_repo) os.makedirs(path_to_repo) os.chdir(path_to_repo) os.system(f"git init") with open("README.md", "w") as handle: handle.write("Readme-line 1\n") with open(".gitignore", "w") as handle: handle.write("foo.bar.*") repo = git.Repo(".") repo.git.add(".") repo.git.commit("-m", "Initial commit") os.chdir("..") # using monkeypath to simulate user input monkeypatch.setattr('builtins.input', lambda x: "Y") initialize_repo(path_to_repo) delete_path(path_to_repo) def test_cache_with_non_rdm_repo(monkeypatch): path_to_repo = Path("non_rdm_repo") if path_to_repo.exists(): delete_path(path_to_repo) os.makedirs(path_to_repo) os.chdir(path_to_repo) os.system(f"git init") with open("README.md", "w") as handle: handle.write("Readme-line 1\n") with open(".gitignore", "w") as handle: handle.write("foo.bar.*") repo = git.Repo(".") repo.git.add(".") repo.git.commit("-m", "Initial commit") os.chdir("..") if Path("test_repo_non_rdm_imports").exists(): delete_path("test_repo_non_rdm_imports") initialize_repo("test_repo_non_rdm_imports") os.chdir("test_repo_non_rdm_imports") repo = BaseRepo(".") if Path("external_cache/non_rdm_repo").exists(): delete_path("external_cache/non_rdm_repo") if Path("foo/bar/non_rdm_repo").exists(): delete_path("foo/bar/non_rdm_repo") # import two repos and confirm verify works. repo.import_remote_repo(source_repo_location=".." / path_to_repo, source_repo_branch="master") repo.import_remote_repo(source_repo_location=".." / path_to_repo, source_repo_branch="master", target_repo_location="foo/bar/non_rdm_repo") repo.verify_unchanged_cache() def test_add_lfs_filetype(): path_to_repo = Path("test_repo_3") if path_to_repo.exists(): delete_path(path_to_repo) os.makedirs(path_to_repo) initialize_repo(path_to_repo) file_type = "*.bak" init_lfs(lfs_filetypes=[file_type], path=path_to_repo) repo = ProjectRepo(path_to_repo) repo.add_all_files() repo.commit(f"Add {file_type} to lfs") delete_path(path_to_repo) def test_cadet_rdm(path_to_repo): # because these depend on one-another and there is no native support afaik for sequential tests # these tests are called sequentially here as try_ functions. try_initialize_git_repo(path_to_repo) # try_initialize_from_remote() try_add_remote(path_to_repo) # try_add_submodule(path_to_repo) try_commit_code(path_to_repo) try_commit_code_without_code_changes(path_to_repo) try_commit_results_with_uncommitted_code_changes(path_to_repo) try_output_function(path_to_repo) results_branch_name = try_commit_results_data(path_to_repo) try_print_log(path_to_repo) try_commit_code(path_to_repo) try_load_previous_output(path_to_repo, results_branch_name) def test_with_external_repos(): path_to_repo = Path("test_repo_external_data") if path_to_repo.exists(): delete_path(path_to_repo) os.makedirs(path_to_repo) initialize_repo(path_to_repo) os.chdir(path_to_repo) # to be able to hand over a valid branch, I first need to determine that branch imported_repo = OutputRepo("../test_repo/results") branch_name = imported_repo.active_branch.name repo = ProjectRepo(".") # import two repos and confirm verify works. repo.import_remote_repo(source_repo_location="../test_repo/results", source_repo_branch=branch_name) repo.import_remote_repo(source_repo_location="../test_repo/results", source_repo_branch=branch_name, target_repo_location="foo/bar/repo") # delete folder and reload delete_path("foo/bar/repo") with pytest.raises(Exception): repo.verify_unchanged_cache() repo.fill_data_from_cadet_rdm_json() repo.verify_unchanged_cache() # Test if re_load correctly re_loads by modifying and then reloading with open("external_cache/results/README.md", "w") as file_handle: file_handle.writelines(["a", "b", "c"]) repo.fill_data_from_cadet_rdm_json(re_load=True) repo.verify_unchanged_cache() # modify file and confirm error raised with open("external_cache/results/README.md", "w") as file_handle: file_handle.writelines(["a", "b", "c"]) with pytest.raises(Exception): repo.verify_unchanged_cache() os.chdir("..")