test_handlers_clusters.py 文件源码

def test_cluster_restart_create(self):
        """
        Verify creating a cluster restart.
        """
        # Process is patched because we don't want to exec the subprocess
        # during unittesting
        with mock.patch('cherrypy.engine.publish') as _publish, \
             mock.patch('commissaire.handlers.clusters.Process'):

            manager = mock.MagicMock(StoreHandlerManager)
            _publish.return_value = [manager]

            test_cluster = make_new(CLUSTER_WITH_HOST)

            manager.get.side_effect = (
                test_cluster,
                Exception,
                MagicMock(StoreHandlerManager),
                make_new(CLUSTER_RESTART))

            body = self.simulate_request(
                '/api/v0/cluster/development/restart',
                method='PUT')
            self.assertEquals(falcon.HTTP_201, self.srmock.status)
            result = json.loads(body[0])
            self.assertEquals('in_process', result['status'])
            self.assertEquals([], result['restarted'])
            self.assertEquals([], result['in_process'])