/
cli_repository.py
144 lines (113 loc) · 4.42 KB
/
cli_repository.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
# -*- coding: utf-8 -*-
#
# Copyright 2021 Swiss Data Science Center (SDSC)
# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Renku CLI fixtures for repository management."""
import contextlib
import os
import secrets
import shutil
import pytest
from tests.utils import format_result_exception
@contextlib.contextmanager
def _isolated_filesystem(tmpdir, name=None, delete=True):
"""Click CliRunner ``isolated_filesystem`` but xdist compatible."""
from renku.core.utils.contexts import chdir
if not name:
name = secrets.token_hex(8)
t = tmpdir.mkdir(name)
with chdir(t):
try:
yield t
finally:
if delete:
try:
shutil.rmtree(t)
except OSError: # noqa: B014
pass
@pytest.fixture
def renku_path(tmpdir):
"""Temporary instance path."""
path = str(tmpdir.mkdir("renku"))
yield path
shutil.rmtree(path)
@pytest.fixture
def instance_path(renku_path, monkeypatch):
"""Temporary instance path."""
with monkeypatch.context() as m:
m.chdir(renku_path)
yield renku_path
@pytest.fixture
def repository(tmpdir):
"""Yield a Renku repository."""
from click.testing import CliRunner
from git.config import GitConfigParser, get_config_path
from renku.cli import cli
runner = CliRunner()
with _isolated_filesystem(tmpdir, delete=True) as project_path:
home = tmpdir.mkdir("user_home")
old_home = os.environ.get("HOME", "")
old_xdg_home = os.environ.get("XDG_CONFIG_HOME", "")
try:
# NOTE: fake user home directory
os.environ["HOME"] = str(home)
os.environ["XDG_CONFIG_HOME"] = str(home)
with GitConfigParser(get_config_path("global"), read_only=False) as global_config:
global_config.set_value("user", "name", "Renku @ SDSC")
global_config.set_value("user", "email", "renku@datascience.ch")
global_config.set_value("pull", "rebase", False)
result = runner.invoke(cli, ["init", ".", "--template-id", "python-minimal"], "\n", catch_exceptions=False)
assert 0 == result.exit_code, format_result_exception(result)
yield os.path.realpath(project_path)
finally:
os.environ["HOME"] = old_home
os.environ["XDG_CONFIG_HOME"] = old_xdg_home
try:
shutil.rmtree(home)
except OSError: # noqa: B014
pass
@pytest.fixture
def project(repository):
"""Create a test project."""
from click.testing import CliRunner
from git import Repo
from renku.cli import cli
from renku.core.utils.contexts import chdir
runner = CliRunner()
repo = Repo(repository, search_parent_directories=True)
commit = repo.head.commit
with chdir(repository):
yield repository
os.chdir(repository)
repo.head.reset(commit, index=True, working_tree=True)
# INFO: remove any extra non-tracked files (.pyc, etc)
repo.git.clean("-xdff")
result = runner.invoke(cli, ["githooks", "install", "--force"])
assert 0 == result.exit_code
@pytest.fixture
def client(project, global_config_dir):
"""Return a Renku repository."""
from renku.core.management.client import LocalClient
from renku.core.models.enums import ConfigFilter
original_get_value = LocalClient.get_value
def mocked_get_value(self, section, key, config_filter=ConfigFilter.ALL):
"""We don't want lfs warnings in tests."""
if key == "show_lfs_message":
return "False"
return original_get_value(self, section, key, config_filter=config_filter)
LocalClient.get_value = mocked_get_value
yield LocalClient(path=project)
LocalClient.get_value = original_get_value