[copr] master: [cli] Restored missing commit, more tests for cli (25e8011)
by vgologuz@fedoraproject.org
Repository : http://git.fedorahosted.org/cgit/copr.git
On branch : master
>---------------------------------------------------------------
commit 25e8011e42832915a1e7d7390da28f7c8f967665
Author: Valentin Gologuzov <vgologuz(a)redhat.com>
Date: Tue Oct 7 17:13:01 2014 +0200
[cli] Restored missing commit, more tests for cli
>---------------------------------------------------------------
cli/.coveragerc | 10 +
cli/copr_cli/main.py | 252 +++++++++++++------------
cli/run_tests.sh | 4 +-
cli/tests/test_cli.py | 504 ++++++++++++++++++++++++++++++++++++++++++++++++-
4 files changed, 639 insertions(+), 131 deletions(-)
diff --git a/cli/.coveragerc b/cli/.coveragerc
new file mode 100644
index 0000000..7aae70d
--- /dev/null
+++ b/cli/.coveragerc
@@ -0,0 +1,10 @@
+[report]
+exclude_lines =
+ pragma: no cover
+ def __repr__
+ if self.debug:
+ if settings.DEBUG
+ raise AssertionError
+ raise NotImplementedError
+ if 0:
+ if __name__ == .__main__.:
diff --git a/cli/copr_cli/main.py b/cli/copr_cli/main.py
index a016161..2c65ecd 100644
--- a/cli/copr_cli/main.py
+++ b/cli/copr_cli/main.py
@@ -6,7 +6,6 @@ __description__ = "CLI tool to run copr"
import argparse
import sys
-import ConfigParser
import datetime
import time
from collections import defaultdict
@@ -28,11 +27,8 @@ no_config_warning = """
class Commands(object):
def __init__(self):
- self.no_config = True
try:
self.client = CoprClient.create_from_file_config()
- self.no_config = False
- #print("CREATED FROM CONFIG")
except (copr_exceptions.CoprNoConfException,
copr_exceptions.CoprConfigException) as e:
print(no_config_warning)
@@ -44,8 +40,9 @@ class Commands(object):
def requires_api_auth(func):
""" Decorator that checks config presence
"""
+
def wrapper(self, args):
- if self.no_config:
+ if self.client.no_config:
print("Error: Operation requires api authentication\n"
"File `~/.config/copr` is missing or incorrect")
sys.exit(6)
@@ -59,8 +56,9 @@ class Commands(object):
def check_username_presence(func):
""" Decorator that checks if username was provided
"""
+
def wrapper(self, args):
- if self.no_config and args.username is None:
+ if self.client.no_config and args.username is None:
print("Error: Operation requires username\n"
"Pass username to command or create `~/.config/copr`")
sys.exit(6)
@@ -68,6 +66,7 @@ class Commands(object):
if args.username is None and self.client.username is None:
print("Error: Operation requires username\n"
"Pass username to command or add it to `~/.config/copr`")
+ sys.exit(6)
return func(self, args)
@@ -75,101 +74,113 @@ class Commands(object):
wrapper.__name__ = func.__name__
return wrapper
- # def _watch_builds(sefl, builds_list):
- # """
- # :param builds_list: list of BuildWrapper
- #
- # """
- # print("Watching build(s): (this may be safely interrupted)")
- # prevstatus = defaultdict(lambda: None)
- # failed_ids = []
- # watched_ids = [bw.build_id for bw in builds_list]
- #
- # try:
- # while True:
- # for build_id in watched_ids:
- # build_details = client.get_build_details(build_id)
- # if build_details.output != "ok":
- # errmsg = " Build {1}: Unable to get build status: {0}".\
- # format(build_details.error, build_id)
- # raise copr_exceptions.CoprRequestException(errmsg)
- #
- # now = datetime.datetime.now()
- # if prevstatus[build_id] != build_details.status:
- # prevstatus[build_id] = build_details.status
- # print(" {0} Build {2}: {1}".format(
- # now.strftime("%H:%M:%S"),
- # build_details.status, build_id))
- #
- # if build_details.status in ["failed"]:
- # failed_ids.append(build_id)
- # if build_details.status in ["succeeded", "skipped",
- # "failed", "canceled"]:
- # watched_ids.remove(build_id)
- # if build_details.status == "unknown":
- # raise copr_exceptions.CoprBuildException(
- # "Unknown status.")
- #
- # if not watched_ids:
- # break
- # time.sleep(3)
- #
- # if failed_ids:
- # raise copr_exceptions.CoprBuildException(
- # "Build(s) {0} failed.".format(
- # ", ".join(str(x) for x in failed_ids)))
- #
- # except KeyboardInterrupt:
- # pass
- #
- #
- # @requires_api_auth
- # def action_build(args):
- # """ Method called when the 'build' action has been selected by the
- # user.
- #
- # :param args: argparse arguments provided by the user
- #
- # """
- # result = client.create_new_build(
- # projectname=args.copr, chroots=args.chroots, pkgs=args.pkgs,
- # memory=args.memory, timeout=args.timeout)
- # if result.output != "ok":
- # print(result.error)
- # return
- # print(result.message)
- # print("Created builds: {0}".format(" ".join(map(str, result.builds_list))))
- #
- # if not args.nowait:
- # _watch_builds(result.builds_list)
- #
- #
- # @requires_api_auth
- # def action_create(args):
- # """ Method called when the 'create' action has been selected by the
- # user.
- #
- # :param args: argparse arguments provided by the user
- #
- # """
- # result = client.create_project(
- # projectname=args.name, description=args.description,
- # instructions=args.instructions, chroots=args.chroots,
- # repos=args.repos, initial_pkgs=args.initial_pkgs)
- # print(result.message)
- #
- #
- # @requires_api_auth
- # def action_delete(args):
- # """ Method called when the 'delete' action has been selected by the
- # user.
- #
- # :param args: argparse arguments provided by the user
- # """
- # result = client.delete_project(projectname=args.copr)
- # print(result.message)
- #
- #
+ def _watch_builds(self, builds_list):
+ """
+ :param builds_list: list of BuildWrapper
+ """
+ print("Watching build(s): {}\n this may be safely interrupted)".format(
+ [b.build_id for b in builds_list]))
+
+ prevstatus = defaultdict(lambda: None)
+ failed_ids = []
+
+ watched = set([bw.build_id for bw in builds_list])
+ done = set()
+
+ try:
+ while watched != done:
+ for build_id in watched:
+ if build_id in done:
+ continue
+
+ build_details = self.client.get_build_details(build_id)
+
+ if build_details.output != "ok":
+ errmsg = " Build {1}: Unable to get build status: {0}". \
+ format(build_details.error, build_id)
+ raise copr_exceptions.CoprRequestException(errmsg)
+
+ now = datetime.datetime.now()
+ if prevstatus[build_id] != build_details.status:
+ prevstatus[build_id] = build_details.status
+ print(" {0} Build {2}: {1}".format(
+ now.strftime("%H:%M:%S"),
+ build_details.status, build_id))
+
+ if build_details.status in ["failed"]:
+ failed_ids.append(build_id)
+ if build_details.status in ["succeeded", "skipped",
+ "failed", "canceled"]:
+ done.add(build_id)
+ if build_details.status == "unknown":
+ raise copr_exceptions.CoprBuildException(
+ "Unknown status.")
+
+ if watched == done:
+ break
+
+ time.sleep(30)
+
+ if failed_ids:
+ raise copr_exceptions.CoprBuildException(
+ "Build(s) {0} failed.".format(
+ ", ".join(str(x) for x in failed_ids)))
+
+ except KeyboardInterrupt:
+ pass
+
+
+ @requires_api_auth
+ def action_build(self, args):
+ """ Method called when the 'build' action has been selected by the
+ user.
+
+ :param args: argparse arguments provided by the user
+
+ """
+ result = self.client.create_new_build(
+ projectname=args.copr, chroots=args.chroots, pkgs=args.pkgs,
+ memory=args.memory, timeout=args.timeout)
+ if result.output != "ok":
+ print(result.error)
+ return
+ print(result.message)
+
+ print("Created builds: {0}".format(" ".join(map(str, result.builds_list))))
+
+ if not args.nowait:
+ print("\n called")
+
+ self._watch_builds(result.builds_list)
+
+
+ @requires_api_auth
+ def action_create(self, args):
+ """ Method called when the 'create' action has been selected by the
+ user.
+
+ :param args: argparse arguments provided by the user
+
+ """
+
+ result = self.client.create_project(
+ projectname=args.name, description=args.description,
+ instructions=args.instructions, chroots=args.chroots,
+ repos=args.repos, initial_pkgs=args.initial_pkgs)
+ print(result.message)
+
+
+ @requires_api_auth
+ def action_delete(self, args):
+ """ Method called when the 'delete' action has been selected by the
+ user.
+
+ :param args: argparse arguments provided by the user
+ """
+ result = self.client.delete_project(projectname=args.copr)
+ print(result.message)
+
+
@check_username_presence
def action_list(self, args):
""" Method called when the 'list' action has been selected by the
@@ -180,10 +191,10 @@ class Commands(object):
"""
username = args.username or self.client.username
result = self.client.get_projects_list(username)
- #import ipdb; ipdb.set_trace()
+ # import ipdb; ipdb.set_trace()
if result.output != "ok":
- print("Error: {}".format(result.error))
- #print("Un-expected data returned, please report this issue")
+ print(result.error)
+ print("Un-expected data returned, please report this issue")
elif not result.projects_list:
print("No copr retrieved for user: '{0}'".format(username))
return
@@ -220,11 +231,11 @@ def setup_parser():
parser_list = subparsers.add_parser("list",
help="List all the copr of the "
"provided "
- )
+ )
parser_list.add_argument("username", nargs="?",
help="The username that you would like to "
- "list the copr of (defaults to current user)"
- )
+ "list the copr of (defaults to current user)"
+ )
parser_list.set_defaults(func="action_list")
# create the parser for the "create" command
@@ -250,21 +261,21 @@ def setup_parser():
parser_delete = subparsers.add_parser("delete",
help="Deletes the entire project")
parser_delete.add_argument("copr",
- help="Name of your project to be deleted.")
+ help="Name of your project to be deleted.")
parser_delete.set_defaults(func="action_delete")
# create the parser for the "build" command
parser_build = subparsers.add_parser("build",
help="Build packages to a "
- "specified copr")
+ "specified copr")
parser_build.add_argument("copr",
- help="The copr repo to build the package in. Can just name of project or even in format username/project."
- )
+ help="The copr repo to build the package in. Can just name of project or even in format username/project."
+ )
parser_build.add_argument("pkgs", nargs="+",
help="URL of packages to build")
parser_build.add_argument("-r", "--chroot", dest="chroots", action="append",
- help="If you don't need this build for all the project's chroots. You can use it several times for each"
- " chroot you need.")
+ help="If you don't need this build for all the project's chroots. You can use it several times for each"
+ " chroot you need.")
parser_build.add_argument("--memory", dest="memory",
help="")
parser_build.add_argument("--timeout", dest="timeout",
@@ -276,14 +287,14 @@ def setup_parser():
# create the parser for the "status" command
parser_build = subparsers.add_parser("status",
help="Get build status of build"
- " specified by its ID")
+ " specified by its ID")
parser_build.add_argument("build_id",
help="Build ID")
parser_build.set_defaults(func="action_status")
# create the parser for the "cancel" command
parser_build = subparsers.add_parser("cancel",
- help="Cancel build specified by its ID")
+ help="Cancel build specified by its ID")
parser_build.add_argument("build_id",
help="Build ID")
parser_build.set_defaults(func="action_cancel")
@@ -295,16 +306,17 @@ import logging
logging.basicConfig(
level=logging.DEBUG,
- format= '[%(asctime)s] {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s',
+ format='[%(asctime)s] {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s',
datefmt='%H:%M:%S'
)
log = logging.getLogger()
log.info("Logger initiated")
+
def main(argv=sys.argv[1:]):
try:
- #print(argv)
+ # print(argv)
# Set up parser for global args
parser = setup_parser()
# Parse the commandline
@@ -328,17 +340,13 @@ def main(argv=sys.argv[1:]):
except copr_exceptions.CoprException as e:
sys.stderr.write("\nError: {0}\n".format(e))
sys.exit(3)
- except ConfigParser.ParsingError as e:
- sys.stderr.write("\nError: {0}\n".format(e))
- sys.stderr.write("Lines in INI file should not be indented.\n")
- sys.exit(3)
except copr_exceptions.CoprBuildException as e:
sys.stderr.write("\nBuild error: {0}\n".format(e))
sys.exit(4)
except copr_exceptions.CoprUnknownResponseException as e:
sys.stderr.write("\nError: {0}\n".format(e))
sys.exit(5)
- # except Exception as e:
+ # except Exception as e:
# print "Error: {0}".format(e)
# sys.exit(100)
diff --git a/cli/run_tests.sh b/cli/run_tests.sh
index 0307c4d..326c3c4 100755
--- a/cli/run_tests.sh
+++ b/cli/run_tests.sh
@@ -1,4 +1,4 @@
#!/bin/sh
-PYTHONPATH=./copr_cli:$PYTHONPATH python -B -m pytest --cov-report term-missing --cov ./copr_cli/ tests
-#PYTHONPATH=./src:$PYTHONPATH python3 -B -m pytest --cov-report term-missing --cov ./src tests
+PYTHONPATH=./copr_cli:$PYTHONPATH python -B -m pytest -s --cov-report term-missing --cov ./copr_cli/ tests $@
+#PYTHONPATH=../python/:./copr_cli:$PYTHONPATH python3 -B -m pytest --cov-report term-missing --cov ./copr_cli/ $@
diff --git a/cli/tests/test_cli.py b/cli/tests/test_cli.py
index 64b83d1..3d38cc1 100644
--- a/cli/tests/test_cli.py
+++ b/cli/tests/test_cli.py
@@ -1,3 +1,4 @@
+import argparse
from collections import defaultdict
import json
from pprint import pprint
@@ -5,10 +6,13 @@ from _pytest.capture import capsys
import pytest
import six
+import time
import copr
from copr.client.parsers import ProjectListParser, CommonMsgErrorOutParser
from copr.client.responses import CoprResponse
-from copr.client.exceptions import CoprConfigException, CoprNoConfException
+from copr.client.exceptions import CoprConfigException, CoprNoConfException, \
+ CoprRequestException, CoprUnknownResponseException, CoprException, \
+ CoprBuildException
from copr.client import CoprClient
import copr_cli
from copr_cli.main import no_config_warning
@@ -25,9 +29,9 @@ else:
import logging
logging.basicConfig(
-level=logging.INFO,
-format= '[%(asctime)s] {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s',
-datefmt='%H:%M:%S'
+ level=logging.INFO,
+ format= '[%(asctime)s] {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s',
+ datefmt='%H:%M:%S'
)
log = logging.getLogger()
@@ -38,8 +42,123 @@ from copr_cli import main
@mock.patch('copr_cli.main.CoprClient')
+def test_error_keyboard_interrupt(mock_cc, capsys):
+
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.get_build_details.side_effect = KeyboardInterrupt()
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["status", "123"])
+
+ assert err.value.code == 1
+ stdout, stderr = capsys.readouterr()
+ assert "Interrupted by user" in stderr
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_error_copr_request(mock_cc, capsys):
+ error_msg = "error message"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.get_build_details.side_effect = CoprRequestException(error_msg)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["status", "123"])
+
+ assert err.value.code == 1
+ stdout, stderr = capsys.readouterr()
+ assert "Something went wrong" in stderr
+ assert error_msg in stderr
+
+(a)mock.patch('copr_cli.main.setup_parser')
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_error_argument_error(mock_cc, mock_setup_parser, capsys):
+ error_msg = "error message"
+
+ mock_client = MagicMock(no_config=False)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ mock_setup_parser.return_value.parse_args.side_effect = \
+ argparse.ArgumentTypeError(error_msg)
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["status", "123"])
+
+ assert err.value.code == 2
+ stdout, stderr = capsys.readouterr()
+ assert error_msg in stderr
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_error_no_args(mock_cc, capsys):
+ mock_client = MagicMock(no_config=False)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ for func_name in ["status", "build", "delete", "create"]:
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=[func_name])
+
+ assert err.value.code == 2
+
+ stdout, stderr = capsys.readouterr()
+ assert "usage: copr-cli" in stderr
+ assert "too few arguments" in stderr
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_error_copr_common_exception(mock_cc, capsys):
+ error_msg = "error message"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.get_build_details.side_effect = \
+ CoprException(error_msg)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["status", "123"])
+
+ assert err.value.code == 3
+ stdout, stderr = capsys.readouterr()
+ assert error_msg in stderr
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_error_copr_build_exception(mock_cc, capsys):
+ error_msg = "error message"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.side_effect = \
+ CoprBuildException(error_msg)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["build", "prj1", "src1"])
+
+ assert err.value.code == 4
+ stdout, stderr = capsys.readouterr()
+ assert error_msg in stderr
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_error_copr_unknown_response(mock_cc, capsys):
+ error_msg = "error message"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.get_build_details.side_effect = \
+ CoprUnknownResponseException(error_msg)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["status", "123"])
+
+ assert err.value.code == 5
+ stdout, stderr = capsys.readouterr()
+ assert error_msg in stderr
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
def test_cancel_build_no_config(mock_cc, capsys):
- #mock_from_config.return_value = CoprClient(dict(no_config=True))
mock_cc.create_from_file_config.side_effect = CoprNoConfException()
with pytest.raises(SystemExit) as err:
@@ -54,7 +173,6 @@ def test_cancel_build_no_config(mock_cc, capsys):
assert expected_warning in out
-
@mock.patch('copr_cli.main.CoprClient')
def test_cancel_build_response(mock_cc, capsys):
response_status = "foobar"
@@ -120,12 +238,384 @@ Name: ruby193
out, err = capsys.readouterr()
assert expected_output in out
-
expected_warning = no_config_warning
assert expected_warning in out
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_list_project_no_username(mock_cc, capsys):
+ mock_cc.create_from_file_config.side_effect = CoprNoConfException()
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["list"])
+
+ assert err.value.code == 6
+ out, err = capsys.readouterr()
+ assert "Pass username to command or create `~/.config/copr`" in out
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_list_project_no_username2(mock_cc, capsys):
+ mock_cc.create_from_file_config.return_value = CoprClient(defaultdict())
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["list"])
+
+ assert err.value.code == 6
+ out, err = capsys.readouterr()
+ assert "Pass username to command or add it to `~/.config/copr`" in out
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_list_project_error_msg(mock_cc, capsys):
+ mock_client = MagicMock(no_config=False, username="dummy")
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ mock_response = MagicMock(CoprResponse(None, None, None),
+ output="notok", error="error_msg",
+ projects_list=[])
+
+ mock_client.get_projects_list.return_value = mock_response
+ main.main(argv=["list", "projectname"])
+ out, err = capsys.readouterr()
+ assert "error_msg" in out
+ assert "No copr retrieved for user: dummy"
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_list_project_empty_list(mock_cc, capsys):
+ mock_client = MagicMock(no_config=False, username="dummy")
+ mock_cc.create_from_file_config.return_value = mock_client
+ mock_response = MagicMock(CoprResponse(None, None, None),
+ output="ok", projects_list=[])
+
+ mock_client.get_projects_list.return_value = mock_response
+ main.main(argv=["list", "projectname"])
+
+ out, err = capsys.readouterr()
+ assert "error" not in out
+ assert "No copr retrieved for user: dummy"
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_status_response(mock_cc, capsys):
+ response_status = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.get_build_details.return_value = \
+ MagicMock(status=response_status)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ main.main(argv=["status", "123"])
+ out, err = capsys.readouterr()
+ assert "{}\n".format(response_status) in out
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_status_response_no_args(mock_cc, capsys):
+ mock_client = MagicMock(no_config=False)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=["status"])
+
+ assert err.value.code == 2
+
+ stdout, stderr = capsys.readouterr()
+ assert "usage: copr-cli" in stderr
+ assert "too few arguments" in stderr
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_delete_project(mock_cc, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.delete_project.return_value = \
+ MagicMock(message=response_message)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ main.main(argv=["delete", "foo"])
+ out, err = capsys.readouterr()
+ assert "{}\n".format(response_message) in out
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_create_project(mock_cc, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_project.return_value = \
+ MagicMock(message=response_message)
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ main.main(argv=[
+ "create", "foo",
+ "--chroot", "f20", "--chroot", "f21",
+ "--description", "desc string",
+ "--instructions", "instruction string",
+ "--repo", "repo1", "--repo", "repo2",
+ "--initial-pkgs", "pkg1"
+ ])
+
+ stdout, stderr = capsys.readouterr()
+
+ mock_client.create_project.assert_called_with(
+ projectname="foo", description="desc string",
+ instructions="instruction string", chroots=["f20", "f21"],
+ repos=["repo1", "repo2"], initial_pkgs=["pkg1"])
+
+ assert "{}\n".format(response_message) in stdout
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_create_build_no_wait_ok(mock_cc, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.return_value = MagicMock(output="ok", message=response_message)
+
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ main.main(argv=[
+ "build", "--nowait",
+ "copr_name", "http://example.com/pkgs.srpm"
+ ])
+
+ stdout, stderr = capsys.readouterr()
+ assert response_message in stdout
+ assert "Created builds" in stdout
+
+ assert not mock_client._watch_build.called
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_create_build_no_wait_error(mock_cc, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.return_value = MagicMock(output="notok", error=response_message)
+
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ main.main(argv=[
+ "build", "--nowait",
+ "copr_name", "http://example.com/pkgs.srpm"
+ ])
+
+ stdout, stderr = capsys.readouterr()
+ assert response_message in stdout
+
+ assert not mock_client._watch_build.called
+
+(a)mock.patch('copr_cli.main.time')
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_create_build_wait_succeeded_no_sleep(mock_cc, mock_time, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.return_value = MagicMock(
+ output="ok",
+ message=response_message,
+ builds_list=[
+ MagicMock(build_id=x)
+ for x in range(3)
+ ])
+ mock_client.get_build_details.return_value = MagicMock(
+ status="succeeded", output="ok"
+ )
+ mock_cc.create_from_file_config.return_value = mock_client
+ main.main(argv=[
+ "build",
+ "copr_name", "http://example.com/pkgs.srpm"
+ ])
+
+ stdout, stderr = capsys.readouterr()
+
+ assert response_message in stdout
+ assert "Created builds" in stdout
+ assert "Watching build" in stdout
+ assert not mock_time.sleep.called
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_create_build_wait_error_status(mock_cc, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.return_value = MagicMock(
+ output="ok",
+ message=response_message,
+ builds_list=[
+ MagicMock(build_id=x)
+ for x in ["1", "2", "3"]
+ ])
+ mock_client.get_build_details.return_value = MagicMock(
+ output="notok"
+ )
+ mock_cc.create_from_file_config.return_value = mock_client
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=[
+ "build",
+ "copr_name", "http://example.com/pkgs.srpm"
+ ])
+ assert err.value.code == 1
+
+
+ stdout, stderr = capsys.readouterr()
+ assert response_message in stdout
+ assert "Created builds" in stdout
+ assert "Watching build" in stdout
+
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_create_build_wait_unknown_build_status(mock_cc, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.return_value = MagicMock(
+ output="ok",
+ message=response_message,
+ builds_list=[
+ MagicMock(build_id=x)
+ for x in ["1", "2", "3"]
+ ])
+ mock_client.get_build_details.return_value = MagicMock(
+ output="ok", status="unknown"
+ )
+ mock_cc.create_from_file_config.return_value = mock_client
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=[
+ "build",
+ "copr_name", "http://example.com/pkgs.srpm"
+ ])
+ assert err.value.code == 1
+
+ stdout, stderr = capsys.readouterr()
+ assert response_message in stdout
+ assert "Created builds" in stdout
+ assert "Watching build" in stdout
+
+(a)mock.patch('copr_cli.main.CoprClient')
+def test_create_build_wait_keyboard_interrupt(mock_cc, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.return_value = MagicMock(
+ output="ok",
+ message=response_message,
+ builds_list=[
+ MagicMock(build_id=x)
+ for x in ["1", "2", "3"]
+ ])
+ mock_client.get_build_details.side_effect = KeyboardInterrupt
+
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ main.main(argv=[
+ "build",
+ "copr_name", "http://example.com/pkgs.srpm"
+ ])
+
+ stdout, stderr = capsys.readouterr()
+ assert response_message in stdout
+ assert "Created builds" in stdout
+ assert "Watching build" in stdout
+
+(a)mock.patch('copr_cli.main.time')
+(a)mock.patch('copr_cli.main.CoprClient')
+class TestCreateBuild(object):
+ def test_create_build_wait_succeeded_complex(self, mock_cc, mock_time, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.return_value = MagicMock(
+ output="ok",
+ message=response_message,
+ builds_list=[
+ MagicMock(build_id=x)
+ for x in range(3)
+ ])
+
+ self.stage = 0
+ def incr(*args, **kwargs):
+ self.stage += 1
+
+ def result_map(build_id, *args, **kwargs):
+ if self.stage == 0:
+ return MagicMock(status="pending", output="ok")
+ elif self.stage == 1:
+ smap = {0: "pending", 1: "starting", 2: "running"}
+ return MagicMock(status=smap[build_id], output="ok")
+ elif self.stage == 2:
+ smap = {0: "starting", 1: "running", 2: "succeeded"}
+ return MagicMock(status=smap[build_id], output="ok")
+ elif self.stage == 3:
+ smap = {0: "skipped", 1: "succeeded", 2: "succeeded"}
+ return MagicMock(status=smap[build_id], output="ok")
+
+ mock_time.sleep.side_effect = incr
+
+ mock_client.get_build_details.side_effect = result_map
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ main.main(argv=[
+ "build",
+ "copr_name", "http://example.com/pkgs.srpm"
+ ])
+
+ stdout, stderr = capsys.readouterr()
+
+ assert response_message in stdout
+ assert "Created builds" in stdout
+ assert "Watching build" in stdout
+ assert len(mock_time.sleep.call_args_list) == 3
+
+ def test_create_build_wait_failed_complex(self, mock_cc, mock_time, capsys):
+ response_message = "foobar"
+
+ mock_client = MagicMock(no_config=False)
+ mock_client.create_new_build.return_value = MagicMock(
+ output="ok",
+ message=response_message,
+ builds_list=[
+ MagicMock(build_id=x)
+ for x in range(3)
+ ])
+
+ self.stage = 0
+ def incr(*args, **kwargs):
+ self.stage += 1
+
+ def result_map(build_id, *args, **kwargs):
+ if self.stage == 0:
+ return MagicMock(status="pending", output="ok")
+ elif self.stage == 1:
+ smap = {0: "pending", 1: "starting", 2: "running"}
+ return MagicMock(status=smap[build_id], output="ok")
+ elif self.stage == 2:
+ smap = {0: "failed", 1: "running", 2: "succeeded"}
+ return MagicMock(status=smap[build_id], output="ok")
+ elif self.stage == 3:
+ smap = {0: "failed", 1: "failed", 2: "succeeded"}
+ return MagicMock(status=smap[build_id], output="ok")
+
+ mock_time.sleep.side_effect = incr
+
+ mock_client.get_build_details.side_effect = result_map
+ mock_cc.create_from_file_config.return_value = mock_client
+
+ with pytest.raises(SystemExit) as err:
+ main.main(argv=[
+ "build",
+ "copr_name", "http://example.com/pkgs.srpm"
+ ])
+
+ stdout, stderr = capsys.readouterr()
+
+ assert response_message in stdout
+ assert "Created builds" in stdout
+ assert "Watching build" in stdout
+ assert "Build(s) 0, 1 failed" in stderr
+ assert len(mock_time.sleep.call_args_list) == 3