← Back to branch summary

~vishvananda/nova/lp827024

« back to all changes in this revision

Viewing changes to nova/tests/api/openstack/test_servers.py

  • Committer: Tarmac
  • Author(s): Chris Behrens
  • Date: 2011-08-09 21:09:10 UTC
  • mfrom: (1250.2.55 servers-search)
  • Revision ID: tarmac-20110809210910-ihhdtu9u7hjfos6i
This adds the servers search capabilities defined in the OS API v1.1 spec.. and more for admins.

For users, flavor=, image=, status=, and name= can be specified.  name= supports regular expression matching.
Most other options are ignored.  (things outside of the spec like 'recurse_zones' and 'reservation_id' still work, also)

If admin_api is enabled and context is an admin: along with the above, one can specify ip= and ip6= which will do regular expression matching.  Also, any other 'Instance' column name can be specified, so you can do regexp matching there as well.  Unknown Instance columns are ignored.

Also fixes up fixed_ip=, making a 404 returned vs a 500 error... and handling this properly with zone recursion as well.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/api/openstack/test_servers.py
236
236
        fakes.stub_out_key_pair_funcs(self.stubs)
237
237
        fakes.stub_out_image_service(self.stubs)
238
238
        self.stubs.Set(utils, 'gen_uuid', fake_gen_uuid)
239
 
        self.stubs.Set(nova.db.api, 'instance_get_all', return_servers)
 
239
        self.stubs.Set(nova.db.api, 'instance_get_all_by_filters',
 
240
                return_servers)
240
241
        self.stubs.Set(nova.db.api, 'instance_get', return_server_by_id)
241
242
        self.stubs.Set(nova.db, 'instance_get_by_uuid',
242
243
                       return_server_by_uuid)
1098
1099
        self.assertEqual(res.status_int, 400)
1099
1100
        self.assertTrue(res.body.find('marker param') > -1)
1100
1101
 
 
1102
    def test_get_servers_with_bad_option_v1_0(self):
 
1103
        # 1.0 API ignores unknown options
 
1104
        def fake_get_all(compute_self, context, search_opts=None):
 
1105
            return [stub_instance(100)]
 
1106
 
 
1107
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1108
 
 
1109
        req = webob.Request.blank('/v1.0/servers?unknownoption=whee')
 
1110
        res = req.get_response(fakes.wsgi_app())
 
1111
        self.assertEqual(res.status_int, 200)
 
1112
        servers = json.loads(res.body)['servers']
 
1113
        self.assertEqual(len(servers), 1)
 
1114
        self.assertEqual(servers[0]['id'], 100)
 
1115
 
 
1116
    def test_get_servers_with_bad_option_v1_1(self):
 
1117
        # 1.1 API also ignores unknown options
 
1118
        def fake_get_all(compute_self, context, search_opts=None):
 
1119
            return [stub_instance(100)]
 
1120
 
 
1121
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1122
 
 
1123
        req = webob.Request.blank('/v1.1/servers?unknownoption=whee')
 
1124
        res = req.get_response(fakes.wsgi_app())
 
1125
        self.assertEqual(res.status_int, 200)
 
1126
        servers = json.loads(res.body)['servers']
 
1127
        self.assertEqual(len(servers), 1)
 
1128
        self.assertEqual(servers[0]['id'], 100)
 
1129
 
 
1130
    def test_get_servers_allows_image_v1_1(self):
 
1131
        def fake_get_all(compute_self, context, search_opts=None):
 
1132
            self.assertNotEqual(search_opts, None)
 
1133
            self.assertTrue('image' in search_opts)
 
1134
            self.assertEqual(search_opts['image'], '12345')
 
1135
            return [stub_instance(100)]
 
1136
 
 
1137
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1138
        self.flags(allow_admin_api=False)
 
1139
 
 
1140
        req = webob.Request.blank('/v1.1/servers?image=12345')
 
1141
        res = req.get_response(fakes.wsgi_app())
 
1142
        # The following assert will fail if either of the asserts in
 
1143
        # fake_get_all() fail
 
1144
        self.assertEqual(res.status_int, 200)
 
1145
        servers = json.loads(res.body)['servers']
 
1146
        self.assertEqual(len(servers), 1)
 
1147
        self.assertEqual(servers[0]['id'], 100)
 
1148
 
 
1149
    def test_get_servers_allows_flavor_v1_1(self):
 
1150
        def fake_get_all(compute_self, context, search_opts=None):
 
1151
            self.assertNotEqual(search_opts, None)
 
1152
            self.assertTrue('flavor' in search_opts)
 
1153
            # flavor is an integer ID
 
1154
            self.assertEqual(search_opts['flavor'], '12345')
 
1155
            return [stub_instance(100)]
 
1156
 
 
1157
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1158
        self.flags(allow_admin_api=False)
 
1159
 
 
1160
        req = webob.Request.blank('/v1.1/servers?flavor=12345')
 
1161
        res = req.get_response(fakes.wsgi_app())
 
1162
        # The following assert will fail if either of the asserts in
 
1163
        # fake_get_all() fail
 
1164
        self.assertEqual(res.status_int, 200)
 
1165
        servers = json.loads(res.body)['servers']
 
1166
        self.assertEqual(len(servers), 1)
 
1167
        self.assertEqual(servers[0]['id'], 100)
 
1168
 
 
1169
    def test_get_servers_allows_status_v1_1(self):
 
1170
        def fake_get_all(compute_self, context, search_opts=None):
 
1171
            self.assertNotEqual(search_opts, None)
 
1172
            self.assertTrue('state' in search_opts)
 
1173
            self.assertEqual(set(search_opts['state']),
 
1174
                    set([power_state.RUNNING, power_state.BLOCKED]))
 
1175
            return [stub_instance(100)]
 
1176
 
 
1177
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1178
        self.flags(allow_admin_api=False)
 
1179
 
 
1180
        req = webob.Request.blank('/v1.1/servers?status=active')
 
1181
        res = req.get_response(fakes.wsgi_app())
 
1182
        # The following assert will fail if either of the asserts in
 
1183
        # fake_get_all() fail
 
1184
        self.assertEqual(res.status_int, 200)
 
1185
        servers = json.loads(res.body)['servers']
 
1186
        self.assertEqual(len(servers), 1)
 
1187
        self.assertEqual(servers[0]['id'], 100)
 
1188
 
 
1189
    def test_get_servers_invalid_status_v1_1(self):
 
1190
        """Test getting servers by invalid status"""
 
1191
 
 
1192
        self.flags(allow_admin_api=False)
 
1193
 
 
1194
        req = webob.Request.blank('/v1.1/servers?status=running')
 
1195
        res = req.get_response(fakes.wsgi_app())
 
1196
        # The following assert will fail if either of the asserts in
 
1197
        # fake_get_all() fail
 
1198
        self.assertEqual(res.status_int, 400)
 
1199
        self.assertTrue(res.body.find('Invalid server status') > -1)
 
1200
 
 
1201
    def test_get_servers_allows_name_v1_1(self):
 
1202
        def fake_get_all(compute_self, context, search_opts=None):
 
1203
            self.assertNotEqual(search_opts, None)
 
1204
            self.assertTrue('name' in search_opts)
 
1205
            self.assertEqual(search_opts['name'], 'whee.*')
 
1206
            return [stub_instance(100)]
 
1207
 
 
1208
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1209
        self.flags(allow_admin_api=False)
 
1210
 
 
1211
        req = webob.Request.blank('/v1.1/servers?name=whee.*')
 
1212
        res = req.get_response(fakes.wsgi_app())
 
1213
        # The following assert will fail if either of the asserts in
 
1214
        # fake_get_all() fail
 
1215
        self.assertEqual(res.status_int, 200)
 
1216
        servers = json.loads(res.body)['servers']
 
1217
        self.assertEqual(len(servers), 1)
 
1218
        self.assertEqual(servers[0]['id'], 100)
 
1219
 
 
1220
    def test_get_servers_unknown_or_admin_options1(self):
 
1221
        """Test getting servers by admin-only or unknown options.
 
1222
        This tests when admin_api is off.  Make sure the admin and
 
1223
        unknown options are stripped before they get to
 
1224
        compute_api.get_all()
 
1225
        """
 
1226
 
 
1227
        self.flags(allow_admin_api=False)
 
1228
 
 
1229
        def fake_get_all(compute_self, context, search_opts=None):
 
1230
            self.assertNotEqual(search_opts, None)
 
1231
            # Allowed by user
 
1232
            self.assertTrue('name' in search_opts)
 
1233
            self.assertTrue('status' in search_opts)
 
1234
            # Allowed only by admins with admin API on
 
1235
            self.assertFalse('ip' in search_opts)
 
1236
            self.assertFalse('unknown_option' in search_opts)
 
1237
            return [stub_instance(100)]
 
1238
 
 
1239
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1240
 
 
1241
        query_str = "name=foo&ip=10.*&status=active&unknown_option=meow"
 
1242
        req = webob.Request.blank('/v1.1/servers?%s' % query_str)
 
1243
        # Request admin context
 
1244
        context = nova.context.RequestContext('testuser', 'testproject',
 
1245
                is_admin=True)
 
1246
        res = req.get_response(fakes.wsgi_app(fake_auth_context=context))
 
1247
        # The following assert will fail if either of the asserts in
 
1248
        # fake_get_all() fail
 
1249
        self.assertEqual(res.status_int, 200)
 
1250
        servers = json.loads(res.body)['servers']
 
1251
        self.assertEqual(len(servers), 1)
 
1252
        self.assertEqual(servers[0]['id'], 100)
 
1253
 
 
1254
    def test_get_servers_unknown_or_admin_options2(self):
 
1255
        """Test getting servers by admin-only or unknown options.
 
1256
        This tests when admin_api is on, but context is a user.
 
1257
        Make sure the admin and unknown options are stripped before
 
1258
        they get to compute_api.get_all()
 
1259
        """
 
1260
 
 
1261
        self.flags(allow_admin_api=True)
 
1262
 
 
1263
        def fake_get_all(compute_self, context, search_opts=None):
 
1264
            self.assertNotEqual(search_opts, None)
 
1265
            # Allowed by user
 
1266
            self.assertTrue('name' in search_opts)
 
1267
            self.assertTrue('status' in search_opts)
 
1268
            # Allowed only by admins with admin API on
 
1269
            self.assertFalse('ip' in search_opts)
 
1270
            self.assertFalse('unknown_option' in search_opts)
 
1271
            return [stub_instance(100)]
 
1272
 
 
1273
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1274
 
 
1275
        query_str = "name=foo&ip=10.*&status=active&unknown_option=meow"
 
1276
        req = webob.Request.blank('/v1.1/servers?%s' % query_str)
 
1277
        # Request admin context
 
1278
        context = nova.context.RequestContext('testuser', 'testproject',
 
1279
                is_admin=False)
 
1280
        res = req.get_response(fakes.wsgi_app(fake_auth_context=context))
 
1281
        # The following assert will fail if either of the asserts in
 
1282
        # fake_get_all() fail
 
1283
        self.assertEqual(res.status_int, 200)
 
1284
        servers = json.loads(res.body)['servers']
 
1285
        self.assertEqual(len(servers), 1)
 
1286
        self.assertEqual(servers[0]['id'], 100)
 
1287
 
 
1288
    def test_get_servers_unknown_or_admin_options3(self):
 
1289
        """Test getting servers by admin-only or unknown options.
 
1290
        This tests when admin_api is on and context is admin.
 
1291
        All options should be passed through to compute_api.get_all()
 
1292
        """
 
1293
 
 
1294
        self.flags(allow_admin_api=True)
 
1295
 
 
1296
        def fake_get_all(compute_self, context, search_opts=None):
 
1297
            self.assertNotEqual(search_opts, None)
 
1298
            # Allowed by user
 
1299
            self.assertTrue('name' in search_opts)
 
1300
            self.assertTrue('status' in search_opts)
 
1301
            # Allowed only by admins with admin API on
 
1302
            self.assertTrue('ip' in search_opts)
 
1303
            self.assertTrue('unknown_option' in search_opts)
 
1304
            return [stub_instance(100)]
 
1305
 
 
1306
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1307
 
 
1308
        query_str = "name=foo&ip=10.*&status=active&unknown_option=meow"
 
1309
        req = webob.Request.blank('/v1.1/servers?%s' % query_str)
 
1310
        # Request admin context
 
1311
        context = nova.context.RequestContext('testuser', 'testproject',
 
1312
                is_admin=True)
 
1313
        res = req.get_response(fakes.wsgi_app(fake_auth_context=context))
 
1314
        # The following assert will fail if either of the asserts in
 
1315
        # fake_get_all() fail
 
1316
        self.assertEqual(res.status_int, 200)
 
1317
        servers = json.loads(res.body)['servers']
 
1318
        self.assertEqual(len(servers), 1)
 
1319
        self.assertEqual(servers[0]['id'], 100)
 
1320
 
 
1321
    def test_get_servers_admin_allows_ip_v1_1(self):
 
1322
        """Test getting servers by ip with admin_api enabled and
 
1323
        admin context
 
1324
        """
 
1325
        self.flags(allow_admin_api=True)
 
1326
 
 
1327
        def fake_get_all(compute_self, context, search_opts=None):
 
1328
            self.assertNotEqual(search_opts, None)
 
1329
            self.assertTrue('ip' in search_opts)
 
1330
            self.assertEqual(search_opts['ip'], '10\..*')
 
1331
            return [stub_instance(100)]
 
1332
 
 
1333
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1334
 
 
1335
        req = webob.Request.blank('/v1.1/servers?ip=10\..*')
 
1336
        # Request admin context
 
1337
        context = nova.context.RequestContext('testuser', 'testproject',
 
1338
                is_admin=True)
 
1339
        res = req.get_response(fakes.wsgi_app(fake_auth_context=context))
 
1340
        # The following assert will fail if either of the asserts in
 
1341
        # fake_get_all() fail
 
1342
        self.assertEqual(res.status_int, 200)
 
1343
        servers = json.loads(res.body)['servers']
 
1344
        self.assertEqual(len(servers), 1)
 
1345
        self.assertEqual(servers[0]['id'], 100)
 
1346
 
 
1347
    def test_get_servers_admin_allows_ip6_v1_1(self):
 
1348
        """Test getting servers by ip6 with admin_api enabled and
 
1349
        admin context
 
1350
        """
 
1351
        self.flags(allow_admin_api=True)
 
1352
 
 
1353
        def fake_get_all(compute_self, context, search_opts=None):
 
1354
            self.assertNotEqual(search_opts, None)
 
1355
            self.assertTrue('ip6' in search_opts)
 
1356
            self.assertEqual(search_opts['ip6'], 'ffff.*')
 
1357
            return [stub_instance(100)]
 
1358
 
 
1359
        self.stubs.Set(nova.compute.API, 'get_all', fake_get_all)
 
1360
 
 
1361
        req = webob.Request.blank('/v1.1/servers?ip6=ffff.*')
 
1362
        # Request admin context
 
1363
        context = nova.context.RequestContext('testuser', 'testproject',
 
1364
                is_admin=True)
 
1365
        res = req.get_response(fakes.wsgi_app(fake_auth_context=context))
 
1366
        # The following assert will fail if either of the asserts in
 
1367
        # fake_get_all() fail
 
1368
        self.assertEqual(res.status_int, 200)
 
1369
        servers = json.loads(res.body)['servers']
 
1370
        self.assertEqual(len(servers), 1)
 
1371
        self.assertEqual(servers[0]['id'], 100)
 
1372
 
1101
1373
    def _setup_for_create_instance(self):
1102
1374
        """Shared implementation for tests below that create instance"""
1103
1375
        def instance_create(context, inst):
1665
1937
    def test_get_all_server_details_v1_0(self):
1666
1938
        req = webob.Request.blank('/v1.0/servers/detail')
1667
1939
        res = req.get_response(fakes.wsgi_app())
 
1940
        self.assertEqual(res.status_int, 200)
1668
1941
        res_dict = json.loads(res.body)
1669
1942
 
1670
1943
        for i, s in enumerate(res_dict['servers']):
1720
1993
            return [stub_instance(i, 'fake', 'fake', None, None, i % 2)
1721
1994
                    for i in xrange(5)]
1722
1995
 
1723
 
        self.stubs.Set(nova.db.api, 'instance_get_all_by_project',
 
1996
        self.stubs.Set(nova.db.api, 'instance_get_all_by_filters',
1724
1997
            return_servers_with_host)
1725
1998
 
1726
1999
        req = webob.Request.blank('/v1.0/servers/detail')