Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_print_red(self):
with captured_output() as (out, err):
print_red(TEST_MESSAGE)
self.assertEqual(out.getvalue().strip(), "\x1b[31m%s\x1b[0m" % TEST_MESSAGE)
def test_print_red(self):
with captured_output() as (out, err):
print_red(TEST_MESSAGE)
self.assertEqual(out.getvalue().strip(), "\x1b[31m%s\x1b[0m" % TEST_MESSAGE)
if result is None:
sys.exit(0)
else:
sys.exit(result)
except HokusaiError as e:
print_red(e.message)
sys.exit(e.return_code)
except SystemExit:
raise
except KeyboardInterrupt:
raise
except (CalledProcessError, Exception) as e:
if get_verbosity() or os.environ.get('DEBUG') is '1':
print_red(traceback.format_exc(e))
else:
print_red("ERROR: %s" % str(e))
sys.exit(1)
return wrapper
@property
def current_tag(self):
images = []
for deployment in self.cache:
containers = deployment['spec']['template']['spec']['containers']
container_names = [container['name'] for container in containers]
container_images = [container['image'] for container in containers]
if not all(x == container_images[0] for x in container_images):
print_red("Deployment's containers do not reference the same image tag")
return None
images.append(containers[0]['image'])
if not all(y == images[0] for y in images):
print_red("Deployments do not reference the same image tag")
return None
return images[0].rsplit(':', 1)[1]
def wrapper(*args, **kwargs):
try:
if config_check:
config.check()
result = func(*args, **kwargs)
if result is None:
sys.exit(0)
else:
sys.exit(result)
except HokusaiError as e:
print_red(e.message)
sys.exit(e.return_code)
except SystemExit:
raise
except KeyboardInterrupt:
raise
except (CalledProcessError, Exception) as e:
if get_verbosity() or os.environ.get('DEBUG') is '1':
print_red(traceback.format_exc(e))
else:
print_red("ERROR: %s" % str(e))
sys.exit(1)
return wrapper
config.check()
result = func(*args, **kwargs)
if result is None:
sys.exit(0)
else:
sys.exit(result)
except HokusaiError as e:
print_red(e.message)
sys.exit(e.return_code)
except SystemExit:
raise
except KeyboardInterrupt:
raise
except (CalledProcessError, Exception) as e:
if get_verbosity() or os.environ.get('DEBUG') is '1':
print_red(traceback.format_exc(e))
else:
print_red("ERROR: %s" % str(e))
sys.exit(1)
return wrapper
@property
def current_tag(self):
images = []
for deployment in self.cache:
containers = deployment['spec']['template']['spec']['containers']
container_names = [container['name'] for container in containers]
container_images = [container['image'] for container in containers]
if not all(x == container_images[0] for x in container_images):
print_red("Deployment's containers do not reference the same image tag")
return None
images.append(containers[0]['image'])
if not all(y == images[0] for y in images):
print_red("Deployments do not reference the same image tag")
return None
return images[0].rsplit(':', 1)[1]
def check_err(check_item):
print_red(u'\u2718 ' + check_item + ' not found')
"containers": deployment_targets
}
},
"progressDeadlineSeconds": timeout
}
}
print_green("Patching deployment %s..." % deployment['metadata']['name'], newline_after=True)
shout(self.kctl.command("patch deployment %s -p '%s'" % (deployment['metadata']['name'], json.dumps(patch))))
# Watch the rollouts in the cache and if any fail, roll back
print_green("Waiting for deployment rollouts to complete...")
rollout_commands = [self.kctl.command("rollout status deployment/%s" % deployment['metadata']['name']) for deployment in self.cache]
return_codes = shout_concurrent(rollout_commands, print_output=True)
if any(return_codes):
print_red("One or more deployment rollouts failed! Rolling back...", newline_before=True, newline_after=True)
rollback_commands = [self.kctl.command("rollout undo deployment/%s" % deployment['metadata']['name']) for deployment in self.cache]
shout_concurrent(rollback_commands, print_output=True)
raise HokusaiError("Deployment failed!")
post_deploy_success = True
# Run the post-deploy hook for the canonical app or a review app
if config.post_deploy and (filename is None or (filename and self.namespace)):
print_green("Running post-deploy hook '%s'..." % config.post_deploy, newline_after=True)
return_code = CommandRunner(self.context, namespace=self.namespace).run(tag, config.post_deploy, constraint=constraint, tty=False)
if return_code:
print_yellow("WARNING: Running the post-deploy hook failed with return code %s" % return_code, newline_before=True, newline_after=True)
print_yellow("The image digest %s has been rolled out. However, you should run the post-deploy hook '%s' manually, or re-run this deployment." % (digest, config.post_deploy), newline_after=True)
post_deploy_success = False
# For the canonical app, create tags