test_handlers_clusters.py 文件源码

def test_cluster_upgrade_create(self):
        """
        Verify creating a cluster upgrade.
        """
        # 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]

            manager.get.side_effect = (
                make_new(CLUSTER_WITH_FLAT_HOST),
                Exception,
                MagicMock(StoreHandlerManager),
                make_new(CLUSTER_UPGRADE))

            # Verify with creation
            body = self.simulate_request(
                '/api/v0/cluster/development/upgrade', method='PUT')
            self.assertEquals(falcon.HTTP_201, self.srmock.status)
            result = json.loads(body[0])
            self.assertEquals('in_process', result['status'])
            self.assertEquals([], result['upgraded'])
            self.assertEquals([], result['in_process'])