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

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())
# @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: payload["response"] = exc.response.content raise argparse.ArgumentTypeError("\n" + yaml.safe_dump( payload, default_flow_style=False, width=float("inf"))) if render: self.render()
def add_parser(cls, subparsers): 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, which_cmd=cls.name, parse_unknown=cls.parse_unknown)
print(json.dumps(self._render_dict(), indent=2, separators=(',', ': ')))
raise NotImplementedError
raise NotImplementedError
print(yaml.safe_dump(self._render_dict(), default_flow_style=False))
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", nargs='?', 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")
if default is None: default = cls.default_media_type 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]')
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)") |