test_hpe_plugin.py 文件源码

def broken_test_hpe_list_volume(self):
        name = 'test-list-volume'
        path = b"/VolumeDriver.Create"
        body = {u"Name": name}

        # Create a volume to be mounted
        headers = Headers({b"content-type": [b"application/json"]})
        body_producer = FileBodyProducer(BytesIO(dumps(body)))
        agent = Agent.usingEndpointFactory(reactor, HPEEndpointFactory())
        d = agent.request(b'POST', b"UNIX://localhost" + path, headers,
                          body_producer)
        d.addCallback(self.checkResponse, json.dumps({u"Err": ''}))
        d.addErrback(self.cbFailed)

        # List volumes
        expected = {u"Err": '',
                    u"Volumes": [{u"Mountpoint": '',
                                  u"Name": name}]}
        d.addCallback(self._list_volumes, name, expected)

        # Remove the previously created volume
        d.addCallback(self._remove_volume_callback, name)

        return d