Python os 模块，EX_SOFTWARE 实例源码

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

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)