def main(app_class):
def main_func():
parser = concierge.endpoints.cli.create_parser()
parser = app_class.specify_parser(parser)
options = parser.parse_args()
app = app_class(options)
LOG.debug("Options: %s", options)
try:
return app.do()
except KeyboardInterrupt:
pass
except Exception as exc:
LOG.exception("Failed with error %s", exc)
return os.EX_SOFTWARE
return main_func
python类EX_SOFTWARE的实例源码
def test_pdb(mocker):
mocker.patch('ldap2pg.script.dictConfig', autospec=True)
mocker.patch('ldap2pg.script.os.environ', {'DEBUG': '1'})
isatty = mocker.patch('ldap2pg.script.sys.stdout.isatty')
isatty.return_value = True
w = mocker.patch('ldap2pg.script.wrapped_main')
w.side_effect = Exception()
pm = mocker.patch('ldap2pg.script.pdb.post_mortem')
from ldap2pg.script import main
with pytest.raises(SystemExit) as ei:
main()
assert pm.called is True
assert os.EX_SOFTWARE == ei.value.code
def catch_errors(func):
"""Decorator which catches all errors and tries to print them."""
@six.wraps(func)
@click.pass_context
def decorator(ctx, *args, **kwargs):
try:
return func(*args, **kwargs)
except exceptions.DecapodAPIError as exc:
utils.format_output_json(ctx, exc.json, True)
except exceptions.DecapodError as exc:
click.echo(six.text_type(exc), err=True)
finally:
ctx.close()
ctx.exit(os.EX_SOFTWARE)
return decorator
def test_task_failed_exit(mocked_plugin, tpool, configure_model, freeze_time):
tsk = create_task()
polled = {"a": False}
def side_effect(*args, **kwargs):
if polled["a"]:
return False
polled["a"] = True
return True
mocked_plugin.alive.side_effect = side_effect
mocked_plugin.returncode = os.EX_SOFTWARE
tpool.submit(tsk)
time.sleep(2)
tsk.refresh()
assert tsk.executor_host == platform.node()
assert tsk.executor_pid == 100
assert tsk.time_failed == int(freeze_time.return_value)
assert not tsk.time_cancelled
assert not tsk.time_completed
assert not tpool.global_stop_event.is_set()
assert tsk._id not in tpool.data
def test_bdb_quit(mocker):
mocker.patch('ldap2pg.script.dictConfig', autospec=True)
w = mocker.patch('ldap2pg.script.wrapped_main')
from ldap2pg.script import main, pdb
w.side_effect = pdb.bdb.BdbQuit()
with pytest.raises(SystemExit) as ei:
main()
assert os.EX_SOFTWARE == ei.value.code
def test_unhandled_error(mocker):
mocker.patch('ldap2pg.script.dictConfig', autospec=True)
w = mocker.patch('ldap2pg.script.wrapped_main')
from ldap2pg.script import main
w.side_effect = Exception()
with pytest.raises(SystemExit) as ei:
main()
assert os.EX_SOFTWARE == ei.value.code
def main():
debug = os.environ.get('DEBUG', '').lower() in ('1', 'y')
verbose = os.environ.get('VERBOSE', '').lower() in ('1', 'y')
config = Configuration()
config['debug'] = debug
config['verbose'] = debug or verbose
config['color'] = sys.stderr.isatty()
dictConfig(config.logging_dict())
logger.debug("Debug mode enabled.")
try:
exit(wrapped_main(config))
except pdb.bdb.BdbQuit:
logger.info("Graceful exit from debugger.")
except UserError as e:
logger.critical("%s", e)
exit(e.exit_code)
except Exception:
logger.exception('Unhandled error:')
if debug and sys.stdout.isatty():
logger.debug("Dropping in debugger.")
pdb.post_mortem(sys.exc_info()[2])
else:
logger.error(
"Please file an issue at "
"https://github.com/dalibo/ldap2pg/issues with full log.",
)
exit(os.EX_SOFTWARE)
def catch_errors(func):
@functools.wraps(func)
def decorator(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as exc:
LOG.error("Version verification has been failed: %s", exc)
return os.EX_SOFTWARE
return os.EX_OK
return decorator
def test_dry_run_commands(self):
unauthorized_ok = [dict(return_codes=[os.EX_OK]),
dict(return_codes=[1, os.EX_SOFTWARE], stderr="UnauthorizedOperation")]
self.call("aegea launch unittest --dry-run --no-verify-ssh-key-pem-file",
shell=True, expect=unauthorized_ok)
self.call("aegea launch unittest --dry-run --spot --no-verify-ssh-key-pem-file",
shell=True, expect=unauthorized_ok)
self.call("aegea launch unittest --dry-run --duration-hours 1 --no-verify-ssh-key-pem-file",
shell=True, expect=unauthorized_ok)
self.call("aegea launch unittest --duration 0.5 --min-mem 6 --cores 2 --dry-run --no-verify --client-token t",
shell=True, expect=unauthorized_ok)
self.call("aegea build_ami i --dry-run --no-verify-ssh-key-pem-file",
shell=True, expect=unauthorized_ok)
def test_secrets(self):
unauthorized_ok = [dict(return_codes=[os.EX_OK]),
dict(return_codes=[1, os.EX_SOFTWARE], stderr="(AccessDenied|NoSuchKey)")]
self.call("test_secret=test aegea secrets put test_secret --iam-role aegea.launch",
shell=True, expect=unauthorized_ok)
self.call("aegea secrets put test_secret --generate-ssh-key --iam-role aegea.launch",
shell=True, expect=unauthorized_ok)
self.call("aegea secrets ls", shell=True, expect=unauthorized_ok)
self.call("aegea secrets ls --json", shell=True, expect=unauthorized_ok)
self.call("aegea secrets get test_secret --iam-role aegea.launch", shell=True, expect=unauthorized_ok)
self.call("aegea secrets delete test_secret --iam-role aegea.launch", shell=True, expect=unauthorized_ok)
def main(args=None):
parsed_args = parser.parse_args(args=args)
logger.setLevel(parsed_args.log_level)
has_attrs = (getattr(parsed_args, "sort_by", None) and
getattr(parsed_args, "columns", None))
if has_attrs and parsed_args.sort_by not in parsed_args.columns:
parsed_args.columns.append(parsed_args.sort_by)
try:
result = parsed_args.entry_point(parsed_args)
except Exception as e:
if isinstance(e, NoRegionError):
msg = "The AWS CLI is not configured."
msg += " Please configure it using instructions at"
msg += " http://docs.aws.amazon.com/cli/latest/userguide/cli-chap-getting-started.html"
exit(msg)
elif logger.level < logging.ERROR:
raise
else:
err_msg = traceback.format_exc()
try:
err_log_filename = os.path.join(config.user_config_dir, "error.log")
with open(err_log_filename, "ab") as fh:
print(datetime.datetime.now().isoformat(), file=fh)
print(err_msg, file=fh)
exit("{}: {}. See {} for error details.".format(e.__class__.__name__, e, err_log_filename))
except Exception:
print(err_msg, file=sys.stderr)
exit(os.EX_SOFTWARE)
if isinstance(result, SystemExit):
raise result
elif result is not None:
if isinstance(result, dict) and "ResponseMetadata" in result:
del result["ResponseMetadata"]
print(json.dumps(result, indent=2, default=lambda x: str(x)))
def playbook_on_stats(self, stats):
if self.enabled:
if len(stats.dark) > 0:
self._display.warning('idempotency test failed: unreachable=%s > 0' % stats.dark)
sys.exit(os.EX_SOFTWARE)
if len(stats.changed) > 0:
self._display.warning('idempotency test failed: changed=%s > 0' % stats.changed)
sys.exit(os.EX_SOFTWARE)
if len(stats.failures) > 0:
self._display.warning('idempotency test failed: failures=%s > 0' % stats.failures)
sys.exit(os.EX_SOFTWARE)
def test_basic_aegea_commands(self):
self.call(["aegea"], expect=[dict(return_codes=[1])])
self.call(["aegea", "--help"])
self.call(["aegea", "--version"])
self.call(["aegea", "pricing"])
self.call(["aegea", "ls", "-w9"])
self.call(["aegea", "ssh", "nonexistent_instance"],
expect=[dict(return_codes=[1, os.EX_SOFTWARE], stderr="AegeaException: Could not resolve")])
instance_id = json.loads(self.call(["aegea", "ls", "--json"]).stdout)[0]["id"]
for subcommand in aegea.parser._actions[-1].choices:
expect = [dict(return_codes=[os.EX_OK]),
dict(return_codes=[1, os.EX_SOFTWARE],
stderr="(UnauthorizedOperation|AccessDenied|DryRunOperation)")]
args = []
if subcommand in ("ssh", "put-alarm", "put_alarm", "batch"):
args += ["--help"]
elif "_" in subcommand:
continue
elif subcommand == "build-docker-image":
args += ["--dry-run", "docker-example"]
elif subcommand == "console":
args += [instance_id]
elif subcommand == "iam":
args += ["users"]
elif subcommand in ("start", "stop", "reboot", "terminate", "rename"):
args += [instance_id, instance_id, "--dry-run"]
elif subcommand == "grep":
args += ["--help"] if USING_PYTHON2 else ["error", "syslog", "--start-time=-2h", "--end-time=-5m"]
expect.append(dict(return_codes=[os.EX_DATAERR]))
elif subcommand in ("launch", "build-ami"):
args += ["--no-verify-ssh-key-pem-file", "--dry-run", "test"]
elif subcommand == "rm":
args += [resolve_ami()]
elif subcommand in ("secrets", "rds", "elb", "flow-logs", "deploy", "zones", "ebs", "buckets", "efs",
"ecr", "lambda"):
args += ["ls"]
elif subcommand == "pricing":
args += ["AmazonEC2", "--json"]
elif subcommand == "billing":
continue # FIXME
args += ["ls", "--min-cost", "0.1"]
if "AWS_BILLING_REPORTS_BUCKET" in os.environ:
args += ["--billing-reports-bucket", os.environ["AWS_BILLING_REPORTS_BUCKET"]]
elif subcommand == "ls":
args += ["--filter", "state=running"]
elif subcommand == "tag":
args += [instance_id, "test=test test2=test"]
elif subcommand == "untag":
args += [instance_id, "test test2"]
self.call(["aegea", subcommand] + args, expect=expect)
