-
Notifications
You must be signed in to change notification settings - Fork 8
/
katanacore.py
140 lines (101 loc) · 3.91 KB
/
katanacore.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
from os import listdir, path
from os.path import isdir, join, dirname, realpath, abspath
import yaml
import katanaerrors
import re
module_dict = {}
locked_modules = []
lock_file_read = False
def load_module_info(path):
with open(path, 'r') as stream:
module_info = yaml.load(stream, Loader=yaml.SafeLoader)
module_info['path'] = dirname(path)
if re.fullmatch('[a-zA-Z][a-zA-Z0-9\-_]+', module_info['name']):
provisioner_class = module_info.get("class", "provisioners.DefaultProvisioner")
if "." in provisioner_class:
class_name = provisioner_class[provisioner_class.rindex(".") + 1:]
else:
class_name = provisioner_class
mod = __import__(provisioner_class, fromlist=[class_name])
klass = getattr(mod, class_name)
provisioner = klass(module_info)
module_dict[module_info.get('name').lower()] = provisioner
return provisioner
else:
print(f"ERROR: Module name is invalid. It must be a valid css id: {module_info['name']}")
def list_modules(path=None, module_list=None):
if module_list is None:
module_list = []
if path is None:
my_path = abspath(dirname(__file__))
path = realpath(join(my_path, "modules"))
if len(module_list) == 0:
module_dict.clear()
file_list = listdir(path)
for f in file_list:
file_path = join(path, f)
if isdir(file_path):
list_modules(file_path, module_list)
elif f.endswith(".yml"):
module_info = load_module_info(file_path)
if module_info is not None:
module_list.append(module_info)
return module_list
def get_module_info(name):
return module_dict.get(name)
def _run_function(module_name, function_name, step=False):
if len(module_dict) == 0:
list_modules()
provisioner = module_dict.get(module_name.lower())
if provisioner is None:
raise katanaerrors.ModuleNotFound(module_name)
if hasattr(provisioner, function_name) and callable(getattr(provisioner, function_name)):
function_to_call = getattr(provisioner, function_name)
return function_to_call(step)
else:
raise katanaerrors.NotImplemented(function_name, type(provisioner).__name__)
def install_module(name, step=False):
_run_function(name, "install", step)
def remove_module(name, step=False):
_run_function(name, "remove", step)
def start_module(name):
_run_function(name, "start")
def stop_module(name):
_run_function(name, "stop")
def status_module(name):
return _run_function(name, "status")
def get_available_actions(module_name):
provisioner = module_dict.get(module_name.lower())
if provisioner is None:
raise katanaerrors.ModuleNotFound(module_name)
return provisioner.has_actions(module_name in load_locked_modules())
def lock_modules():
global lock_file_read
locked_modules.clear()
if len(module_dict) == 0:
list_modules()
for module_name in module_dict.keys():
status = status_module(module_name)
if status in ['running', 'installed', 'stopped']:
locked_modules.append(module_name)
my_path = abspath(dirname(__file__))
lock_file = join(my_path, "katana.lock")
with open(lock_file, 'w') as lf:
lf.write("\n".join(locked_modules))
lock_file_read = False
def load_locked_modules():
global lock_file_read
if lock_file_read:
return locked_modules
else:
my_path = abspath(dirname(__file__))
lock_file = join(my_path, "katana.lock")
if path.exists(lock_file):
locked_modules.clear()
with open(lock_file, 'r') as lf:
for module in lf.readlines():
locked_modules.append(module.strip())
else:
locked_modules.clear()
lock_file_read = True
return locked_modules