File: test_marathon_groups.py

package info (click to toggle)
python-dcos 0.2.0-10
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,440 kB
  • sloc: python: 8,196; sh: 194; makefile: 36
file content (215 lines) | stat: -rw-r--r-- 6,377 bytes parent folder | download | duplicates (4)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
import contextlib
import json

from .common import (assert_command, assert_lines, exec_command, show_app,
                     watch_all_deployments)

GOOD_GROUP = 'tests/data/marathon/groups/good.json'


def test_deploy_group():
    _deploy_group(GOOD_GROUP)
    _remove_group('test-group')


def test_group_list_table():
    with _group(GOOD_GROUP, 'test-group'):
        assert_lines(['dcos', 'marathon', 'group', 'list'], 3)


def test_validate_complicated_group_and_app():
    _deploy_group('tests/data/marathon/groups/complicated.json')
    _remove_group('test-group')


def test_optional_deploy_group():
    _deploy_group(GOOD_GROUP, False)
    _remove_group('test-group')


def test_add_existing_group():
    with _group(GOOD_GROUP, 'test-group'):
        with open(GOOD_GROUP) as fd:
            stderr = b"Group '/test-group' already exists\n"
            assert_command(['dcos', 'marathon', 'group', 'add'],
                           returncode=1,
                           stderr=stderr,
                           stdin=fd)


def test_show_group():
    with _group(GOOD_GROUP, 'test-group'):
        _show_group('test-group')


def test_add_bad_complicated_group():
    with open('tests/data/marathon/groups/complicated_bad.json') as fd:
        returncode, stdout, stderr = exec_command(
            ['dcos', 'marathon', 'group', 'add'],
            stdin=fd)

        err = b"""{
  "details": [
    {
      "errors": [
        "error.path.missing"
      ],
      "path": "/groups(0)/apps(0)/id"
    }
  ],
  "message": "Invalid JSON"
}
"""
        assert returncode == 1
        assert stdout == b''
        assert err in stderr


def test_update_group():
    with _group(GOOD_GROUP, 'test-group'):
        newapp = json.dumps([{"id": "appadded", "cmd": "sleep 0"}])
        appjson = "apps={}".format(newapp)
        returncode, stdout, stderr = exec_command(
            ['dcos', 'marathon', 'group', 'update', 'test-group/sleep',
                appjson])

        assert returncode == 0
        assert stdout.decode().startswith('Created deployment ')
        assert stderr == b''

        watch_all_deployments()
        show_app('test-group/sleep/appadded')


def test_update_group_from_stdin():
    with _group(GOOD_GROUP, 'test-group'):
        _update_group(
            'test-group',
            'tests/data/marathon/groups/update_good.json')
        show_app('test-group/updated')


def test_update_missing_group():
    assert_command(['dcos', 'marathon', 'group', 'update', 'missing-id'],
                   stderr=b"Error: Group '/missing-id' does not exist\n",
                   returncode=1)


def test_scale_group():
    _deploy_group('tests/data/marathon/groups/scale.json')
    returncode, stdout, stderr = exec_command(['dcos', 'marathon', 'group',
                                               'scale', 'scale-group', '2'])
    assert stderr == b''
    assert returncode == 0
    watch_all_deployments()
    returncode, stdout, stderr = exec_command(
        ['dcos', 'marathon', 'group', 'show',
         'scale-group'])
    res = json.loads(stdout.decode('utf-8'))

    assert res['groups'][0]['apps'][0]['instances'] == 2
    _remove_group('scale-group')


def test_scale_group_not_exist():
    returncode, stdout, stderr = exec_command(['dcos', 'marathon', 'group',
                                               'scale', 'scale-group', '2'])
    assert stderr == b''
    watch_all_deployments()
    returncode, stdout, stderr = exec_command(
        ['dcos', 'marathon', 'group', 'show',
         'scale-group'])
    res = json.loads(stdout.decode('utf-8'))

    assert len(res['apps']) == 0
    _remove_group('scale-group')


def test_scale_group_when_scale_factor_negative():
    _deploy_group('tests/data/marathon/groups/scale.json')
    returncode, stdout, stderr = exec_command(['dcos', 'marathon', 'group',
                                               'scale', 'scale-group', '-2'])
    assert b'Command not recognized' in stdout
    assert returncode == 1
    watch_all_deployments()
    _remove_group('scale-group')


def test_scale_group_when_scale_factor_not_float():
    _deploy_group('tests/data/marathon/groups/scale.json')
    returncode, stdout, stderr = exec_command(['dcos', 'marathon', 'group',
                                               'scale', 'scale-group', '1.a'])
    assert stderr == b'Error parsing string as float\n'
    assert returncode == 1
    watch_all_deployments()
    _remove_group('scale-group')


def _remove_group(group_id):
    assert_command(['dcos', 'marathon', 'group', 'remove', group_id])

    # Let's make sure that we don't return until the deployment has finished
    watch_all_deployments()


def _deploy_group(file_path, stdin=True):
    if stdin:
        with open(file_path) as fd:
            assert_command(['dcos', 'marathon', 'group', 'add'], stdin=fd)
    else:
        assert_command(['dcos', 'marathon', 'group', 'add', file_path])

    # Let's make sure that we don't return until the deployment has finished
    watch_all_deployments()


def _show_group(group_id, version=None):
    if version is None:
        cmd = ['dcos', 'marathon', 'group', 'show', group_id]
    else:
        cmd = ['dcos', 'marathon', 'group', 'show',
               '--group-version={}'.format(version), group_id]

    returncode, stdout, stderr = exec_command(cmd)

    result = json.loads(stdout.decode('utf-8'))

    assert returncode == 0
    assert isinstance(result, dict)
    assert result['id'] == '/' + group_id
    assert stderr == b''

    return result


def _update_group(group_id, file_path):
    with open(file_path) as fd:
        returncode, stdout, stderr = exec_command(
            ['dcos', 'marathon', 'group', 'update', group_id],
            stdin=fd)

        assert returncode == 0
        assert stdout.decode().startswith('Created deployment ')
        assert stderr == b''

    # Let's make sure that we don't return until the deployment has finished
    watch_all_deployments()


@contextlib.contextmanager
def _group(path, group_id):
    """Context manager that deploys a group on entrance, and removes it on
    exit.

    :param path: path to group's json definition
    :type path: str
    :param group_id: group id
    :type group_id: str
    :rtype: None
    """

    _deploy_group(path)
    try:
        yield
    finally:
        _remove_group(group_id)