tests.test_cli() - Code Metrics - Inspection of "Merge branch 'release/0.2.0'" - fabfuel/ecs-deploy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( a30f7a...98e9e8 )

by Fabian

created 2016-03-19 18:12 UTC

tests.test_cli() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	4
rs	10
cc	3

import pytest
from click.testing import CliRunner
from mock.mock import patch

from ecs_deploy import cli
from ecs_deploy.cli import get_client
from ecs_deploy.ecs import EcsClient
from tests.test_ecs import EcsTestClient, CLUSTER_NAME, SERVICE_NAME


@pytest.fixture
def runner():
    return CliRunner()


@patch.object(EcsClient, '__init__')
def test_get_client(ecs_client):
    ecs_client.return_value = None
    client = get_client('access_key_id', 'secret_access_key', 'region', 'profile')
    ecs_client.assert_called_once_with('access_key_id', 'secret_access_key', 'region', 'profile')
    assert isinstance(client, EcsClient)


def test_ecs(runner):
    result = runner.invoke(cli.ecs)
    assert result.exit_code == 0
    assert not result.exception
    assert 'Usage: ecs [OPTIONS] COMMAND [ARGS]' in result.output
    assert '  deploy  ' in result.output
    assert '  scale   ' in result.output


@patch('ecs_deploy.cli.get_client')
def test_deploy_without_credentials(get_client, runner):
    get_client.return_value = EcsTestClient()
    result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME))
    assert result.exit_code == 1
    assert result.output == u'Unable to locate credentials. Configure credentials by running "aws configure".\n\n'


@patch('ecs_deploy.cli.get_client')
def test_deploy_with_invalid_cluster(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.deploy, ('unknown-cluster', SERVICE_NAME))
    assert result.exit_code == 1
    assert result.output == u'An error occurred (ClusterNotFoundException) when calling the DescribeServices ' \
                            u'operation: Cluster not found.\n\n'


@patch('ecs_deploy.cli.get_client')
def test_deploy_with_invalid_service(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.deploy, (CLUSTER_NAME, 'unknown-service'))
    assert result.exit_code == 1
    assert result.output == u'An error occurred when calling the DescribeServices operation: Service not found.\n\n'


@patch('ecs_deploy.cli.get_client')
def test_deploy(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME))
    assert result.exit_code == 0
    assert not result.exception
    assert u'Successfully created revision: 2' in result.output
    assert u'Successfully deregistered revision: 1' in result.output
    assert u'Successfully changed task definition to: test-task:2' in result.output
    assert u'Deployment successful' in result.output
    assert u"Updating task definition" not in result.output


@patch('ecs_deploy.cli.get_client')
def test_deploy_new_tag(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME, '-t', 'latest'))
    assert result.exit_code == 0
    assert not result.exception
    assert u"Updating task definition" in result.output
    assert u"Changed image of container 'webserver' to: webserver:latest (was: webserver:123)" in result.output
    assert u"Changed image of container 'application' to: application:latest (was: application:123)" in result.output
    assert u'Successfully created revision: 2' in result.output
    assert u'Successfully deregistered revision: 1' in result.output
    assert u'Successfully changed task definition to: test-task:2' in result.output
    assert u'Deployment successful' in result.output


@patch('ecs_deploy.cli.get_client')
def test_deploy_one_new_image(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME, '-i', 'application', 'application:latest'))
    assert result.exit_code == 0
    assert not result.exception
    assert u"Updating task definition" in result.output
    assert u"Changed image of container 'application' to: application:latest (was: application:123)" in result.output
    assert u'Successfully created revision: 2' in result.output
    assert u'Successfully deregistered revision: 1' in result.output
    assert u'Successfully changed task definition to: test-task:2' in result.output
    assert u'Deployment successful' in result.output


@patch('ecs_deploy.cli.get_client')
def test_deploy_two_new_images(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME, '-i', 'application', 'application:latest',
                                        '-i', 'webserver', 'webserver:latest'))
    assert result.exit_code == 0
    assert not result.exception
    assert u"Updating task definition" in result.output
    assert u"Changed image of container 'webserver' to: webserver:latest (was: webserver:123)" in result.output
    assert u"Changed image of container 'application' to: application:latest (was: application:123)" in result.output
    assert u'Successfully created revision: 2' in result.output
    assert u'Successfully deregistered revision: 1' in result.output
    assert u'Successfully changed task definition to: test-task:2' in result.output
    assert u'Deployment successful' in result.output


@patch('ecs_deploy.cli.get_client')
def test_deploy_one_new_command(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME, '-c', 'application', 'foobar'))
    assert result.exit_code == 0
    assert not result.exception
    assert u"Updating task definition" in result.output
    assert u"Changed command of container 'application' to: foobar (was: run)" in result.output
    assert u'Successfully created revision: 2' in result.output
    assert u'Successfully deregistered revision: 1' in result.output
    assert u'Successfully changed task definition to: test-task:2' in result.output
    assert u'Deployment successful' in result.output


@patch('ecs_deploy.cli.get_client')
def test_deploy_with_errors(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key', errors=True)
    result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME))
    assert result.exit_code == 1
    assert u"Deployment failed" in result.output
    assert u"ERROR: Service was unable to Lorem Ipsum" in result.output


@patch('ecs_deploy.cli.get_client')
def test_scale_without_credentials(get_client, runner):
    get_client.return_value = EcsTestClient()
    result = runner.invoke(cli.scale, (CLUSTER_NAME, SERVICE_NAME, '2'))
    assert result.exit_code == 1
    assert result.output == u'Unable to locate credentials. Configure credentials by running "aws configure".\n\n'


@patch('ecs_deploy.cli.get_client')
def test_scale_with_invalid_cluster(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.scale, ('unknown-cluster', SERVICE_NAME, '2'))
    assert result.exit_code == 1
    assert result.output == u'An error occurred (ClusterNotFoundException) when calling the DescribeServices ' \
                            u'operation: Cluster not found.\n\n'


@patch('ecs_deploy.cli.get_client')
def test_scale_with_invalid_service(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.scale, (CLUSTER_NAME, 'unknown-service', '2'))
    assert result.exit_code == 1
    assert result.output == u'An error occurred when calling the DescribeServices operation: Service not found.\n\n'


@patch('ecs_deploy.cli.get_client')
def test_scale(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
    result = runner.invoke(cli.scale, (CLUSTER_NAME, SERVICE_NAME, '2'))
    assert not result.exception
    assert result.exit_code == 0
    assert u"Successfully changed desired count to: 2" in result.output
    assert u"Scaling successful" in result.output


@patch('ecs_deploy.cli.get_client')
def test_scale_with_errors(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key', errors=True)
    result = runner.invoke(cli.scale, (CLUSTER_NAME, SERVICE_NAME, '2'))
    assert result.exit_code == 1
    assert u"Scaling failed" in result.output
    assert u"ERROR: Service was unable to Lorem Ipsum" in result.output


@patch('ecs_deploy.cli.get_client')
def test_scale_with_timeout(get_client, runner):
    get_client.return_value = EcsTestClient('acces_key', 'secret_key', wait=2)
    result = runner.invoke(cli.scale, (CLUSTER_NAME, SERVICE_NAME, '2', '--timeout', '1'))
    assert result.exit_code == 1
    assert u"Scaling failed (timeout)" in result.output


1			import pytest
2			from click.testing import CliRunner
3			from mock.mock import patch
4
5			from ecs_deploy import cli
6			from ecs_deploy.cli import get_client
7			from ecs_deploy.ecs import EcsClient
8			from tests.test_ecs import EcsTestClient, CLUSTER_NAME, SERVICE_NAME
9
10
11			@pytest.fixture
12			def runner():
13			return CliRunner()
14
15
16			@patch.object(EcsClient, '__init__')
17			def test_get_client(ecs_client):
18			ecs_client.return_value = None
19			client = get_client('access_key_id', 'secret_access_key', 'region', 'profile')
20			ecs_client.assert_called_once_with('access_key_id', 'secret_access_key', 'region', 'profile')
21			assert isinstance(client, EcsClient)
22
23
24			def test_ecs(runner):
25			result = runner.invoke(cli.ecs)
26			assert result.exit_code == 0
27			assert not result.exception
28			assert 'Usage: ecs [OPTIONS] COMMAND [ARGS]' in result.output
29			assert ' deploy ' in result.output
30			assert ' scale ' in result.output
31
32
33			@patch('ecs_deploy.cli.get_client')
34			def test_deploy_without_credentials(get_client, runner):
35			get_client.return_value = EcsTestClient()
36			result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME))
37			assert result.exit_code == 1
38			assert result.output == u'Unable to locate credentials. Configure credentials by running "aws configure".\n\n'
39
40
41			@patch('ecs_deploy.cli.get_client')
42			def test_deploy_with_invalid_cluster(get_client, runner):
43			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
44			result = runner.invoke(cli.deploy, ('unknown-cluster', SERVICE_NAME))
45			assert result.exit_code == 1
46			assert result.output == u'An error occurred (ClusterNotFoundException) when calling the DescribeServices ' \
47			u'operation: Cluster not found.\n\n'
48
49
50			@patch('ecs_deploy.cli.get_client')
51			def test_deploy_with_invalid_service(get_client, runner):
52			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
53			result = runner.invoke(cli.deploy, (CLUSTER_NAME, 'unknown-service'))
54			assert result.exit_code == 1
55			assert result.output == u'An error occurred when calling the DescribeServices operation: Service not found.\n\n'
56
57
58			@patch('ecs_deploy.cli.get_client')
59			def test_deploy(get_client, runner):
60			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
61			result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME))
62			assert result.exit_code == 0
63			assert not result.exception
64			assert u'Successfully created revision: 2' in result.output
65			assert u'Successfully deregistered revision: 1' in result.output
66			assert u'Successfully changed task definition to: test-task:2' in result.output
67			assert u'Deployment successful' in result.output
68			assert u"Updating task definition" not in result.output
69
70
71			@patch('ecs_deploy.cli.get_client')
72			def test_deploy_new_tag(get_client, runner):
73			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
74			result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME, '-t', 'latest'))
75			assert result.exit_code == 0
76			assert not result.exception
77			assert u"Updating task definition" in result.output
78			assert u"Changed image of container 'webserver' to: webserver:latest (was: webserver:123)" in result.output
79			assert u"Changed image of container 'application' to: application:latest (was: application:123)" in result.output
80			assert u'Successfully created revision: 2' in result.output
81			assert u'Successfully deregistered revision: 1' in result.output
82			assert u'Successfully changed task definition to: test-task:2' in result.output
83			assert u'Deployment successful' in result.output
84
85
86			@patch('ecs_deploy.cli.get_client')
87			def test_deploy_one_new_image(get_client, runner):
88			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
89			result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME, '-i', 'application', 'application:latest'))
90			assert result.exit_code == 0
91			assert not result.exception
92			assert u"Updating task definition" in result.output
93			assert u"Changed image of container 'application' to: application:latest (was: application:123)" in result.output
94			assert u'Successfully created revision: 2' in result.output
95			assert u'Successfully deregistered revision: 1' in result.output
96			assert u'Successfully changed task definition to: test-task:2' in result.output
97			assert u'Deployment successful' in result.output
98
99
100			@patch('ecs_deploy.cli.get_client')
101			def test_deploy_two_new_images(get_client, runner):
102			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
103			result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME, '-i', 'application', 'application:latest',
104			'-i', 'webserver', 'webserver:latest'))
105			assert result.exit_code == 0
106			assert not result.exception
107			assert u"Updating task definition" in result.output
108			assert u"Changed image of container 'webserver' to: webserver:latest (was: webserver:123)" in result.output
109			assert u"Changed image of container 'application' to: application:latest (was: application:123)" in result.output
110			assert u'Successfully created revision: 2' in result.output
111			assert u'Successfully deregistered revision: 1' in result.output
112			assert u'Successfully changed task definition to: test-task:2' in result.output
113			assert u'Deployment successful' in result.output
114
115
116			@patch('ecs_deploy.cli.get_client')
117			def test_deploy_one_new_command(get_client, runner):
118			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
119			result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME, '-c', 'application', 'foobar'))
120			assert result.exit_code == 0
121			assert not result.exception
122			assert u"Updating task definition" in result.output
123			assert u"Changed command of container 'application' to: foobar (was: run)" in result.output
124			assert u'Successfully created revision: 2' in result.output
125			assert u'Successfully deregistered revision: 1' in result.output
126			assert u'Successfully changed task definition to: test-task:2' in result.output
127			assert u'Deployment successful' in result.output
128
129
130			@patch('ecs_deploy.cli.get_client')
131			def test_deploy_with_errors(get_client, runner):
132			get_client.return_value = EcsTestClient('acces_key', 'secret_key', errors=True)
133			result = runner.invoke(cli.deploy, (CLUSTER_NAME, SERVICE_NAME))
134			assert result.exit_code == 1
135			assert u"Deployment failed" in result.output
136			assert u"ERROR: Service was unable to Lorem Ipsum" in result.output
137
138
139			@patch('ecs_deploy.cli.get_client')
140			def test_scale_without_credentials(get_client, runner):
141			get_client.return_value = EcsTestClient()
142			result = runner.invoke(cli.scale, (CLUSTER_NAME, SERVICE_NAME, '2'))
143			assert result.exit_code == 1
144			assert result.output == u'Unable to locate credentials. Configure credentials by running "aws configure".\n\n'
145
146
147			@patch('ecs_deploy.cli.get_client')
148			def test_scale_with_invalid_cluster(get_client, runner):
149			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
150			result = runner.invoke(cli.scale, ('unknown-cluster', SERVICE_NAME, '2'))
151			assert result.exit_code == 1
152			assert result.output == u'An error occurred (ClusterNotFoundException) when calling the DescribeServices ' \
153			u'operation: Cluster not found.\n\n'
154
155
156			@patch('ecs_deploy.cli.get_client')
157			def test_scale_with_invalid_service(get_client, runner):
158			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
159			result = runner.invoke(cli.scale, (CLUSTER_NAME, 'unknown-service', '2'))
160			assert result.exit_code == 1
161			assert result.output == u'An error occurred when calling the DescribeServices operation: Service not found.\n\n'
162
163
164			@patch('ecs_deploy.cli.get_client')
165			def test_scale(get_client, runner):
166			get_client.return_value = EcsTestClient('acces_key', 'secret_key')
167			result = runner.invoke(cli.scale, (CLUSTER_NAME, SERVICE_NAME, '2'))
168			assert not result.exception
169			assert result.exit_code == 0
170			assert u"Successfully changed desired count to: 2" in result.output
171			assert u"Scaling successful" in result.output
172
173
174			@patch('ecs_deploy.cli.get_client')
175			def test_scale_with_errors(get_client, runner):
176			get_client.return_value = EcsTestClient('acces_key', 'secret_key', errors=True)
177			result = runner.invoke(cli.scale, (CLUSTER_NAME, SERVICE_NAME, '2'))
178			assert result.exit_code == 1
179			assert u"Scaling failed" in result.output
180			assert u"ERROR: Service was unable to Lorem Ipsum" in result.output
181
182
183			@patch('ecs_deploy.cli.get_client')
184			def test_scale_with_timeout(get_client, runner):
185			get_client.return_value = EcsTestClient('acces_key', 'secret_key', wait=2)
186			result = runner.invoke(cli.scale, (CLUSTER_NAME, SERVICE_NAME, '2', '--timeout', '1'))
187			assert result.exit_code == 1
188			assert u"Scaling failed (timeout)" in result.output
189

fabfuel / ecs-deploy

Push — master ( a30f7a...98e9e8 )

tests.test_cli() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like