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
|
# Copyright (c) 2016 Canonical Ltd
#
# 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 random
import string
import unittest
import uuid
from integration.busybox import create_busybox_image
from pylxd import exceptions
from pylxd.client import Client
class IntegrationTestCase(unittest.TestCase):
"""A base test case for pylxd integration tests."""
def setUp(self):
super(IntegrationTestCase, self).setUp()
self.client = Client()
self.lxd = self.client.api
def generate_object_name(self):
"""Generate a random object name."""
# Underscores are not allowed in container names.
test = self.id().split('.')[-1].replace('_', '')
rando = str(uuid.uuid1()).split('-')[-1]
return '{}-{}'.format(test, rando)
def create_container(self):
"""Create a container in lxd."""
fingerprint, alias = self.create_image()
name = self.generate_object_name()
machine = {
'name': name,
'architecture': '2',
'profiles': ['default'],
'ephemeral': False,
'config': {'limits.cpu': '2'},
'source': {'type': 'image',
'alias': alias},
}
result = self.lxd['containers'].post(json=machine)
operation_uuid = result.json()['operation'].split('/')[-1]
result = self.lxd.operations[operation_uuid].wait.get()
self.addCleanup(self.delete_container, name)
return name
def delete_container(self, name, enforce=False):
"""Delete a container in lxd."""
# enforce is a hack. There's a race somewhere in the delete.
# To ensure we don't get an infinite loop, let's count.
count = 0
try:
result = self.lxd['containers'][name].delete()
except exceptions.LXDAPIException as e:
if e.response.status_code in (400, 404):
return
raise
while enforce and result.status_code == 404 and count < 10:
try:
result = self.lxd['containers'][name].delete()
except exceptions.LXDAPIException as e:
if e.response.status_code in (400, 404):
return
raise
count += 1
try:
operation_uuid = result.json()['operation'].split('/')[-1]
result = self.lxd.operations[operation_uuid].wait.get()
except KeyError:
pass # 404 cases are okay.
def create_image(self):
"""Create an image in lxd."""
path, fingerprint = create_busybox_image()
with open(path, 'rb') as f:
headers = {
'X-LXD-Public': '1',
}
response = self.lxd.images.post(data=f.read(), headers=headers)
operation_uuid = response.json()['operation'].split('/')[-1]
self.lxd.operations[operation_uuid].wait.get()
alias = self.generate_object_name()
response = self.lxd.images.aliases.post(json={
'description': '',
'target': fingerprint,
'name': alias
})
self.addCleanup(self.delete_image, fingerprint)
return fingerprint, alias
def delete_image(self, fingerprint):
"""Delete an image in lxd."""
try:
self.lxd.images[fingerprint].delete()
except exceptions.LXDAPIException as e:
if e.response.status_code == 404:
return
raise
def create_profile(self):
"""Create a profile."""
name = self.generate_object_name()
config = {'limits.memory': '1GB'}
self.lxd.profiles.post(json={
'name': name,
'config': config
})
return name
def delete_profile(self, name):
"""Delete a profile."""
try:
self.lxd.profiles[name].delete()
except exceptions.LXDAPIException as e:
if e.response.status_code == 404:
return
raise
def create_network(self):
# get interface name in format xxx0
name = ''.join(random.sample(string.ascii_lowercase, 3)) + '0'
self.lxd.networks.post(json={
'name': name,
'config': {},
})
return name
def delete_network(self, name):
try:
self.lxd.networks[name].delete()
except exceptions.NotFound:
pass
def assertCommon(self, response):
"""Assert common LXD responses.
LXD responses are relatively standard. This function makes assertions
to all those standards.
"""
self.assertEqual(response.status_code, response.json()['status_code'])
self.assertEqual(
['metadata', 'operation', 'status', 'status_code', 'type'],
sorted(response.json().keys()))
|
