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
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
|
from tests.compat import mock, unittest
from boto.ec2.snapshot import Snapshot
from boto.ec2.tag import Tag, TagSet
from boto.ec2.volume import Volume, AttachmentSet, VolumeAttribute
class VolumeTests(unittest.TestCase):
def setUp(self):
self.attach_data = AttachmentSet()
self.attach_data.id = 1
self.attach_data.instance_id = 2
self.attach_data.status = "some status"
self.attach_data.attach_time = 5
self.attach_data.device = "/dev/null"
self.volume_one = Volume()
self.volume_one.id = 1
self.volume_one.create_time = 5
self.volume_one.status = "one_status"
self.volume_one.size = "one_size"
self.volume_one.snapshot_id = 1
self.volume_one.attach_data = self.attach_data
self.volume_one.zone = "one_zone"
self.volume_two = Volume()
self.volume_two.connection = mock.Mock()
self.volume_two.id = 1
self.volume_two.create_time = 6
self.volume_two.status = "two_status"
self.volume_two.size = "two_size"
self.volume_two.snapshot_id = 2
self.volume_two.attach_data = None
self.volume_two.zone = "two_zone"
@mock.patch("boto.ec2.volume.TaggedEC2Object.startElement")
def test_startElement_calls_TaggedEC2Object_startElement_with_correct_args(self, startElement):
volume = Volume()
volume.startElement("some name", "some attrs", None)
startElement.assert_called_with(
"some name",
"some attrs",
None
)
@mock.patch("boto.ec2.volume.TaggedEC2Object.startElement")
def test_startElement_retval_not_None_returns_correct_thing(self, startElement):
tag_set = mock.Mock(TagSet)
startElement.return_value = tag_set
volume = Volume()
retval = volume.startElement(None, None, None)
self.assertEqual(retval, tag_set)
@mock.patch("boto.ec2.volume.TaggedEC2Object.startElement")
@mock.patch("boto.resultset.ResultSet")
def test_startElement_with_name_tagSet_calls_ResultSet(self, ResultSet, startElement):
startElement.return_value = None
result_set = mock.Mock(ResultSet([("item", Tag)]))
volume = Volume()
volume.tags = result_set
retval = volume.startElement("tagSet", None, None)
self.assertEqual(retval, volume.tags)
@mock.patch("boto.ec2.volume.TaggedEC2Object.startElement")
def test_startElement_with_name_attachmentSet_returns_AttachmentSet(self, startElement):
startElement.return_value = None
attach_data = AttachmentSet()
volume = Volume()
volume.attach_data = attach_data
retval = volume.startElement("attachmentSet", None, None)
self.assertEqual(retval, volume.attach_data)
@mock.patch("boto.ec2.volume.TaggedEC2Object.startElement")
def test_startElement_else_returns_None(self, startElement):
startElement.return_value = None
volume = Volume()
retval = volume.startElement("not tagSet or attachmentSet", None, None)
self.assertEqual(retval, None)
def check_that_attribute_has_been_set(self, name, value, attribute, obj_value=None):
volume = Volume()
volume.endElement(name, value, None)
expected_value = obj_value if obj_value is not None else value
self.assertEqual(getattr(volume, attribute), expected_value)
def test_endElement_sets_correct_attributes_with_values(self):
for arguments in [("volumeId", "some value", "id"),
("createTime", "some time", "create_time"),
("status", "some status", "status"),
("size", 5, "size"),
("snapshotId", 1, "snapshot_id"),
("availabilityZone", "some zone", "zone"),
("someName", "some value", "someName"),
("encrypted", "true", "encrypted", True)]:
self.check_that_attribute_has_been_set(*arguments)
def test_endElement_with_name_status_and_empty_string_value_doesnt_set_status(self):
volume = Volume()
volume.endElement("status", "", None)
self.assertNotEqual(volume.status, "")
def test_update_with_result_set_greater_than_0_updates_dict(self):
self.volume_two.connection.get_all_volumes.return_value = [self.volume_one]
self.volume_two.update()
assert all([self.volume_two.create_time == 5,
self.volume_two.status == "one_status",
self.volume_two.size == "one_size",
self.volume_two.snapshot_id == 1,
self.volume_two.attach_data == self.attach_data,
self.volume_two.zone == "one_zone"])
def test_update_with_validate_true_raises_value_error(self):
self.volume_one.connection = mock.Mock()
self.volume_one.connection.get_all_volumes.return_value = []
with self.assertRaisesRegexp(ValueError, "^1 is not a valid Volume ID$"):
self.volume_one.update(True)
def test_update_returns_status(self):
self.volume_one.connection = mock.Mock()
self.volume_one.connection.get_all_volumes.return_value = [self.volume_two]
retval = self.volume_one.update()
self.assertEqual(retval, "two_status")
def test_delete_calls_delete_volume(self):
self.volume_one.connection = mock.Mock()
self.volume_one.delete()
self.volume_one.connection.delete_volume.assert_called_with(
1,
dry_run=False
)
def test_attach_calls_attach_volume(self):
self.volume_one.connection = mock.Mock()
self.volume_one.attach("instance_id", "/dev/null")
self.volume_one.connection.attach_volume.assert_called_with(
1,
"instance_id",
"/dev/null",
dry_run=False
)
def test_detach_calls_detach_volume(self):
self.volume_one.connection = mock.Mock()
self.volume_one.detach()
self.volume_one.connection.detach_volume.assert_called_with(
1, 2, "/dev/null", False, dry_run=False)
def test_detach_with_no_attach_data(self):
self.volume_two.connection = mock.Mock()
self.volume_two.detach()
self.volume_two.connection.detach_volume.assert_called_with(
1, None, None, False, dry_run=False)
def test_detach_with_force_calls_detach_volume_with_force(self):
self.volume_one.connection = mock.Mock()
self.volume_one.detach(True)
self.volume_one.connection.detach_volume.assert_called_with(
1, 2, "/dev/null", True, dry_run=False)
def test_create_snapshot_calls_connection_create_snapshot(self):
self.volume_one.connection = mock.Mock()
self.volume_one.create_snapshot()
self.volume_one.connection.create_snapshot.assert_called_with(
1,
None,
dry_run=False
)
def test_create_snapshot_with_description(self):
self.volume_one.connection = mock.Mock()
self.volume_one.create_snapshot("some description")
self.volume_one.connection.create_snapshot.assert_called_with(
1,
"some description",
dry_run=False
)
def test_volume_state_returns_status(self):
retval = self.volume_one.volume_state()
self.assertEqual(retval, "one_status")
def test_attachment_state_returns_state(self):
retval = self.volume_one.attachment_state()
self.assertEqual(retval, "some status")
def test_attachment_state_no_attach_data_returns_None(self):
retval = self.volume_two.attachment_state()
self.assertEqual(retval, None)
def test_snapshots_returns_snapshots(self):
snapshot_one = Snapshot()
snapshot_one.volume_id = 1
snapshot_two = Snapshot()
snapshot_two.volume_id = 2
self.volume_one.connection = mock.Mock()
self.volume_one.connection.get_all_snapshots.return_value = [snapshot_one, snapshot_two]
retval = self.volume_one.snapshots()
self.assertEqual(retval, [snapshot_one])
def test_snapshots__with_owner_and_restorable_by(self):
self.volume_one.connection = mock.Mock()
self.volume_one.connection.get_all_snapshots.return_value = []
self.volume_one.snapshots("owner", "restorable_by")
self.volume_one.connection.get_all_snapshots.assert_called_with(
owner="owner", restorable_by="restorable_by", dry_run=False)
class AttachmentSetTests(unittest.TestCase):
def check_that_attribute_has_been_set(self, name, value, attribute):
attachment_set = AttachmentSet()
attachment_set.endElement(name, value, None)
self.assertEqual(getattr(attachment_set, attribute), value)
def test_endElement_with_name_volumeId_sets_id(self):
return self.check_that_attribute_has_been_set("volumeId", "some value", "id")
def test_endElement_with_name_instanceId_sets_instance_id(self):
return self.check_that_attribute_has_been_set("instanceId", 1, "instance_id")
def test_endElement_with_name_status_sets_status(self):
return self.check_that_attribute_has_been_set("status", "some value", "status")
def test_endElement_with_name_attachTime_sets_attach_time(self):
return self.check_that_attribute_has_been_set("attachTime", 5, "attach_time")
def test_endElement_with_name_device_sets_device(self):
return self.check_that_attribute_has_been_set("device", "/dev/null", "device")
def test_endElement_with_other_name_sets_other_name_attribute(self):
return self.check_that_attribute_has_been_set("someName", "some value", "someName")
class VolumeAttributeTests(unittest.TestCase):
def setUp(self):
self.volume_attribute = VolumeAttribute()
self.volume_attribute._key_name = "key_name"
self.volume_attribute.attrs = {"key_name": False}
def test_startElement_with_name_autoEnableIO_sets_key_name(self):
self.volume_attribute.startElement("autoEnableIO", None, None)
self.assertEqual(self.volume_attribute._key_name, "autoEnableIO")
def test_startElement_without_name_autoEnableIO_returns_None(self):
retval = self.volume_attribute.startElement("some name", None, None)
self.assertEqual(retval, None)
def test_endElement_with_name_value_and_value_true_sets_attrs_key_name_True(self):
self.volume_attribute.endElement("value", "true", None)
self.assertEqual(self.volume_attribute.attrs['key_name'], True)
def test_endElement_with_name_value_and_value_false_sets_attrs_key_name_False(self):
self.volume_attribute._key_name = "other_key_name"
self.volume_attribute.endElement("value", "false", None)
self.assertEqual(self.volume_attribute.attrs['other_key_name'], False)
def test_endElement_with_name_volumeId_sets_id(self):
self.volume_attribute.endElement("volumeId", "some_value", None)
self.assertEqual(self.volume_attribute.id, "some_value")
def test_endElement_with_other_name_sets_other_name_attribute(self):
self.volume_attribute.endElement("someName", "some value", None)
self.assertEqual(self.volume_attribute.someName, "some value")
if __name__ == "__main__":
unittest.main()
|
