test_hpe_plugin.py 文件源码

def test_hpe_get_volume(self):
        name = 'test-get-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)

        # Get the previously created volume
        expected = {u"Volume": {u"Status": {},
                                u"Mountpoint": '',
                                u"Name": name},
                    u"Err": ''}
        d.addCallback(self._get_volume, name, expected)

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