test_handlers_clusters.py 文件源码

def test_cluster_hosts_overwrite(self):
        """
        Verify overwriting a cluster host list.
        """
        with mock.patch('cherrypy.engine.publish') as _publish:
            manager = mock.MagicMock(StoreHandlerManager)
            _publish.return_value = [manager]

            # Verify setting host list works with a proper request
            manager.get.return_value = make_new(CLUSTER_WITH_FLAT_HOST)
            body = self.simulate_request(
                '/api/v0/cluster/development/hosts', method='PUT',
                body='{"old": ["10.2.0.2"], "new": ["10.2.0.2", "10.2.0.3"]}')
            self.assertEqual(falcon.HTTP_200, self.srmock.status)
            self.assertEqual({}, json.loads(body[0]))

            # Verify bad request (KeyError) returns the proper result
            manager.get.side_effect = KeyError
            body = self.simulate_request(
                '/api/v0/cluster/development/hosts', method='PUT',
                body='{"new": ["10.2.0.2", "10.2.0.3"]}')
            self.assertEqual(falcon.HTTP_400, self.srmock.status)
            self.assertEqual({}, json.loads(body[0]))

            # Verify bad request (TypeError) returns the proper result
            manager.get.side_effect = TypeError
            body = self.simulate_request(
                '/api/v0/cluster/development/hosts', method='PUT',
                body='["10.2.0.2", "10.2.0.3"]')
            self.assertEqual(falcon.HTTP_400, self.srmock.status)
            self.assertEqual({}, json.loads(body[0]))

            # Verify bad cluster name returns the proper result
            manager.get.side_effect = Exception
            body = self.simulate_request(
                '/api/v0/cluster/bogus/hosts', method='PUT',
                body='{"old": ["10.2.0.2"], "new": ["10.2.0.2", "10.2.0.3"]}')
            self.assertEqual(falcon.HTTP_404, self.srmock.status)
            self.assertEqual({}, json.loads(body[0]))

            # Verify host list conflict returns the proper result
            manager.get.side_effect = None
            body = self.simulate_request(
                '/api/v0/cluster/development/hosts', method='PUT',
                body='{"old": [], "new": ["10.2.0.2", "10.2.0.3"]}')
            self.assertEqual(falcon.HTTP_409, self.srmock.status)
            self.assertEqual({}, json.loads(body[0]))