Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUpClass(cls):
kubernetes.ECR = MockECR
kubernetes.HOKUSAI_CONFIG_DIR = os.path.join(CWD, 'test/fixtures/project/hokusai')
cls.kubernetes_yml = os.path.abspath(os.path.join(kubernetes.HOKUSAI_CONFIG_DIR, 'minikube.yml'))
cls.kctl = Kubectl(TEST_KUBE_CONTEXT)
if tag is 'latest' and not ecr.tag_exists(context):
ecr.retag(tag, context)
print_green("Updated tag 'latest' -> %s" % context)
if filename is None:
configmap = ConfigMap(context, namespace=namespace)
for s in environment:
if '=' not in s:
raise HokusaiError("Error: environment variables must be of the form 'KEY=VALUE'")
split = s.split('=', 1)
configmap.update(split[0], split[1])
configmap.create()
print_green("Created configmap %s-environment" % config.project_name)
kctl = Kubectl(context, namespace=namespace)
kubernetes_spec = KubernetesSpec(kubernetes_yml).to_file()
try:
shout(kctl.command("create --save-config -f %s" % kubernetes_spec), print_output=True)
print_green("Created Kubernetes environment %s" % kubernetes_yml)
finally:
os.unlink(kubernetes_spec)
def k8s_status(context, resources, pods, describe, top, namespace=None, filename=None):
if filename is None:
kubernetes_yml = os.path.join(CWD, HOKUSAI_CONFIG_DIR, "%s.yml" % context)
else:
kubernetes_yml = filename
if not os.path.isfile(kubernetes_yml):
raise HokusaiError("Yaml file %s does not exist." % kubernetes_yml)
kctl = Kubectl(context, namespace=namespace)
kubernetes_spec = KubernetesSpec(kubernetes_yml).to_file()
try:
if describe:
kctl_cmd = "describe"
output = ""
else:
kctl_cmd = "get"
output = " -o wide"
if resources:
print_green("Resources", newline_before=True)
print_green("===========")
shout(kctl.command("%s -f %s%s" % (kctl_cmd, kubernetes_spec, output)), print_output=True)
if pods:
print_green("Pods", newline_before=True)
print_green("===========")
shout(kctl.command("%s pods --selector app=%s,layer=application%s" % (kctl_cmd, config.project_name, output)), print_output=True)
def __init__(self, context):
self.context = context
self.kctl = Kubectl(self.context)
self.cache = self.kctl.get_object('deployment', selector="app=%s,layer=application" % config.project_name)
return_code += 1
if os.path.isfile(os.path.join(os.getcwd(), 'hokusai/development.yml')):
check_ok('./hokusai/development.yml')
else:
check_err('./hokusai/development.yml')
return_code += 1
if os.path.isfile(os.path.join(os.getcwd(), 'hokusai/test.yml')):
check_ok('./hokusai/test.yml')
else:
check_err('./hokusai/test.yml')
return_code += 1
for context in ['staging', 'production']:
if context in Kubectl('staging').contexts():
check_ok("kubectl context '%s'" % context)
else:
check_err("kubectl context '%s'" % context)
return_code += 1
if os.path.isfile(os.path.join(os.getcwd(), "hokusai/%s.yml" % context)):
check_ok("./hokusai/%s.yml" % context)
else:
check_err("./hokusai/%s.yml" % context)
return_code += 1
return return_code
def __init__(self, context, deployment_name=None, namespace=None):
self.context = context
self.namespace = namespace
self.kctl = Kubectl(self.context, namespace=namespace)
self.ecr = ECR()
if deployment_name:
self.cache = [self.kctl.get_object("deployment %s" % deployment_name)]
else:
self.cache = self.kctl.get_objects('deployment', selector="app=%s,layer=application" % config.project_name)
def __init__(self, context, namespace='default', name=None):
self.context = context
self.kctl = Kubectl(context, namespace=namespace)
self.name = name or "%s-environment" % config.project_name
self.metadata = {
'name': self.name,
'namespace': namespace
}
if name is None:
self.metadata['labels'] = { 'app': config.project_name }
self.struct = {
'apiVersion': 'v1',
'kind': 'ConfigMap',
'metadata': self.metadata,
'data': {}
}
def __init__(self, context, application_only=True):
self.context = context
self.kctl = Kubectl(self.context)
if application_only:
selector = "app=%s,layer=application" % config.project_name
else:
selector = "app=%s" % config.project_name
self.cache = self.kctl.get_object('service', selector=selector)
def k8s_delete(context, namespace=None, filename=None):
if filename is None:
kubernetes_yml = os.path.join(CWD, HOKUSAI_CONFIG_DIR, "%s.yml" % context)
else:
kubernetes_yml = filename
if not os.path.isfile(kubernetes_yml):
raise HokusaiError("Yaml file %s does not exist." % kubernetes_yml)
if filename is None:
configmap = ConfigMap(context, namespace=namespace)
configmap.destroy()
print_green("Deleted configmap %s-environment" % config.project_name)
kctl = Kubectl(context, namespace=namespace)
kubernetes_spec = KubernetesSpec(kubernetes_yml).to_file()
try:
shout(kctl.command("delete -f %s" % kubernetes_spec), print_output=True)
print_green("Deleted Kubernetes environment %s" % kubernetes_yml)
finally:
os.unlink(kubernetes_spec)
def __init__(self, context):
self.context = context
self.kctl = Kubectl(self.context)
self.cache = self.kctl.get_object('service', selector="app=%s,layer=application" % config.project_name)