-
Notifications
You must be signed in to change notification settings - Fork 20
/
main_test.py
145 lines (104 loc) · 4.41 KB
/
main_test.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
145
import os
import json
import unittest
import pytest
from botocore.exceptions import ClientError
from mock import patch, MagicMock
from greengo import greengo
# Do not override the production state, work with testing state.
greengo.STATE_FILE = '.gg_state_test.json'
state = None
@pytest.fixture(scope="session", autouse=True)
def load_state(request):
# Load test state once per session.
global state
with open('tests/test_state.json', 'r') as f:
state = greengo.State(json.load(f))
class SessionFixture():
region_name = 'moon-darkside'
def __init__(self):
self.mock = MagicMock()
self.mock.describe_endpoint = MagicMock(
return_value={'endpointAddress': "xxx.iot.moon-darkside.amazonaws.com"})
def client(self, name):
return self.mock
def rinse(d):
# Overriding greengo.rinse so it won't give `KeyError` for missed `ResponseMetadata`
return d
@patch('greengo.greengo.rinse', rinse)
class GroupCommandTest(unittest.TestCase):
def setUp(self):
with patch.object(greengo.session, 'Session', SessionFixture):
self.gg = greengo.GroupCommands()
def tearDown(self):
try:
os.remove(greengo.STATE_FILE)
except OSError:
pass
def test_create_subscriptions(self):
self.gg.state = greengo.State(state.copy())
self.gg.state.pop('Subscriptions')
self.gg._gg.create_subscription_definition = MagicMock(return_value=state['Subscriptions'])
self.gg._gg.get_subscription_definition_version = MagicMock(
return_value=state['Subscriptions']['LatestVersionDetails'])
self.gg.create_subscriptions()
# TODO: assert calls
def test_create_subscriptions_empty(self):
self.gg.group.pop('Subscriptions')
self.gg.create_subscriptions()
def test_remove_subscriptions(self):
self.gg._gg.delete_subscription_definition = MagicMock(return_value=state['Subscriptions'])
self.gg.state = greengo.State(state.copy())
self.gg.remove_subscriptions()
self.assertFalse(self.gg.state.get('Subscriptions'), "Subscriptions shall be removed")
def test_create_group_version_fullset(self):
self.gg.state = greengo.State(state.copy())
m = MagicMock()
self.gg._gg.create_group_version = m
self.gg.create_group_version()
args, kwargs = m.call_args
self.assertEqual(len(kwargs), 6) # TODO: Refine expected kwarg count
def test_create_group_version_subset(self):
self.gg.state = greengo.State(state.copy())
self.gg.state.pop('Subscriptions')
self.gg.state.pop('FunctionDefinition')
m = MagicMock()
self.gg._gg.create_group_version = m
self.gg.create_group_version()
args, kwargs = m.call_args
self.assertEqual(len(kwargs), 4) # TODO: Refine expected kwarg count
def test_create_resources(self):
self.gg.group.pop('Resources')
self.gg.create_resources()
def test_remove_resources(self):
self.gg._gg.delete_resource_definition = MagicMock(return_value=state['Resources'])
self.gg.state = greengo.State(state.copy())
self.gg.remove_resources()
self.assertFalse(self.gg.state.get('Resources'), "Resources shall be removed")
def test_create_loggers(self):
self.gg.group.pop('Loggers')
self.gg.create_loggers()
def test_remove_loggers(self):
self.gg._gg.delete_resource_definition = MagicMock(return_value=state['Loggers'])
self.gg.state = greengo.State(state.copy())
self.gg.remove_loggers()
self.assertFalse(self.gg.state.get('Loggers'), "Loggers shall be removed")
@patch('greengo.greengo.rinse', rinse)
class LambdaTest(unittest.TestCase):
def setUp(self):
with patch.object(greengo.session, 'Session', SessionFixture):
self.gg = greengo.GroupCommands()
def tearDown(self):
try:
os.remove(greengo.STATE_FILE)
except OSError:
pass
def test_create_lambdas_empty(self):
self.gg.group.pop('Lambdas')
self.gg.create_lambdas() # Doesn't blow up
def test_role_exists(self):
error = ClientError(
error_response={'Error': {'Code': 'EntityAlreadyExists'}},
operation_name='CreateRole')
self.gg._iam.create_role = MagicMock(side_effect=error)
self.gg._default_lambda_role_arn() # Doesn't blow up