This repository has been archived by the owner on Sep 16, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 151
/
Copy pathtest_models_unified_jobs.py
99 lines (86 loc) · 4.1 KB
/
test_models_unified_jobs.py
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
# Copyright 2016, Ansible by Red Hat
# Alan Rominger <[email protected]>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import time
import click
from six.moves import StringIO
import tower_cli
from tower_cli.api import client
from tests.compat import unittest, mock
def project_update_registration(t):
t.register_json('/project_updates/54/', {
'elapsed': 1335024000.0,
'failed': False,
'status': 'successful',
}, method='GET')
class StandardOutTests(unittest.TestCase):
"""
Test that standard out lookup and wrapper methods
"""
def setUp(self):
# Representative of a Unified Job Template
self.res = tower_cli.get_resource('project')
# Representative of a Unified Job
self.job_res = tower_cli.get_resource('job')
def test_lookup_stdout(self):
"Test the method that makes a call to get standard out."
with client.test_mode as t:
# note:
# 'foobar' = 'Zm9vYmFy' via
# base64.standard_b64encode('foobar'.encode('ascii'))
# but python versions can't all agree on how to trim the string
t.register_json(
'/project_updates/42/stdout/',
{'content': 'Zm9vYmFy'}, method='GET')
stdout = self.res.lookup_stdout(42, start_line=0, end_line=1)
assert 'foobar' in str(stdout)
def test_stdout(self):
"Test that printing standard out works with project-like things."
with mock.patch.object(type(self.res), 'lookup_stdout') as lookup:
with mock.patch.object(type(self.res), 'last_job_data') as job:
with mock.patch.object(click, 'echo') as mock_echo:
with mock.patch('sys.stdout', new=StringIO()) as fake_out:
job.return_value = {'id': 42}
lookup.return_value = 'foobar'
result = self.res.stdout(42, outfile=fake_out)
assert not result['changed']
mock_echo.assert_called_once_with('foobar', nl=1, file=fake_out)
def test_stdout_with_lookup(self):
"Test that unified job will be automatically looked up."
with mock.patch.object(type(self.job_res), 'lookup_stdout'):
with mock.patch.object(type(self.job_res), 'get') as get:
with mock.patch.object(click, 'echo'):
self.job_res.stdout(pk=None, name="test-proj")
get.assert_called_once_with(name="test-proj")
def test_call_wait_with_parent(self):
"Test auto-lookup of last job is called for wait"
with client.test_mode as t:
project_update_registration(t)
with mock.patch.object(type(self.res), 'last_job_data') as job:
job.return_value = {'id': 54}
with mock.patch.object(click, 'echo'):
with mock.patch.object(time, 'sleep'):
self.res.wait(pk=None, parent_pk=42)
job.assert_called_once_with(42)
def test_call_monitor_with_parent(self):
"Test auto-lookup when the monitor method is called"
with client.test_mode as t:
project_update_registration(t)
with mock.patch.object(type(self.res), 'last_job_data') as job:
job.return_value = {'id': 54}
with mock.patch.object(click, 'echo'):
with mock.patch.object(type(self.res), 'wait'):
with mock.patch.object(time, 'sleep'):
self.res.monitor(pk=None, parent_pk=42)
job.assert_called_once_with(42)