← Back to branch summary

~ubuntu-branches/ubuntu/trusty/swift/trusty-updates

« back to all changes in this revision

Viewing changes to test/unit/account/test_server.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, James Page, Chuck Short
  • Date: 2013-08-13 10:37:13 UTC
  • mfrom: (1.2.21)
  • Revision ID: package-import@ubuntu.com-20130813103713-1ctbx4zifyljs2aq
Tags: 1.9.1-0ubuntu1
[ James Page ]
* d/control: Update VCS fields for new branch locations.

[ Chuck Short ]
* New upstream release.

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/unit/account/test_server.py
25
25
 
26
26
from swift.common.swob import Request
27
27
from swift.account.server import AccountController, ACCOUNT_LISTING_LIMIT
28
 
from swift.common.utils import normalize_timestamp
 
28
from swift.common.utils import normalize_timestamp, replication, public
29
29
 
30
30
 
31
31
class TestAccountController(unittest.TestCase):
145
145
        self.controller.PUT(req)
146
146
        req = Request.blank('/sda1/p/a', environ={'REQUEST_METHOD': 'DELETE',
147
147
                                                  'HTTP_X_TIMESTAMP': '1'})
148
 
        resp = self.controller.DELETE(req)
 
148
        resp = req.get_response(self.controller)
149
149
        req = Request.blank('/sda1/p/a', environ={'REQUEST_METHOD': 'HEAD'})
150
 
        resp = self.controller.HEAD(req)
 
150
        resp = req.get_response(self.controller)
151
151
        self.assertEquals(resp.status_int, 404)
152
152
        self.assertEquals(resp.headers['X-Account-Status'], 'Deleted')
153
153
 
231
231
        format = '%D1%BD%8A9'  # invalid UTF-8; should be %E1%BD%8A9 (E -> D)
232
232
        req = Request.blank('/sda1/p/a?format=' + format,
233
233
                            environ={'REQUEST_METHOD': 'HEAD'})
234
 
        resp = self.controller.HEAD(req)
 
234
        resp = req.get_response(self.controller)
235
235
        self.assertEquals(resp.status_int, 400)
236
236
 
237
237
    def test_PUT_not_found(self):
1251
1251
        for format in ('xml', 'json'):
1252
1252
            req = Request.blank('/sda1/p/a?format=%s' % format,
1253
1253
                                environ={'REQUEST_METHOD': 'GET'})
1254
 
            resp = self.controller.GET(req)
 
1254
            resp = req.get_response(self.controller)
1255
1255
            self.assertEquals(resp.status_int, 200)
1256
1256
 
1257
1257
    def test_params_utf8(self):
1260
1260
                      'format'):
1261
1261
            req = Request.blank('/sda1/p/a?%s=\xce' % param,
1262
1262
                                environ={'REQUEST_METHOD': 'GET'})
1263
 
            resp = self.controller.GET(req)
 
1263
            resp = req.get_response(self.controller)
1264
1264
            self.assertEquals(resp.status_int, 400,
1265
1265
                              "%d on param %s" % (resp.status_int, param))
1266
1266
        # Good UTF8 sequence for delimiter, too long (1 byte delimiters only)
1347
1347
 
1348
1348
    def test_list_allowed_methods(self):
1349
1349
        """ Test list of allowed_methods """
1350
 
        methods = ['DELETE', 'PUT', 'HEAD', 'GET', 'REPLICATE', 'POST']
1351
 
        self.assertEquals(self.controller.allowed_methods, methods)
1352
 
 
1353
 
    def test_allowed_methods_from_configuration_file(self):
1354
 
        """
1355
 
        Test list of allowed_methods which
1356
 
        were set from configuration file.
1357
 
        """
1358
 
        conf = {'devices': self.testdir, 'mount_check': 'false'}
1359
 
        self.assertEquals(AccountController(conf).allowed_methods,
1360
 
                          ['DELETE', 'PUT', 'HEAD', 'GET', 'REPLICATE',
1361
 
                           'POST'])
1362
 
        conf['replication_server'] = 'True'
1363
 
        self.assertEquals(AccountController(conf).allowed_methods,
1364
 
                          ['REPLICATE'])
1365
 
        conf['replication_server'] = 'False'
1366
 
        self.assertEquals(AccountController(conf).allowed_methods,
1367
 
                          ['DELETE', 'PUT', 'HEAD', 'GET', 'POST'])
 
1350
        obj_methods = ['DELETE', 'PUT', 'HEAD', 'GET', 'POST']
 
1351
        repl_methods = ['REPLICATE']
 
1352
        for method_name in obj_methods:
 
1353
            method = getattr(self.controller, method_name)
 
1354
            self.assertFalse(hasattr(method, 'replication'))
 
1355
        for method_name in repl_methods:
 
1356
            method = getattr(self.controller, method_name)
 
1357
            self.assertEquals(method.replication, True)
1368
1358
 
1369
1359
    def test_correct_allowed_method(self):
1370
1360
        """
1374
1364
        inbuf = StringIO()
1375
1365
        errbuf = StringIO()
1376
1366
        outbuf = StringIO()
 
1367
        self.controller = AccountController(
 
1368
                {'devices': self.testdir, 'mount_check': 'false',
 
1369
                 'replication_server': 'false'})
1377
1370
 
1378
1371
        def start_response(*args):
1379
1372
            """ Sends args to outbuf """
1380
1373
            outbuf.writelines(args)
1381
1374
 
1382
 
        method = self.controller.allowed_methods[0]
1383
 
 
 
1375
        method = 'PUT'
1384
1376
        env = {'REQUEST_METHOD': method,
1385
1377
               'SCRIPT_NAME': '',
1386
1378
               'PATH_INFO': '/sda1/p/a/c',
1396
1388
               'wsgi.multiprocess': False,
1397
1389
               'wsgi.run_once': False}
1398
1390
 
1399
 
        answer = ['<html><h1>Method Not Allowed</h1><p>The method is not '
1400
 
                  'allowed for this resource.</p></html>']
1401
 
 
 
1391
        method_res = mock.MagicMock()
 
1392
        mock_method = public(lambda x: mock.MagicMock(return_value=method_res))
1402
1393
        with mock.patch.object(self.controller, method,
1403
 
                               return_value=mock.MagicMock()) as mock_method:
 
1394
                               new=mock_method):
 
1395
            mock_method.replication = False
1404
1396
            response = self.controller.__call__(env, start_response)
1405
 
            self.assertNotEqual(response, answer)
1406
 
            self.assertEqual(mock_method.call_count, 1)
 
1397
            self.assertEqual(response, method_res)
1407
1398
 
1408
1399
    def test_not_allowed_method(self):
1409
1400
        """
1413
1404
        inbuf = StringIO()
1414
1405
        errbuf = StringIO()
1415
1406
        outbuf = StringIO()
 
1407
        self.controller = AccountController(
 
1408
            {'devices': self.testdir, 'mount_check': 'false',
 
1409
             'replication_server': 'false'})
1416
1410
 
1417
1411
        def start_response(*args):
1418
1412
            """ Sends args to outbuf """
1419
1413
            outbuf.writelines(args)
1420
1414
 
1421
 
        method = self.controller.allowed_methods[0]
1422
 
 
 
1415
        method = 'PUT'
1423
1416
        env = {'REQUEST_METHOD': method,
1424
1417
               'SCRIPT_NAME': '',
1425
1418
               'PATH_INFO': '/sda1/p/a/c',
1437
1430
 
1438
1431
        answer = ['<html><h1>Method Not Allowed</h1><p>The method is not '
1439
1432
                  'allowed for this resource.</p></html>']
1440
 
 
 
1433
        mock_method = replication(public(lambda x: mock.MagicMock()))
1441
1434
        with mock.patch.object(self.controller, method,
1442
 
                               return_value=mock.MagicMock()) as mock_method:
1443
 
            self.controller.allowed_methods.remove(method)
 
1435
                               new=mock_method):
 
1436
            mock_method.replication = True
1444
1437
            response = self.controller.__call__(env, start_response)
1445
 
            self.assertEqual(mock_method.call_count, 0)
1446
1438
            self.assertEqual(response, answer)
1447
 
            self.controller.allowed_methods.append(method)
1448
1439
 
1449
1440
if __name__ == '__main__':
1450
1441
    unittest.main()