Coverage for appr/commands/command_base.py : 32%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
parsed_version = parse_version(package_parts['version']) setattr(namespace, "registry_host", package_parts['host']) setattr(namespace, 'version', parsed_version['value']) setattr(namespace, 'version_parts', parsed_version) package = "%s/%s" % (package_parts['namespace'], package_parts['package']) setattr(namespace, dest, package) setattr(namespace, "package_parts", package_parts)
try: name = value[0] package_parts = parse_package_name(name) _set_package(parser, namespace, self.dest, package_parts) except ValueError as exc: raise parser.error(str(exc))
setattr(namespace, self.dest, value[0])
name = value package_parts = split_package_name(name) _set_package(parser, namespace, self.dest, package_parts)
r = {} try: return json.loads(var) except: for v in var.split(","): sp = re.match("(.+?)=(.+)", v) if sp is None: raise ValueError("Malformed variable: %s" % v) key, value = sp.group(1), sp.group(2) r[key] = value return r
with open(filename, 'r') as f: if ext in ['.yml', '.yaml']: return yaml.load(f.read()) elif ext == '.json': return json.loads(f.read()) elif ext in [".jsonnet", "libjsonnet"]: r = RenderJsonnet() return r.render_jsonnet(f.read()) else: raise ValueError("File extension is not in [yaml, json, jsonnet]: %s" % filename)
_, ext = os.path.splitext(var) if ext not in ['.yaml', '.yml', '.json', '.jsonnet']: return self._parse_cmd(var) else: return self._load_from_file(var, ext)
items = copy.copy(argparse._ensure_value(namespace, self.dest, {})) try: items.update(self.load_variables(values)) except ValueError as exc: raise parser.error(option_string + ": " + str(exc)) setattr(namespace, self.dest, items)
self.unknown = unknown self.args_options = args_options self.output = args_options.output
if self.output == 'none': return elif self.output == 'json': self._render_json() elif self.output == 'yaml': self._render_yaml() else: print(self._render_console())
if self.output == 'json': self._render_json(payload) elif self.output == 'yaml': self._render_yaml(payload) else: raise argparse.ArgumentTypeError("\n" + yaml.safe_dump( payload, default_flow_style=False, width=float("inf")))
# @TODO(ant31): all methods should have the 'unknown' parameter if cls.parse_unknown: obj = cls(options, unknown) else: obj = cls(options) obj.exec_cmd(render=render)
try: self._call() except requests.exceptions.RequestException as exc: payload = {"message": str(exc)} if exc.response is not None: content = None try: content = exc.response.json() except ValueError: content = exc.response.content payload["response"] = content self.render_error(payload) exit(2) except subprocess.CalledProcessError as exc: payload = {"message": str(exc.output)} self.render_error(payload) exit(exc.returncode)
if render: self.render()
parser = subparsers.add_parser(cls.name, help=cls.help_message) cls._add_output_option(parser) cls._add_arguments(parser) parser.set_defaults(func=cls.call, env=env, which_cmd=cls.name, parse_unknown=cls.parse_unknown)
if not value: value = self._render_dict() print(json.dumps(value, indent=2, separators=(',', ': ')))
raise NotImplementedError
raise NotImplementedError
if not value: value = self._render_dict() print(yaml.safe_dump(value, default_flow_style=False, width=float("inf")))
raise NotImplementedError
def _add_arguments(cls, parser): raise NotImplementedError
def _add_registryhost_option(cls, parser): parser.add_argument("-H", "--registry-host", default=None, help=argparse.SUPPRESS) parser.add_argument("-k", "--insecure", action="store_true", default=False, help="turn off verification of the https certificate") parser.add_argument("--cacert", default=None, help="CA certificate to verify peer against (SSL)")
def _add_output_option(cls, parser): parser.add_argument("--output", default=cls.output_default, choices=[ 'text', 'none', 'json', 'yaml'], help="output format")
default = os.getenv("APPR_DEFAULT_MEDIA_TYPE", default)
if default is not None: required = False
parser.add_argument( "-t", "--media-type", default=default, required=required, help='package format: [kpm, kpm-compose, helm, docker-compose, kubernetes, appr]')
def _add_packagename_option(cls, parser): parser.add_argument('package', nargs=1, default=None, action=PackageName, help="package-name")
def _add_packagesplit_option(cls, parser): parser.add_argument('package', nargs="?", default=None, action=PackageSplit, help="registry-host.com/namespace/name")
def _add_packageversion_option(cls, parser): parser.add_argument("-v", "--version", help="package VERSION", default='default')
def _add_registryhost_arg(cls, parser): parser.add_argument("registry_host", nargs=1, action=RegistryHost, help='registry API url') parser.add_argument("-k", "--insecure", action="store_true", default=False, help="turn off verification of the https certificate") parser.add_argument("--cacert", nargs='?', default=None, help="CA certificate to verify peer against (SSL)") |