-
Notifications
You must be signed in to change notification settings - Fork 814
/
Copy pathtest_orchestrator.py
144 lines (107 loc) · 4.33 KB
/
test_orchestrator.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
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
# stdlib
import unittest
# 3rd party
import mock
import requests # noqa: F401
# project
from utils.orchestrator import BaseUtil
from utils.dockerutil import DockerUtil
CO_ID = 1234
class MockResponse:
"""
Helper class to mock a json response from requests
"""
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
return self.json_data
class TestBaseUtil(unittest.TestCase):
class DummyUtil(BaseUtil):
def _get_cacheable_tags(self, cid, co=None):
return ["test:tag"]
class NeedLabelsUtil(BaseUtil):
def __init__(self):
BaseUtil.__init__(self)
self.needs_inspect_labels = True
def _get_cacheable_tags(self, cid, co=None):
return ["test:tag"]
class NeedEnvUtil(BaseUtil):
def __init__(self):
BaseUtil.__init__(self)
self.needs_inspect_config = True
def _get_cacheable_tags(self, cid, co=None):
return ["test:tag"]
@mock.patch('docker.Client.__init__')
def test_extract_tags(self, mock_init):
mock_init.return_value = None
dummy = self.DummyUtil()
dummy.reset_cache()
self.assertEqual(["test:tag"], dummy.get_container_tags(cid=CO_ID))
@mock.patch('docker.Client.__init__')
def test_cache_invalidation_event(self, mock_init):
mock_init.return_value = None
dummy = self.DummyUtil()
dummy.reset_cache()
dummy.get_container_tags(cid=CO_ID)
self.assertTrue(CO_ID in dummy._container_tags_cache)
EVENT = {'status': 'die', 'id': CO_ID}
dummy.invalidate_cache([EVENT])
self.assertFalse(CO_ID in dummy._container_tags_cache)
@mock.patch('docker.Client.__init__')
def test_reset_cache(self, mock_init):
mock_init.return_value = None
dummy = self.DummyUtil()
dummy.reset_cache()
dummy.get_container_tags(cid=CO_ID)
self.assertTrue(CO_ID in dummy._container_tags_cache)
dummy.reset_cache()
self.assertFalse(CO_ID in dummy._container_tags_cache)
def test_auto_inspect(self):
du = DockerUtil()
du._client = mock.MagicMock()
mock_inspect = mock.MagicMock(name='inspect_container', return_value = {'RepoTags': ["redis:3.2"], 'RepoDigests': []})
du._client.inspect_container = mock_inspect
dummy = self.NeedLabelsUtil()
dummy.reset_cache()
dummy.get_container_tags(cid=CO_ID)
mock_inspect.assert_called_once()
def test_no_inspect_if_cached(self):
du = DockerUtil()
du._client = mock.MagicMock()
mock_inspect = mock.MagicMock(name='inspect_container', return_value = {'RepoTags': ["redis:3.2"], 'RepoDigests': []})
du._client.inspect_container = mock_inspect
dummy = self.NeedLabelsUtil()
dummy.reset_cache()
dummy.get_container_tags(cid=CO_ID)
mock_inspect.assert_called_once()
dummy.get_container_tags(cid=CO_ID)
mock_inspect.assert_called_once()
@mock.patch('docker.Client.inspect_container')
@mock.patch('docker.Client.__init__')
def test_no_useless_inspect(self, mock_init, mock_inspect):
mock_init.return_value = None
dummy = self.NeedLabelsUtil()
dummy.reset_cache()
co = {'Id': CO_ID, 'Created': 1, 'Labels': {}}
dummy.get_container_tags(co=co)
mock_inspect.assert_not_called()
dummy.get_container_tags(co=co)
mock_inspect.assert_not_called()
def test_auto_env_inspect(self):
du = DockerUtil()
du._client = mock.MagicMock()
mock_inspect = mock.MagicMock(name='inspect_container', return_value = {'RepoTags': ["redis:3.2"], 'RepoDigests': []})
du._client.inspect_container = mock_inspect
dummy = self.NeedEnvUtil()
dummy.reset_cache()
dummy.get_container_tags(co={'Id': CO_ID})
mock_inspect.assert_called_once()
@mock.patch('docker.Client.inspect_container')
@mock.patch('docker.Client.__init__')
def test_no_useless_env_inspect(self, mock_init, mock_inspect):
mock_init.return_value = None
dummy = self.NeedEnvUtil()
dummy.reset_cache()
dummy.get_container_tags(co={'Id': CO_ID, 'Config': {'Env': {1: 1}}})
mock_inspect.assert_not_called()