almalinux
/
lorax
14
0
Fork 0
Code Issues Pull Requests Releases Activity

Add composer-cli and tests

This commit is contained in:
Brian C. Lane 2018-05-04 11:35:31 -07:00
parent b67c821706
commit fb58277028
15 changed files with 1818 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -29,7 +29,7 @@ test:
@echo "*** Running tests ***"
PYTHONPATH=$(PYTHONPATH):./src/ $(PYTHON) -m nose -v --with-coverage --cover-erase --cover-branches \
--cover-package=pylorax --cover-inclusive \
./tests/pylorax/
./tests/pylorax/ ./tests/composer/
coverage3 report -m
[ -f "/usr/bin/coveralls" ] && [ -n "$(COVERALLS_REPO_TOKEN)" ] && coveralls || echo

137
src/bin/composer-cli Executable file
View File

@ -0,0 +1,137 @@
#!/usr/bin/python3
#
# composer-cli
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
log = logging.getLogger("composer-cli")
import os
import sys
import argparse
from composer import vernum
from composer.cli import main
VERSION = "{0}-{1}".format(os.path.basename(sys.argv[0]), vernum)
# Documentation for the commands
epilog = """
compose start <blueprint> <type> Start a compose using the selected blueprint and output type.
types List the supported output types.
status List the status of all running and finished composes.
log <uuid> [<size>kB] Show the last 1kB of the compose log.
cancel <uuid> Cancel a running compose and delete any intermediate results.
delete <uuid,...> Delete the listed compose results.
info <uuid> Show detailed information on the compose.
metadata <uuid> Download the metadata use to create the compose to <uuid>-metadata.tar
logs <uuid> Download the compose logs to <uuid>-logs.tar
results <uuid> Download all of the compose results; metadata, logs, and image to <uuid>.tar
image <uuid> Download the output image from the compose. Filename depends on the type.
blueprints list List the names of the available blueprints.
show <blueprint,...> Display the blueprint in TOML format.
changes <blueprint,...> Display the changes for each blueprint.
diff <blueprint-name> Display the differences between 2 versions of a blueprint.
<from-commit> Commit hash or NEWEST
<to-commit> Commit hash, NEWEST, or WORKSPACE
save <blueprint,...> Save the blueprint to a file, <blueprint-name>.toml
delete <blueprint> Delete a blueprint from the server
depsolve <blueprint,...> Display the packages needed to install the blueprint.
push <blueprint> Push a blueprint TOML file to the server.
freeze <blueprint,...> Display the frozen blueprint's modules and packages.
freeze show <blueprint,...> Display the frozen blueprint in TOML format.
freeze save <blueprint,...> Save the frozen blueprint to a file, <blueprint-name>.frozen.toml.
tag <blueprint> Tag the most recent blueprint commit as a release.
undo <blueprint> <commit> Undo changes to a blueprint by reverting to the selected commit.
workspace <blueprint> Push the blueprint TOML to the temporary workspace storage.
modules list List the available modules.
projects list List the available projects.
projects info <project,...> Show details about the listed projects.
"""
def get_parser():
""" Return the ArgumentParser for composer-cli"""
parser = argparse.ArgumentParser(description="Lorax Composer commandline tool",
epilog=epilog,
formatter_class=argparse.RawDescriptionHelpFormatter,
fromfile_prefix_chars="@")
parser.add_argument("-j", "--json", action="store_true", default=False,
help="Output the raw JSON response instead of the normal output.")
parser.add_argument("-s", "--socket", default="/run/weldr/api.socket", metavar="SOCKET",
help="Path to the socket file to listen on")
parser.add_argument("--log", dest="logfile", default="./composer-cli.log", metavar="LOG",
help="Path to logfile (./composer-cli.log)")
parser.add_argument("-a", "--api", dest="api_version", default="0", metavar="APIVER",
help="API Version to use")
parser.add_argument("--test", dest="testmode", default=0, type=int, metavar="TESTMODE",
help="Pass test mode to compose. 1=Mock compose with fail. 2=Mock compose with finished.")
parser.add_argument("-V", action="store_true", dest="showver",
help="show program's version number and exit")
# Commands are implemented by parsing the remaining arguments outside of argparse
parser.add_argument('args', nargs=argparse.REMAINDER)
return parser
def setup_logging(logfile):
# Setup logging to console and to logfile
log.setLevel(logging.DEBUG)
sh = logging.StreamHandler()
sh.setLevel(logging.INFO)
fmt = logging.Formatter("%(asctime)s: %(message)s")
sh.setFormatter(fmt)
log.addHandler(sh)
fh = logging.FileHandler(filename=logfile)
fh.setLevel(logging.DEBUG)
fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
fh.setFormatter(fmt)
log.addHandler(fh)
if __name__ == '__main__':
# parse the arguments
opts = get_parser().parse_args()
if opts.showver:
print(VERSION)
sys.exit(0)
logpath = os.path.abspath(os.path.dirname(opts.logfile))
if not os.path.isdir(logpath):
os.makedirs(logpath)
setup_logging(opts.logfile)
log.debug("opts=%s", opts)
errors = []
# Check to see if the socket exists and can be accessed
if not os.path.exists(opts.socket):
errors.append("%s does not exist" % opts.socket)
elif not os.access(opts.socket, os.R_OK|os.W_OK):
errors.append("This user cannot access %s" % opts.socket)
# No point in continuing if there are errors
if errors:
for e in errors:
log.error(e)
sys.exit(1)
sys.exit(main(opts))

27
src/composer/__init__.py Normal file
View File

@ -0,0 +1,27 @@
#!/usr/bin/python
#
# composer-cli
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# get composer version
try:
import composer.version
except ImportError:
vernum = "devel"
else:
vernum = composer.version.num

56
src/composer/cli/__init__.py Normal file
View File

@ -0,0 +1,56 @@
#!/usr/bin/python
#
# composer-cli
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
log = logging.getLogger("composer-cli")
from composer.cli.blueprints import blueprints_cmd
from composer.cli.modules import modules_cmd
from composer.cli.projects import projects_cmd
from composer.cli.compose import compose_cmd
command_map = {
"blueprints": blueprints_cmd,
"modules": modules_cmd,
"projects": projects_cmd,
"compose": compose_cmd
}
def main(opts):
""" Main program execution
:param opts: Cmdline arguments
:type opts: argparse.Namespace
"""
if len(opts.args) == 0:
log.error("Missing command")
return 1
elif opts.args[0] not in command_map:
log.error("Unknown command %s", opts.args[0])
return 1
if len(opts.args) == 1:
log.error("Missing %s sub-command", opts.args[0])
return 1
else:
try:
return command_map[opts.args[0]](opts)
except Exception as e:
log.error(str(e))
return 1

520
src/composer/cli/blueprints.py Normal file
View File

@ -0,0 +1,520 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
log = logging.getLogger("composer-cli")
import os
import json
from composer import http_client as client
from composer.cli.utilities import argify, frozen_toml_filename, toml_filename, handle_api_result
from composer.cli.utilities import packageNEVRA
def blueprints_cmd(opts):
"""Process blueprints commands
:param opts: Cmdline arguments
:type opts: argparse.Namespace
:returns: Value to return from sys.exit()
:rtype: int
This dispatches the blueprints commands to a function
"""
cmd_map = {
"list": blueprints_list,
"show": blueprints_show,
"changes": blueprints_changes,
"diff": blueprints_diff,
"save": blueprints_save,
"delete": blueprints_delete,
"depsolve": blueprints_depsolve,
"push": blueprints_push,
"freeze": blueprints_freeze,
"tag": blueprints_tag,
"undo": blueprints_undo,
"workspace": blueprints_workspace
}
if opts.args[1] not in cmd_map:
log.error("Unknown blueprints command: %s", opts.args[1])
return 1
return cmd_map[opts.args[1]](opts.socket, opts.api_version, opts.args[2:], opts.json)
def blueprints_list(socket_path, api_version, args, show_json=False):
"""Output the list of available blueprints
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints list
"""
api_route = client.api_url(api_version, "/blueprints/list")
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
print("blueprints: " + ", ".join([r for r in result["blueprints"]]))
return 0
def blueprints_show(socket_path, api_version, args, show_json=False):
"""Show the blueprints, in TOML format
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints show <blueprint,...> Display the blueprint in TOML format.
Multiple blueprints will be separated by \n\n
"""
for blueprint in argify(args):
api_route = client.api_url(api_version, "/blueprints/info/%s?format=toml" % blueprint)
print(client.get_url_raw(socket_path, api_route) + "\n\n")
return 0
def blueprints_changes(socket_path, api_version, args, show_json=False):
"""Display the changes for each of the blueprints
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints changes <blueprint,...> Display the changes for each blueprint.
"""
api_route = client.api_url(api_version, "/blueprints/changes/%s" % (",".join(argify(args))))
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
for blueprint in result["blueprints"]:
print(blueprint["name"])
for change in blueprint["changes"]:
prettyCommitDetails(change)
return 0
def prettyCommitDetails(change, indent=4):
"""Print the blueprint's change in a nice way
:param change: The individual blueprint change dict
:type change: dict
:param indent: Number of spaces to indent
:type indent: int
"""
def revision():
if change["revision"]:
return " revision %d" % change["revision"]
else:
return ""
print(" " * indent + change["timestamp"] + " " + change["commit"] + revision())
print(" " * indent + change["message"] + "\n")
def blueprints_diff(socket_path, api_version, args, show_json=False):
"""Display the differences between 2 versions of a blueprint
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints diff <blueprint-name> Display the differences between 2 versions of a blueprint.
<from-commit> Commit hash or NEWEST
<to-commit> Commit hash, NEWEST, or WORKSPACE
"""
if len(args) == 0:
log.error("blueprints diff is missing the blueprint name, from commit, and to commit")
return 1
elif len(args) == 1:
log.error("blueprints diff is missing the from commit, and the to commit")
return 1
elif len(args) == 2:
log.error("blueprints diff is missing the to commit")
return 1
api_route = client.api_url(api_version, "/blueprints/diff/%s/%s/%s" % (args[0], args[1], args[2]))
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
for err in result.get("errors", []):
log.error(err)
if result.get("errors", False):
return 1
for diff in result["diff"]:
print(prettyDiffEntry(diff))
return 0
def prettyDiffEntry(diff):
"""Generate nice diff entry string.
:param diff: Difference entry dict
:type diff: dict
:returns: Nice string
"""
def change(diff):
if diff["old"] and diff["new"]:
return "Changed"
elif diff["new"] and not diff["old"]:
return "Added"
elif diff["old"] and not diff["new"]:
return "Removed"
else:
return "Unknown"
def name(diff):
if diff["old"]:
return list(diff["old"].keys())[0]
elif diff["new"]:
return list(diff["new"].keys())[0]
else:
return "Unknown"
def details(diff):
if change(diff) == "Changed":
if name(diff) == "Description":
return '"%s" -> "%s"' % (diff["old"][name(diff)], diff["new"][name(diff)])
elif name(diff) == "Version":
return "%s -> %s" % (diff["old"][name(diff)], diff["new"][name(diff)])
elif name(diff) in ["Module", "Package"]:
return "%s %s -> %s" % (diff["old"][name(diff)]["name"], diff["old"][name(diff)]["version"],
diff["new"][name(diff)]["version"])
else:
return "Unknown"
elif change(diff) == "Added":
if name(diff) in ["Module", "Package"]:
return "%s %s" % (diff["new"][name(diff)]["name"], diff["new"][name(diff)]["version"])
else:
return " ".join([diff["new"][k] for k in diff["new"]])
elif change(diff) == "Removed":
if name(diff) in ["Module", "Package"]:
return "%s %s" % (diff["old"][name(diff)]["name"], diff["old"][name(diff)]["version"])
else:
return " ".join([diff["old"][k] for k in diff["old"]])
return change(diff) + " " + name(diff) + " " + details(diff)
def blueprints_save(socket_path, api_version, args, show_json=False):
"""Save the blueprint to a TOML file
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints save <blueprint,...> Save the blueprint to a file, <blueprint-name>.toml
"""
for blueprint in argify(args):
api_route = client.api_url(api_version, "/blueprints/info/%s?format=toml" % blueprint)
blueprint_toml = client.get_url_raw(socket_path, api_route)
open(toml_filename(blueprint), "w").write(blueprint_toml)
return 0
def blueprints_delete(socket_path, api_version, args, show_json=False):
"""Delete a blueprint from the server
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
delete <blueprint> Delete a blueprint from the server
"""
api_route = client.api_url(api_version, "/blueprints/delete/%s" % args[0])
result = client.delete_url_json(socket_path, api_route)
return handle_api_result(result, show_json)
def blueprints_depsolve(socket_path, api_version, args, show_json=False):
"""Display the packages needed to install the blueprint
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints depsolve <blueprint,...> Display the packages needed to install the blueprint.
"""
api_route = client.api_url(api_version, "/blueprints/depsolve/%s" % (",".join(argify(args))))
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
for blueprint in result["blueprints"]:
if blueprint["blueprint"].get("version", ""):
print("blueprint: %s v%s" % (blueprint["blueprint"]["name"], blueprint["blueprint"]["version"]))
else:
print("blueprint: %s" % (blueprint["blueprint"]["name"]))
for dep in blueprint["dependencies"]:
print(" " + packageNEVRA(dep))
return 0
def blueprints_push(socket_path, api_version, args, show_json=False):
"""Push a blueprint TOML file to the server, updating the blueprint
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
push <blueprint> Push a blueprint TOML file to the server.
"""
api_route = client.api_url(api_version, "/blueprints/new")
rval = 0
for blueprint in argify(args):
if not os.path.exists(blueprint):
log.error("Missing blueprint file: %s", blueprint)
continue
blueprint_toml = open(blueprint, "r").read()
result = client.post_url_toml(socket_path, api_route, blueprint_toml)
if handle_api_result(result, show_json):
rval = 1
return rval
def blueprints_freeze(socket_path, api_version, args, show_json=False):
"""Handle the blueprints freeze commands
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints freeze <blueprint,...> Display the frozen blueprint's modules and packages.
blueprints freeze show <blueprint,...> Display the frozen blueprint in TOML format.
blueprints freeze save <blueprint,...> Save the frozen blueprint to a file, <blueprint-name>.frozen.toml.
"""
if args[0] == "show":
return blueprints_freeze_show(socket_path, api_version, args[1:], show_json)
elif args[0] == "save":
return blueprints_freeze_save(socket_path, api_version, args[1:], show_json)
if len(args) == 0:
log.error("freeze is missing the blueprint name")
return 1
api_route = client.api_url(api_version, "/blueprints/freeze/%s" % (",".join(argify(args))))
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
else:
for entry in result["blueprints"]:
blueprint = entry["blueprint"]
if blueprint.get("version", ""):
print("blueprint: %s v%s" % (blueprint["name"], blueprint["version"]))
else:
print("blueprint: %s" % (blueprint["name"]))
for m in blueprint["modules"]:
print(" %s-%s" % (m["name"], m["version"]))
for p in blueprint["packages"]:
print(" %s-%s" % (p["name"], p["version"]))
# Print any errors
for err in result.get("errors", []):
log.error(err)
# Return a 1 if there are any errors
if result.get("errors", []):
return 1
else:
return 0
def blueprints_freeze_show(socket_path, api_version, args, show_json=False):
"""Show the frozen blueprint in TOML format
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints freeze show <blueprint,...> Display the frozen blueprint in TOML format.
"""
if len(args) == 0:
log.error("freeze show is missing the blueprint name")
return 1
for blueprint in argify(args):
api_route = client.api_url(api_version, "/blueprints/freeze/%s?format=toml" % blueprint)
print(client.get_url_raw(socket_path, api_route))
return 0
def blueprints_freeze_save(socket_path, api_version, args, show_json=False):
"""Save the frozen blueprint to a TOML file
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints freeze save <blueprint,...> Save the frozen blueprint to a file, <blueprint-name>.frozen.toml.
"""
if len(args) == 0:
log.error("freeze save is missing the blueprint name")
return 1
for blueprint in argify(args):
api_route = client.api_url(api_version, "/blueprints/freeze/%s?format=toml" % blueprint)
blueprint_toml = client.get_url_raw(socket_path, api_route)
open(frozen_toml_filename(blueprint), "w").write(blueprint_toml)
return 0
def blueprints_tag(socket_path, api_version, args, show_json=False):
"""Tag the most recent blueprint commit as a release
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints tag <blueprint> Tag the most recent blueprint commit as a release.
"""
api_route = client.api_url(api_version, "/blueprints/tag/%s" % args[0])
result = client.post_url(socket_path, api_route, "")
return handle_api_result(result, show_json)
def blueprints_undo(socket_path, api_version, args, show_json=False):
"""Undo changes to a blueprint
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints undo <blueprint> <commit> Undo changes to a blueprint by reverting to the selected commit.
"""
if len(args) == 0:
log.error("undo is missing the blueprint name and commit hash")
return 1
elif len(args) == 1:
log.error("undo is missing commit hash")
return 1
api_route = client.api_url(api_version, "/blueprints/undo/%s/%s" % (args[0], args[1]))
result = client.post_url(socket_path, api_route, "")
return handle_api_result(result, show_json)
def blueprints_workspace(socket_path, api_version, args, show_json=False):
"""Push the blueprint TOML to the temporary workspace storage
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
blueprints workspace <blueprint> Push the blueprint TOML to the temporary workspace storage.
"""
api_route = client.api_url(api_version, "/blueprints/workspace")
rval = 0
for blueprint in argify(args):
if not os.path.exists(blueprint):
log.error("Missing blueprint file: %s", blueprint)
continue
blueprint_toml = open(blueprint, "r").read()
result = client.post_url_toml(socket_path, api_route, blueprint_toml)
if show_json:
print(json.dumps(result, indent=4))
for err in result.get("errors", []):
log.error(err)
# Any errors results in returning a 1, but we continue with the rest first
if not result.get("status", False):
rval = 1
return rval

457
src/composer/cli/compose.py Normal file
View File

@ -0,0 +1,457 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
log = logging.getLogger("composer-cli")
import sys
import json
from composer import http_client as client
from composer.cli.utilities import argify, handle_api_result, packageNEVRA
def compose_cmd(opts):
"""Process compose commands
:param opts: Cmdline arguments
:type opts: argparse.Namespace
:returns: Value to return from sys.exit()
:rtype: int
This dispatches the compose commands to a function
"""
cmd_map = {
"status": compose_status,
"types": compose_types,
"start": compose_start,
"log": compose_log,
"cancel": compose_cancel,
"delete": compose_delete,
"info": compose_info,
"metadata": compose_metadata,
"results": compose_results,
"logs": compose_logs,
"image": compose_image,
}
if opts.args[1] not in cmd_map:
log.error("Unknown compose command: %s", opts.args[1])
return 1
return cmd_map[opts.args[1]](opts.socket, opts.api_version, opts.args[2:], opts.json, opts.testmode)
def compose_status(socket_path, api_version, args, show_json=False, testmode=0):
"""Return the status of all known composes
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
This doesn't map directly to an API command, it combines the results from queue, finished,
and failed so raw JSON output is not available.
"""
def get_status(compose):
return {"id": compose["id"],
"blueprint": compose["blueprint"],
"version": compose["version"],
"compose_type": compose["compose_type"],
"image_size": compose["image_size"],
"status": compose["queue_status"]}
# Sort the status in a specific order
def sort_status(a):
order = ["RUNNING", "WAITING", "FINISHED", "FAILED"]
return (order.index(a["status"]), a["blueprint"], a["version"], a["compose_type"])
status = []
# Get the composes currently in the queue
api_route = client.api_url(api_version, "/compose/queue")
result = client.get_url_json(socket_path, api_route)
status.extend(list(map(get_status, result["run"] + result["new"])))
# Get the list of finished composes
api_route = client.api_url(api_version, "/compose/finished")
result = client.get_url_json(socket_path, api_route)
status.extend(list(map(get_status, result["finished"])))
# Get the list of failed composes
api_route = client.api_url(api_version, "/compose/failed")
result = client.get_url_json(socket_path, api_route)
status.extend(list(map(get_status, result["failed"])))
# Sort them by status (running, waiting, finished, failed) and then by name and version.
status.sort(key=sort_status)
if show_json:
print(json.dumps(status, indent=4))
return 0
# Print them as UUID blueprint STATUS
for c in status:
if c["image_size"] > 0:
image_size = str(c["image_size"])
else:
image_size = ""
print("%s %-8s %-15s %s %-16s %s" % (c["id"], c["status"], c["blueprint"], c["version"], c["compose_type"],
image_size))
def compose_types(socket_path, api_version, args, show_json=False, testmode=0):
"""Return information about the supported compose types
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
Add additional details to types that are known to composer-cli. Raw JSON output does not
include this extra information.
"""
api_route = client.api_url(api_version, "/compose/types")
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
print("Compose Types: " + ", ".join([t["name"] for t in result["types"]]))
def compose_start(socket_path, api_version, args, show_json=False, testmode=0):
"""Start a new compose using the selected blueprint and type
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: Set to 1 to simulate a failed compose, set to 2 to simulate a finished one.
:type testmode: int
compose start <blueprint-name> <compose-type>
"""
if len(args) == 0:
log.error("start is missing the blueprint name and output type")
return 1
if len(args) == 1:
log.error("start is missing the output type")
return 1
config = {
"blueprint_name": args[0],
"compose_type": args[1],
"branch": "master"
}
if testmode:
test_url = "?test=%d" % testmode
else:
test_url = ""
api_route = client.api_url(api_version, "/compose" + test_url)
result = client.post_url_json(socket_path, api_route, json.dumps(config))
if show_json:
print(json.dumps(result, indent=4))
return 0
for err in result.get("errors", []):
log.error(err)
if result["status"] == False or result.get("errors", False):
return 1
print("Compose %s added to the queue" % result["build_id"])
return 0
def compose_log(socket_path, api_version, args, show_json=False, testmode=0):
"""Show the last part of the compose log
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
compose log <uuid> [<size>kB]
This will display the last 1kB of the compose's log file. Can be used to follow progress
during the build.
"""
if len(args) == 0:
log.error("log is missing the compose build id")
return 1
if len(args) == 2:
try:
log_size = int(args[1])
except ValueError:
log.error("Log size must be an integer.")
return 1
else:
log_size = 1024
api_route = client.api_url(api_version, "/compose/log/%s?size=%d" % (args[0], log_size))
try:
result = client.get_url_raw(socket_path, api_route)
except RuntimeError as e:
print(str(e))
return 1
print(result)
return 0
def compose_cancel(socket_path, api_version, args, show_json=False, testmode=0):
"""Cancel a running compose
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
compose cancel <uuid>
This will cancel a running compose. It does nothing if the compose has finished.
"""
if len(args) == 0:
log.error("cancel is missing the compose build id")
return 1
api_route = client.api_url(api_version, "/compose/cancel/%s" % args[0])
result = client.delete_url_json(socket_path, api_route)
return handle_api_result(result, show_json)
def compose_delete(socket_path, api_version, args, show_json=False, testmode=0):
"""Delete a finished compose's results
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
compose delete <uuid,...>
Delete the listed compose results. It will only delete results for composes that have finished
or failed, not a running compose.
"""
if len(args) == 0:
log.error("delete is missing the compose build id")
return 1
api_route = client.api_url(api_version, "/compose/delete/%s" % (",".join(argify(args))))
result = client.delete_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
# Print any errors
for err in result.get("errors", []):
log.error(err)
if result.get("errors", []):
return 1
else:
return 0
def compose_info(socket_path, api_version, args, show_json=False, testmode=0):
"""Return detailed information about the compose
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
compose info <uuid>
This returns information about the compose, including the blueprint and the dependencies.
"""
if len(args) == 0:
log.error("info is missing the compose build id")
return 1
api_route = client.api_url(api_version, "/compose/info/%s" % args[0])
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
for err in result.get("errors", []):
log.error(err)
if result.get("errors", []):
return 1
if result["image_size"] > 0:
image_size = str(result["image_size"])
else:
image_size = ""
print("%s %-8s %-15s %s %-16s %s" % (result["id"],
result["queue_status"],
result["blueprint"]["name"],
result["blueprint"]["version"],
result["compose_type"],
image_size))
print("Packages:")
for p in result["blueprint"]["packages"]:
print(" %s-%s" % (p["name"], p["version"]))
print("Modules:")
for m in result["blueprint"]["modules"]:
print(" %s-%s" % (m["name"], m["version"]))
print("Dependencies:")
for d in result["deps"]["packages"]:
print(" " + packageNEVRA(d))
def compose_metadata(socket_path, api_version, args, show_json=False, testmode=0):
"""Download a tar file of the compose's metadata
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
compose metadata <uuid>
Saves the metadata as uuid-metadata.tar
"""
if len(args) == 0:
log.error("metadata is missing the compose build id")
return 1
api_route = client.api_url(api_version, "/compose/metadata/%s" % args[0])
return client.download_file(socket_path, api_route)
def compose_results(socket_path, api_version, args, show_json=False, testmode=0):
"""Download a tar file of the compose's results
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
compose results <uuid>
The results includes the metadata, output image, and logs.
It is saved as uuid.tar
"""
if len(args) == 0:
log.error("results is missing the compose build id")
return 1
api_route = client.api_url(api_version, "/compose/results/%s" % args[0])
return client.download_file(socket_path, api_route, sys.stdout.isatty())
def compose_logs(socket_path, api_version, args, show_json=False, testmode=0):
"""Download a tar of the compose's logs
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
compose logs <uuid>
Saves the logs as uuid-logs.tar
"""
if len(args) == 0:
log.error("logs is missing the compose build id")
return 1
api_route = client.api_url(api_version, "/compose/logs/%s" % args[0])
return client.download_file(socket_path, api_route, sys.stdout.isatty())
def compose_image(socket_path, api_version, args, show_json=False, testmode=0):
"""Download the compose's output image
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
:param testmode: unused in this function
:type testmode: int
compose image <uuid>
This downloads only the result image, saving it as the image name, which depends on the type
of compose that was selected.
"""
if len(args) == 0:
log.error("logs is missing the compose build id")
return 1
api_route = client.api_url(api_version, "/compose/image/%s" % args[0])
return client.download_file(socket_path, api_route, sys.stdout.isatty())

44
src/composer/cli/modules.py Normal file
View File

@ -0,0 +1,44 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
log = logging.getLogger("composer-cli")
import json
from composer import http_client as client
def modules_cmd(opts):
"""Process modules commands
:param opts: Cmdline arguments
:type opts: argparse.Namespace
:returns: Value to return from sys.exit()
:rtype: int
"""
if opts.args[1] != "list":
log.error("Unknown modules command: %s", opts.args[1])
return 1
api_route = client.api_url(opts.api_version, "/modules/list")
result = client.get_url_json(opts.socket, api_route)
if opts.json:
print(json.dumps(result, indent=4))
return 0
print("Modules:\n" + "\n".join([" "+r["name"] for r in result["modules"]]))
return 0

106
src/composer/cli/projects.py Normal file
View File

@ -0,0 +1,106 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
log = logging.getLogger("composer-cli")
import json
import textwrap
from composer import http_client as client
def projects_cmd(opts):
"""Process projects commands
:param opts: Cmdline arguments
:type opts: argparse.Namespace
:returns: Value to return from sys.exit()
:rtype: int
"""
cmd_map = {
"list": projects_list,
"info": projects_info,
}
if opts.args[1] not in cmd_map:
log.error("Unknown projects command: %s", opts.args[1])
return 1
return cmd_map[opts.args[1]](opts.socket, opts.api_version, opts.args[2:], opts.json)
def projects_list(socket_path, api_version, args, show_json=False):
"""Output the list of available projects
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
projects list
"""
api_route = client.api_url(api_version, "/projects/list")
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
for proj in result["projects"]:
for k in ["name", "summary", "homepage", "description"]:
print("%s: %s" % (k.title(), textwrap.fill(proj[k], subsequent_indent=" " * (len(k)+2))))
print("\n\n")
return 0
def projects_info(socket_path, api_version, args, show_json=False):
"""Output info on a list of projects
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param api_version: Version of the API to talk to. eg. "0"
:type api_version: str
:param args: List of remaining arguments from the cmdline
:type args: list of str
:param show_json: Set to True to show the JSON output instead of the human readable output
:type show_json: bool
projects info <project,...>
"""
if len(args) == 0:
log.error("projects info is missing the packages")
return 1
api_route = client.api_url(api_version, "/projects/info/%s" % ",".join(args))
result = client.get_url_json(socket_path, api_route)
if show_json:
print(json.dumps(result, indent=4))
return 0
for proj in result["projects"]:
for k in ["name", "summary", "homepage", "description"]:
print("%s: %s" % (k.title(), textwrap.fill(proj[k], subsequent_indent=" " * (len(k)+2))))
print("Builds: ")
for build in proj["builds"]:
print(" %s%s-%s.%s at %s for %s" % ("" if not build["epoch"] else build["epoch"] + ":",
build["source"]["version"],
build["release"],
build["arch"],
build["build_time"],
build["changelog"]))
print("")
return 0

84
src/composer/cli/utilities.py Normal file
View File

@ -0,0 +1,84 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
log = logging.getLogger("composer-cli")
import json
def argify(args):
"""Take a list of human args and return a list with each item
:param args: list of strings with possible commas and spaces
:type args: list of str
:returns: List of all the items
:rtype: list of str
Examples:
["one,two", "three", ",four", ",five,"] returns ["one", "two", "three", "four", "five"]
"""
return [i for i in [arg for entry in args for arg in entry.split(",")] if i]
def toml_filename(blueprint_name):
"""Convert a blueprint name into a filename.toml
:param blueprint_name: The blueprint's name
:type blueprint_name: str
:returns: The blueprint name with ' ' converted to - and .toml appended
:rtype: str
"""
return blueprint_name.replace(" ", "-") + ".toml"
def frozen_toml_filename(blueprint_name):
"""Convert a blueprint name into a filename.toml
:param blueprint_name: The blueprint's name
:type blueprint_name: str
:returns: The blueprint name with ' ' converted to - and .toml appended
:rtype: str
"""
return blueprint_name.replace(" ", "-") + ".frozen.toml"
def handle_api_result(result, show_json=False):
"""Log any errors, return the correct value
:param result: JSON result from the http query
:type result: dict
"""
if show_json:
print(json.dumps(result, indent=4))
for err in result.get("errors", []):
log.error(err)
if result["status"] == True:
return 0
else:
return 1
def packageNEVRA(pkg):
"""Return the package info as a NEVRA
:param pkg: The package details
:type pkg: dict
:returns: name-[epoch:]version-release-arch
:rtype: str
"""
if pkg["epoch"]:
return "%s-%s:%s-%s.%s" % (pkg["name"], pkg["epoch"], pkg["version"], pkg["release"], pkg["arch"])
else:
return "%s-%s-%s.%s" % (pkg["name"], pkg["version"], pkg["release"], pkg["arch"])

201
src/composer/http_client.py Normal file
View File

@ -0,0 +1,201 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
log = logging.getLogger("composer-cli")
import os
import sys
import json
from composer.unix_socket import UnixHTTPConnectionPool
def api_url(api_version, url):
"""Return the versioned path to the API route
:param api_version: The version of the API to talk to. eg. "0"
:type api_version: str
:param url: The API route to talk to
:type url: str
:returns: The full url to use for the route and API version
:rtype: str
"""
return os.path.normpath("/api/v%s/%s" % (api_version, url))
def get_url_raw(socket_path, url):
"""Return the raw results of a GET request
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param url: URL to request
:type url: str
:returns: The raw response from the server
:rtype: str
"""
http = UnixHTTPConnectionPool(socket_path)
r = http.request("GET", url)
if r.status == 400:
err = json.loads(r.data.decode("utf-8"))
if "status" in err and err["status"] == False:
raise RuntimeError(", ".join(err["errors"]))
return r.data.decode('utf-8')
def get_url_json(socket_path, url):
"""Return the JSON results of a GET request
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param url: URL to request
:type url: str
:returns: The json response from the server
:rtype: dict
"""
http = UnixHTTPConnectionPool(socket_path)
r = http.request("GET", url)
return json.loads(r.data.decode('utf-8'))
def delete_url_json(socket_path, url):
"""Send a DELETE request to the url and return JSON response
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param url: URL to send DELETE to
:type url: str
:returns: The json response from the server
:rtype: dict
"""
http = UnixHTTPConnectionPool(socket_path)
r = http.request("DELETE", url)
return json.loads(r.data.decode("utf-8"))
def post_url(socket_path, url, body):
"""POST raw data to the URL
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param url: URL to send POST to
:type url: str
:param body: The data for the body of the POST
:type body: str
:returns: The json response from the server
:rtype: dict
"""
http = UnixHTTPConnectionPool(socket_path)
r = http.request("POST", url,
body=body.encode("utf-8"))
return json.loads(r.data.decode("utf-8"))
def post_url_toml(socket_path, url, body):
"""POST a TOML string to the URL
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param url: URL to send POST to
:type url: str
:param body: The data for the body of the POST
:type body: str
:returns: The json response from the server
:rtype: dict
"""
http = UnixHTTPConnectionPool(socket_path)
r = http.request("POST", url,
body=body.encode("utf-8"),
headers={"Content-Type": "text/x-toml"})
return json.loads(r.data.decode("utf-8"))
def post_url_json(socket_path, url, body):
"""POST some JSON data to the URL
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param url: URL to send POST to
:type url: str
:param body: The data for the body of the POST
:type body: str
:returns: The json response from the server
:rtype: dict
"""
http = UnixHTTPConnectionPool(socket_path)
r = http.request("POST", url,
body=body.encode("utf-8"),
headers={"Content-Type": "application/json"})
return json.loads(r.data.decode("utf-8"))
def get_filename(headers):
"""Get the filename from the response header
:param response: The urllib3 response object
:type response: Response
:raises: RuntimeError if it cannot find a filename in the header
:returns: Filename from content-disposition header
:rtype: str
"""
log.debug("Headers = %s", headers)
if "content-disposition" not in headers:
raise RuntimeError("No Content-Disposition header; cannot get filename")
try:
k, _, v = headers["content-disposition"].split(";")[1].strip().partition("=")
if k != "filename":
raise RuntimeError("No filename= found in content-disposition header")
except RuntimeError:
raise
except Exception as e:
raise RuntimeError("Error parsing filename from content-disposition header: %s" % str(e))
return os.path.basename(v)
def download_file(socket_path, url, progress=True):
"""Download a file, saving it to the CWD with the included filename
:param socket_path: Path to the Unix socket to use for API communication
:type socket_path: str
:param url: URL to send POST to
:type url: str
"""
http = UnixHTTPConnectionPool(socket_path)
r = http.request("GET", url, preload_content=False)
if r.status == 400:
err = json.loads(r.data.decode("utf-8"))
if not err["status"]:
raise RuntimeError(", ".join(err["errors"]))
filename = get_filename(r.headers)
if os.path.exists(filename):
msg = "%s exists, skipping download" % filename
log.error(msg)
raise RuntimeError(msg)
with open(filename, "wb") as f:
while True:
data = r.read(10 * 1024**2)
if not data:
break
f.write(data)
if progress:
data_written = f.tell()
if data_written > 5 * 1024**2:
sys.stdout.write("%s: %0.2f MB \r" % (filename, data_written / 1024**2))
else:
sys.stdout.write("%s: %0.2f kB\r" % (filename, data_written / 1024))
sys.stdout.flush()
print("")
r.release_conn()
return 0

61
src/composer/unix_socket.py Normal file
View File

@ -0,0 +1,61 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import http.client
import socket
import urllib3
# These 2 classes were adapted and simplified for use with just urllib3.
# Originally from https://github.com/msabramo/requests-unixsocket/blob/master/requests_unixsocket/adapters.py
# The following was adapted from some code from docker-py
# https://github.com/docker/docker-py/blob/master/docker/transport/unixconn.py
class UnixHTTPConnection(http.client.HTTPConnection, object):
def __init__(self, socket_path, timeout=60):
"""Create an HTTP connection to a unix domain socket
:param socket_path: The path to the Unix domain socket
:param timeout: Number of seconds to timeout the connection
"""
super(UnixHTTPConnection, self).__init__('localhost', timeout=timeout)
self.socket_path = socket_path
self.sock = None
def __del__(self): # base class does not have d'tor
if self.sock:
self.sock.close()
def connect(self):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(self.timeout)
sock.connect(self.socket_path)
self.sock = sock
class UnixHTTPConnectionPool(urllib3.connectionpool.HTTPConnectionPool):
def __init__(self, socket_path, timeout=60):
"""Create a connection pool using a Unix domain socket
:param socket_path: The path to the Unix domain socket
:param timeout: Number of seconds to timeout the connection
"""
super(UnixHTTPConnectionPool, self).__init__('localhost', timeout=timeout)
self.socket_path = socket_path
def _new_conn(self):
return UnixHTTPConnection(self.socket_path, self.timeout)

40
tests/composer/test_blueprints.py Normal file
View File

@ -0,0 +1,40 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import unittest
from composer.cli.blueprints import prettyDiffEntry
diff_entries = [{'new': {'Description': 'Shiny new description'}, 'old': {'Description': 'Old reliable description'}},
{'new': {'Version': '0.3.1'}, 'old': {'Version': '0.1.1'}},
{'new': {'Module': {'name': 'openssh', 'version': '2.8.1'}}, 'old': None},
{'new': None, 'old': {'Module': {'name': 'bash', 'version': '4.*'}}},
{'new': {'Module': {'name': 'httpd', 'version': '3.8.*'}},
'old': {'Module': {'name': 'httpd', 'version': '3.7.*'}}},
{'new': {'Package': {'name': 'git', 'version': '2.13.*'}}, 'old': None}]
diff_result = [
'Changed Description "Old reliable description" -> "Shiny new description"',
'Changed Version 0.1.1 -> 0.3.1',
'Added Module openssh 2.8.1',
'Removed Module bash 4.*',
'Changed Module httpd 3.7.* -> 3.8.*',
'Added Package git 2.13.*']
class BlueprintsTest(unittest.TestCase):
def test_prettyDiffEntry(self):
"""Return a nice representation of a diff entry"""
self.assertEqual([prettyDiffEntry(entry) for entry in diff_entries], diff_result)

36
tests/composer/test_http_client.py Normal file
View File

@ -0,0 +1,36 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import unittest
from composer.http_client import api_url, get_filename
headers = {'content-disposition': 'attachment; filename=e7b9b9b0-5867-493d-89c3-115cfe9227d7-metadata.tar;',
'access-control-max-age': '21600',
'transfer-encoding': 'chunked',
'date': 'Tue, 13 Mar 2018 17:37:18 GMT',
'access-control-allow-origin': '*',
'access-control-allow-methods': 'HEAD, OPTIONS, GET',
'content-type': 'application/x-tar'}
class HttpClientTest(unittest.TestCase):
def test_api_url(self):
"""Return the API url including the API version"""
self.assertEqual(api_url("0", "/path/to/enlightenment"), "/api/v0/path/to/enlightenment")
def test_get_filename(self):
"""Return the filename from a content-disposition header"""
self.assertEqual(get_filename(headers), "e7b9b9b0-5867-493d-89c3-115cfe9227d7-metadata.tar")

47
tests/composer/test_utilities.py Normal file
View File

@ -0,0 +1,47 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import unittest
from composer.cli.utilities import argify, toml_filename, frozen_toml_filename, packageNEVRA
class CliUtilitiesTest(unittest.TestCase):
def test_argify(self):
"""Convert an optionally comma-separated cmdline into a list of args"""
self.assertEqual(argify(["one,two", "three", ",four", ",five,"]), ["one", "two", "three", "four", "five"])
def test_toml_filename(self):
"""Return the recipe's toml filename"""
self.assertEqual(toml_filename("http server"), "http-server.toml")
def test_frozen_toml_filename(self):
"""Return the recipe's frozen toml filename"""
self.assertEqual(frozen_toml_filename("http server"), "http-server.frozen.toml")
def test_packageNEVRA(self):
"""Return a string with the NVRA or NEVRA"""
epoch_0 = {"arch": "noarch",
"epoch": 0,
"name": "basesystem",
"release": "7.el7",
"version": "10.0"}
epoch_3 = {"arch": "noarch",
"epoch": 3,
"name": "basesystem",
"release": "7.el7",
"version": "10.0"}
self.assertEqual(packageNEVRA(epoch_0), "basesystem-10.0-7.el7.noarch")
self.assertEqual(packageNEVRA(epoch_3), "basesystem-3:10.0-7.el7.noarch")

1
tests/pylint/runpylint.py
View File

@ -14,6 +14,7 @@ class LoraxLintConfig(PocketLintConfig):
FalsePositive(r"Context manager 'lock' doesn't implement __enter__ and __exit__"),
FalsePositive(r"Catching too general exception Exception"),
FalsePositive(r"^E0712.*: Catching an exception which doesn't inherit from (Base|)Exception: GError$"),
FalsePositive(r"Module 'composer' has no 'version' member"),
]
@property