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
|
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Hewlett-Packard Development Company, L.P.
# All Rights Reserved.
#
# 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.
"""Tests for manipulating Ports via the DB API"""
import six
from ironic.common import exception
from ironic.common import utils as ironic_utils
from ironic.db import api as dbapi
from ironic.tests.db import base
from ironic.tests.db import utils
class DbPortTestCase(base.DbTestCase):
def setUp(self):
# This method creates a port for every test and
# replaces a test for creating a port.
super(DbPortTestCase, self).setUp()
self.dbapi = dbapi.get_instance()
ndict = utils.get_test_node()
self.n = self.dbapi.create_node(ndict)
self.p = utils.get_test_port()
def test_get_port_by_id(self):
self.dbapi.create_port(self.p)
res = self.dbapi.get_port(self.p['id'])
self.assertEqual(self.p['address'], res['address'])
def test_get_port_by_uuid(self):
self.dbapi.create_port(self.p)
res = self.dbapi.get_port(self.p['uuid'])
self.assertEqual(self.p['id'], res['id'])
def test_get_port_list(self):
uuids = []
for i in xrange(1, 6):
n = utils.get_test_port(id=i, uuid=ironic_utils.generate_uuid())
self.dbapi.create_port(n)
uuids.append(six.text_type(n['uuid']))
res = self.dbapi.get_port_list()
res_uuids = [r.uuid for r in res]
self.assertEqual(uuids.sort(), res_uuids.sort())
def test_get_port_by_address(self):
self.dbapi.create_port(self.p)
res = self.dbapi.get_port(self.p['address'])
self.assertEqual(self.p['id'], res['id'])
self.assertRaises(exception.PortNotFound,
self.dbapi.get_port, 99)
self.assertRaises(exception.PortNotFound,
self.dbapi.get_port, 'aa:bb:cc:dd:ee:ff')
self.assertRaises(exception.InvalidIdentity,
self.dbapi.get_port, 'not-a-mac')
def test_get_ports_by_node_id(self):
p = utils.get_test_port(node_id=self.n['id'])
self.dbapi.create_port(p)
res = self.dbapi.get_ports_by_node(self.n['id'])
self.assertEqual(self.p['address'], res[0]['address'])
def test_get_ports_by_node_uuid(self):
p = utils.get_test_port(node_id=self.n['id'])
self.dbapi.create_port(p)
res = self.dbapi.get_ports_by_node(self.n['uuid'])
self.assertEqual(self.p['address'], res[0]['address'])
def test_get_ports_by_node_that_does_not_exist(self):
self.dbapi.create_port(self.p)
self.assertRaises(exception.NodeNotFound,
self.dbapi.get_ports_by_node,
99)
self.assertRaises(exception.NodeNotFound,
self.dbapi.get_ports_by_node,
'12345678-9999-0000-aaaa-123456789012')
def test_destroy_port(self):
self.dbapi.create_port(self.p)
self.dbapi.destroy_port(self.p['id'])
self.assertRaises(exception.PortNotFound,
self.dbapi.destroy_port, self.p['id'])
def test_update_port(self):
self.dbapi.create_port(self.p)
old_address = self.p['address']
new_address = 'ff.ee.dd.cc.bb.aa'
self.assertNotEqual(old_address, new_address)
res = self.dbapi.update_port(self.p['id'], {'address': new_address})
self.assertEqual(new_address, res['address'])
def test_destroy_port_on_reserved_node(self):
p = self.dbapi.create_port(utils.get_test_port(node_id=self.n['id']))
uuid = self.n['uuid']
self.dbapi.reserve_nodes('fake-reservation', [uuid])
self.assertRaises(exception.NodeLocked,
self.dbapi.destroy_port, p['id'])
def test_update_port_on_reserved_node(self):
p = self.dbapi.create_port(utils.get_test_port(node_id=self.n['id']))
uuid = self.n['uuid']
self.dbapi.reserve_nodes('fake-reservation', [uuid])
new_address = 'ff.ee.dd.cc.bb.aa'
self.assertRaises(exception.NodeLocked,
self.dbapi.update_port, p['id'],
{'address': new_address})
|
