-
-
Notifications
You must be signed in to change notification settings - Fork 23
/
client.py
232 lines (187 loc) · 6.96 KB
/
client.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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
__author__ = "Vanessa Sochat"
__copyright__ = "Copyright 2021, Vanessa Sochat"
__license__ = "MPL 2.0"
from shpc.logger import logger
import shpc.main.container as container
import shpc.utils as utils
from .settings import Settings
import os
import re
import shutil
import sys
class Client:
"""
Interact with a local filesystem Singularity HPC registry.
This client has handles to the registry files, lmod (and other) files,
and settings to make it easy to otherwise updating your config or
running commands. This client should be retrieved with:
from shpc.main import get_client
client = get_client()
As this will ensure that the proper database functions are added. The
database will not be accessible otherwise. This is also a really hairy
name to type out and import.
"""
# Setup
def __init__(self, settings_file=None):
# We don't necessarily need a container technology handle
if not hasattr(self, "container"):
self.container = None
# If we don't have default settings, load
if not hasattr(self, "settings"):
self.settings = Settings(settings_file)
def __repr__(self):
return str(self)
def __str__(self):
return "[shpc-client]"
def install(self, name, tag=None, **kwargs):
"""
Install must be implemented by the subclass (e.g., lmod)
"""
raise NotImplementedError
def uninstall(self, name, tag=None):
"""
Uninstall must also implemented by the subclass (e.g., lmod)
"""
raise NotImplementedError
def get(self, module_name):
"""
Get a container path or uri.
"""
raise NotImplementedError
def add(self, sif, module_name):
"""
Add a container directly as a module
"""
raise NotImplementedError
def inspect(self, module_name):
"""
Return complete metadata for the user from a container.
"""
raise NotImplementedError
def add_namespace(self, name):
"""
If a namespace is defined in settings, use it
"""
if self.settings.namespace:
name = "%s/%s" % (self.settings.namespace.strip("/"), name)
return name
def load_registry_config(self, name):
"""
Given an identifier, find the first match in the registry.
"""
for registry, fullpath in self.container.iter_registry():
package_dir = os.path.join(registry, name)
package_file = os.path.join(package_dir, "container.yaml")
if package_file == fullpath:
return container.ContainerConfig(package_file)
logger.exit("%s is not a known recipe in any registry." % name)
def _load_container(self, name, tag=None):
"""
Given a name and an optional tag to default to, load a package
"""
# Split name and tag
if ":" in name:
name, tag = name.split(":", 1)
# If the user provides a tag, set it
config = self.load_registry_config(name)
config.set_tag(tag)
return config
def test(
self,
module_name,
stage=False,
test_exec=False,
skip_module=False,
test_commands=False,
template=None,
):
"""
Test install of a module
"""
def cleanup(tmpdir):
if not stage:
shutil.rmtree(tmpdir)
# Create temporary directory to work in
tmpdir = utils.get_tmpdir(prefix="shpc-test")
if hasattr(self, "_test_setup"):
self._test_setup(tmpdir)
# Derive the registry entry from the module_name
config = self._load_container(module_name)
# Generate a test template
test_file = os.path.join(tmpdir, "test.sh")
# If the module name has a tag, only test it
if ":" in module_name:
module_name = module_name.split(":", 1)[0]
tags = [config.tag.name]
else:
tags = list(config.tags.keys())
# Test all tags (this could be subsetted)
for tag in tags:
image = self.install(module_name, tag)
# Do we want to test loading?
if not skip_module and hasattr(self, "_test"):
result = self._test(module_name, tmpdir, tag, template)
if result != 0:
cleanup(tmpdir)
logger.exit("Test of %s was not successful." % module_name)
# Do we want to test the test commands?
if test_commands and config.test:
utils.write_file(test_file, config.test)
return_code = self.container.test_script(image, test_file)
if return_code != 0:
cleanup(tmpdir)
logger.exit("Test of %s was not successful." % module_name)
# Test the commands
if not test_exec:
continue
for alias in config.get_aliases():
result = self.client.execute(image, alias["command"])
# Cleanup the test install
if stage:
logger.info(tmpdir)
else:
self.uninstall("%s:%s" % (module_name, tag), force=True)
cleanup(tmpdir)
def check(self, module_name):
"""
Given a module name, check if the latest is installed.
If the user provides a top level folder, assume we want to look
at updates for entire tags. If a specific folder is provided with
a container, check the digest.
"""
raise NotImplementedError
def list(self, pattern=None, names_only=False, out=None):
"""
List installed modules.
"""
raise NotImplementedError
def docgen(self, module_name):
"""
Render documentation for a module.
"""
raise NotImplementedError
def show(self, name, names_only=False, out=None, filter_string=None):
"""
Show available packages
"""
if name:
name = self.add_namespace(name)
config = self._load_container(name)
config.dump(out)
else:
out = out or sys.stdout
# List the known registry modules
for registry, fullpath in self.container.iter_registry():
if fullpath.endswith("container.yaml"):
module_name = (
os.path.dirname(fullpath).replace(registry, "").strip(os.sep)
)
# If the user has provided a filter, honor it
if filter_string and not re.search(filter_string, module_name):
continue
if names_only:
out.write("%s\n" % module_name)
else:
config = self._load_container(module_name)
for version in config.tags.keys():
out.write("%s:%s\n" % (module_name, version))