test_handlers_hosts.py 文件源码

def test_host_create(self):
        """
        Verify creation of a Host.
        """
        with mock.patch('cherrypy.engine.publish') as _publish:
            manager = mock.MagicMock(StoreHandlerManager)
            _publish.return_value = [manager]

            manager.save.return_value = make_new(HOST)
            test_cluster = make_new(CLUSTER)
            manager.get.side_effect = (
                Exception,
                test_cluster,
                make_new(HOST),
                test_cluster,
                test_cluster)

            data = ('{"ssh_priv_key": "dGVzdAo=", "remote_user": "root",'
                    ' "cluster": "cluster"}')
            body = self.simulate_request(
                '/api/v0/host/10.2.0.2', method='PUT', body=data)
            self.assertEqual(self.srmock.status, falcon.HTTP_201)
            self.assertEqual(
                json.loads(INITIAL_HOST_JSON),
                json.loads(body[0]))

            # Make sure creation fails if the cluster doesn't exist
            manager.get.side_effect = Exception

            body = self.simulate_request(
                '/api/v0/host/10.2.0.2', method='PUT', body=data)
            self.assertEqual(self.srmock.status, falcon.HTTP_409)
            self.assertEqual({}, json.loads(body[0]))
            # Make sure creation is idempotent if the request parameters
            # agree with an existing host.
            manager.get.side_effect = (
                make_new(HOST),
                test_cluster)

            manager.get.side_effect = (
                make_new(HOST),
                make_new(CLUSTER_WITH_FLAT_HOST))
            body = self.simulate_request(
                '/api/v0/host/10.2.0.2', method='PUT', body=data)
            # datasource's set should not have been called
            self.assertEqual(self.srmock.status, falcon.HTTP_200)
            self.assertEqual(json.loads(HOST_JSON), json.loads(body[0]))

            # Make sure creation fails if the request parameters conflict
            # with an existing host.
            manager.get.side_effect = (
                make_new(HOST),
                make_new(CLUSTER_WITH_HOST))
            bad_data = '{"ssh_priv_key": "boguskey"}'
            body = self.simulate_request(
                '/api/v0/host/10.2.0.2', method='PUT', body=bad_data)
            # datasource's set should not have been called once
            self.assertEqual({}, json.loads(body[0]))
            self.assertEqual(self.srmock.status, falcon.HTTP_409)