| # -*- coding: utf-8 -*- |
| # |
| # Copyright (c) The PyAMF Project. |
| # See LICENSE.txt for details. |
| |
| """ |
| Tests for Remoting client. |
| |
| @since: 0.1.0 |
| """ |
| |
| import unittest |
| import urllib2 |
| |
| import pyamf |
| from pyamf import remoting, util |
| from pyamf.remoting import client |
| |
| |
| class ServiceMethodProxyTestCase(unittest.TestCase): |
| def test_create(self): |
| x = client.ServiceMethodProxy('a', 'b') |
| |
| self.assertEqual(x.service, 'a') |
| self.assertEqual(x.name, 'b') |
| |
| def test_call(self): |
| tc = self |
| |
| class TestService(object): |
| def __init__(self, s, args): |
| self.service = s |
| self.args = args |
| |
| def _call(self, service, *args): |
| tc.assertTrue(self.service, service) |
| tc.assertTrue(self.args, args) |
| |
| x = client.ServiceMethodProxy(None, None) |
| ts = TestService(x, [1, 2, 3]) |
| x.service = ts |
| |
| x(1, 2, 3) |
| |
| def test_str(self): |
| x = client.ServiceMethodProxy('spam', 'eggs') |
| self.assertEqual(str(x), 'spam.eggs') |
| |
| x = client.ServiceMethodProxy('spam', None) |
| self.assertEqual(str(x), 'spam') |
| |
| |
| class ServiceProxyTestCase(unittest.TestCase): |
| def test_create(self): |
| x = client.ServiceProxy('spam', 'eggs') |
| |
| self.assertEqual(x._gw, 'spam') |
| self.assertEqual(x._name, 'eggs') |
| self.assertEqual(x._auto_execute, True) |
| |
| x = client.ServiceProxy('hello', 'world', True) |
| |
| self.assertEqual(x._gw, 'hello') |
| self.assertEqual(x._name, 'world') |
| self.assertEqual(x._auto_execute, True) |
| |
| x = client.ServiceProxy(ord, chr, False) |
| |
| self.assertEqual(x._gw, ord) |
| self.assertEqual(x._name, chr) |
| self.assertEqual(x._auto_execute, False) |
| |
| def test_getattr(self): |
| x = client.ServiceProxy(None, None) |
| y = x.spam |
| |
| self.assertTrue(isinstance(y, client.ServiceMethodProxy)) |
| self.assertEqual(y.name, 'spam') |
| |
| def test_call(self): |
| class DummyGateway(object): |
| def __init__(self, tc): |
| self.tc = tc |
| |
| def addRequest(self, method_proxy, *args): |
| self.tc.assertEqual(method_proxy, self.method_proxy) |
| self.tc.assertEqual(args, self.args) |
| |
| self.request = {'method_proxy': method_proxy, 'args': args} |
| return self.request |
| |
| def execute_single(self, request): |
| self.tc.assertEqual(request, self.request) |
| |
| return pyamf.ASObject(body=None, status=None) |
| |
| gw = DummyGateway(self) |
| x = client.ServiceProxy(gw, 'test') |
| y = x.spam |
| |
| gw.method_proxy = y |
| gw.args = () |
| |
| y() |
| gw.args = (1, 2, 3) |
| |
| y(1, 2, 3) |
| |
| def test_service_call(self): |
| class DummyGateway(object): |
| def __init__(self, tc): |
| self.tc = tc |
| |
| def addRequest(self, method_proxy, *args): |
| self.tc.assertEqual(method_proxy.service, self.x) |
| self.tc.assertEqual(method_proxy.name, None) |
| |
| return pyamf.ASObject(method_proxy=method_proxy, args=args) |
| |
| def execute_single(self, request): |
| return pyamf.ASObject(body=None, status=None) |
| |
| gw = DummyGateway(self) |
| x = client.ServiceProxy(gw, 'test') |
| gw.x = x |
| |
| x() |
| |
| def test_pending_call(self): |
| class DummyGateway(object): |
| def __init__(self, tc): |
| self.tc = tc |
| |
| def addRequest(self, method_proxy, *args): |
| self.tc.assertEqual(method_proxy, self.method_proxy) |
| self.tc.assertEqual(args, self.args) |
| |
| self.request = pyamf.ASObject(method_proxy=method_proxy, args=args) |
| |
| return self.request |
| |
| gw = DummyGateway(self) |
| x = client.ServiceProxy(gw, 'test', False) |
| y = x.eggs |
| |
| gw.method_proxy = y |
| gw.args = () |
| |
| res = y() |
| |
| self.assertEqual(id(gw.request), id(res)) |
| |
| def test_str(self): |
| x = client.ServiceProxy(None, 'test') |
| |
| self.assertEqual(str(x), 'test') |
| |
| def test_error(self): |
| class DummyGateway(object): |
| def __init__(self, tc): |
| self.tc = tc |
| |
| def addRequest(self, method_proxy, *args): |
| self.request = pyamf.ASObject(method_proxy=method_proxy, args=args) |
| |
| return self.request |
| |
| def execute_single(self, request): |
| body = remoting.ErrorFault(code='TypeError', description='foobar') |
| |
| return remoting.Response(status=remoting.STATUS_ERROR, body=body) |
| |
| gw = DummyGateway(self) |
| |
| proxy = client.ServiceProxy(gw, 'test') |
| |
| self.assertRaises(TypeError, proxy) |
| |
| |
| class RequestWrapperTestCase(unittest.TestCase): |
| def test_create(self): |
| x = client.RequestWrapper(1, 2, 3, 4) |
| |
| self.assertEqual(x.gw, 1) |
| self.assertEqual(x.id, 2) |
| self.assertEqual(x.service, 3) |
| self.assertEqual(x.args, (4,)) |
| |
| def test_str(self): |
| x = client.RequestWrapper(None, '/1', None, None) |
| |
| self.assertEqual(str(x), '/1') |
| |
| def test_null_response(self): |
| x = client.RequestWrapper(None, None, None, None) |
| |
| self.assertRaises(AttributeError, getattr, x, 'result') |
| |
| def test_set_response(self): |
| x = client.RequestWrapper(None, None, None, None) |
| |
| y = pyamf.ASObject(body='spam.eggs') |
| |
| x.setResponse(y) |
| |
| self.assertEqual(x.response, y) |
| self.assertEqual(x.result, 'spam.eggs') |
| |
| |
| class MockOpener(object): |
| """ |
| opener for urllib2.install_opener |
| """ |
| |
| def __init__(self, test, response=None): |
| self.test = test |
| self.response = response |
| |
| def open(self, request, data=None, timeout=None): |
| if self.response.code != 200: |
| raise urllib2.URLError(self.response.code) |
| |
| self.request = request |
| self.data = data |
| |
| return self.response |
| |
| |
| class MockHeaderCollection(object): |
| |
| def __init__(self, headers): |
| self.headers = headers |
| |
| def getheader(self, name): |
| return self.headers.get(name, None) |
| |
| def __repr__(self): |
| return repr(self.headers) |
| |
| |
| class MockResponse(object): |
| """ |
| """ |
| |
| headers = None |
| body = None |
| |
| def info(self): |
| return MockHeaderCollection(self.headers) |
| |
| def read(self, amount): |
| return self.body[0:amount] |
| |
| |
| class BaseServiceTestCase(unittest.TestCase): |
| """ |
| """ |
| |
| canned_response = ('\x00\x00\x00\x00\x00\x01\x00\x0b/1/onResult\x00' |
| '\x04null\x00\x00\x00\x00\n\x00\x00\x00\x03\x00?\xf0\x00\x00' |
| '\x00\x00\x00\x00\x00@\x00\x00\x00\x00\x00\x00\x00\x00@\x08\x00' |
| '\x00\x00\x00\x00\x00') |
| |
| headers = { |
| 'Content-Type': 'application/x-amf', |
| } |
| |
| def setUp(self): |
| unittest.TestCase.setUp(self) |
| |
| self.response = MockResponse() |
| self.opener = MockOpener(self, self.response) |
| |
| self.gw = client.RemotingService('http://example.org/amf-gateway', opener=self.opener.open) |
| |
| self.headers = self.__class__.headers.copy() |
| self.headers['Content-Length'] = len(self.canned_response) |
| |
| self.setResponse(200, self.canned_response, self.headers) |
| |
| def setResponse(self, status, body, headers=None): |
| self.response.code = status |
| self.response.body = body |
| self.response.headers = headers or { |
| 'Content-Type': remoting.CONTENT_TYPE |
| } |
| |
| |
| class RemotingServiceTestCase(BaseServiceTestCase): |
| """ |
| """ |
| |
| def test_create(self): |
| self.assertRaises(TypeError, client.RemotingService) |
| x = client.RemotingService('http://example.org') |
| |
| self.assertEqual(x.url, ('http', 'example.org', '', '', '', '')) |
| |
| # amf version |
| x = client.RemotingService('http://example.org', pyamf.AMF3) |
| self.assertEqual(x.amf_version, pyamf.AMF3) |
| |
| def test_schemes(self): |
| x = client.RemotingService('http://example.org') |
| self.assertEqual(x.url, ('http', 'example.org', '', '', '', '')) |
| |
| x = client.RemotingService('https://example.org') |
| self.assertEqual(x.url, ('https', 'example.org', '', '', '', '')) |
| |
| self.assertRaises(ValueError, client.RemotingService, |
| 'ftp://example.org') |
| |
| def test_port(self): |
| x = client.RemotingService('http://example.org:8080') |
| |
| self.assertEqual(x.url, ('http', 'example.org:8080', '', '', '', '')) |
| |
| def test_get_service(self): |
| x = client.RemotingService('http://example.org') |
| |
| y = x.getService('spam') |
| |
| self.assertTrue(isinstance(y, client.ServiceProxy)) |
| self.assertEqual(y._name, 'spam') |
| self.assertEqual(y._gw, x) |
| |
| self.assertRaises(TypeError, x.getService, 1) |
| |
| def test_add_request(self): |
| gw = client.RemotingService('http://spameggs.net') |
| |
| self.assertEqual(gw.request_number, 1) |
| self.assertEqual(gw.requests, []) |
| service = gw.getService('baz') |
| wrapper = gw.addRequest(service, 1, 2, 3) |
| |
| self.assertEqual(gw.requests, [wrapper]) |
| self.assertEqual(wrapper.gw, gw) |
| self.assertEqual(gw.request_number, 2) |
| self.assertEqual(wrapper.id, '/1') |
| self.assertEqual(wrapper.service, service) |
| self.assertEqual(wrapper.args, (1, 2, 3)) |
| |
| # add 1 arg |
| wrapper2 = gw.addRequest(service, None) |
| |
| self.assertEqual(gw.requests, [wrapper, wrapper2]) |
| self.assertEqual(wrapper2.gw, gw) |
| self.assertEqual(gw.request_number, 3) |
| self.assertEqual(wrapper2.id, '/2') |
| self.assertEqual(wrapper2.service, service) |
| self.assertEqual(wrapper2.args, (None,)) |
| |
| # add no args |
| wrapper3 = gw.addRequest(service) |
| |
| self.assertEqual(gw.requests, [wrapper, wrapper2, wrapper3]) |
| self.assertEqual(wrapper3.gw, gw) |
| self.assertEqual(gw.request_number, 4) |
| self.assertEqual(wrapper3.id, '/3') |
| self.assertEqual(wrapper3.service, service) |
| self.assertEqual(wrapper3.args, tuple()) |
| |
| def test_remove_request(self): |
| gw = client.RemotingService('http://spameggs.net') |
| self.assertEqual(gw.requests, []) |
| |
| service = gw.getService('baz') |
| wrapper = gw.addRequest(service, 1, 2, 3) |
| self.assertEqual(gw.requests, [wrapper]) |
| |
| gw.removeRequest(wrapper) |
| self.assertEqual(gw.requests, []) |
| |
| wrapper = gw.addRequest(service, 1, 2, 3) |
| self.assertEqual(gw.requests, [wrapper]) |
| |
| gw.removeRequest(service, 1, 2, 3) |
| self.assertEqual(gw.requests, []) |
| |
| self.assertRaises(LookupError, gw.removeRequest, service, 1, 2, 3) |
| |
| def test_get_request(self): |
| gw = client.RemotingService('http://spameggs.net') |
| |
| service = gw.getService('baz') |
| wrapper = gw.addRequest(service, 1, 2, 3) |
| |
| wrapper2 = gw.getRequest(str(wrapper)) |
| self.assertEqual(wrapper, wrapper2) |
| |
| wrapper2 = gw.getRequest('/1') |
| self.assertEqual(wrapper, wrapper2) |
| |
| wrapper2 = gw.getRequest(wrapper.id) |
| self.assertEqual(wrapper, wrapper2) |
| |
| def test_get_amf_request(self): |
| gw = client.RemotingService('http://example.org', pyamf.AMF3) |
| |
| service = gw.getService('baz') |
| method_proxy = service.gak |
| wrapper = gw.addRequest(method_proxy, 1, 2, 3) |
| |
| envelope = gw.getAMFRequest([wrapper]) |
| |
| self.assertEqual(envelope.amfVersion, pyamf.AMF3) |
| self.assertEqual(envelope.keys(), ['/1']) |
| |
| request = envelope['/1'] |
| self.assertEqual(request.target, 'baz.gak') |
| self.assertEqual(request.body, [1, 2, 3]) |
| |
| envelope2 = gw.getAMFRequest(gw.requests) |
| |
| self.assertEqual(envelope2.amfVersion, pyamf.AMF3) |
| self.assertEqual(envelope2.keys(), ['/1']) |
| |
| request = envelope2['/1'] |
| self.assertEqual(request.target, 'baz.gak') |
| self.assertEqual(request.body, [1, 2, 3]) |
| |
| def test_execute_single(self): |
| service = self.gw.getService('baz', auto_execute=False) |
| wrapper = service.gak() |
| |
| response = self.gw.execute_single(wrapper) |
| self.assertEqual(self.gw.requests, []) |
| |
| r = self.opener.request |
| |
| self.assertEqual(r.headers, { |
| 'Content-type': remoting.CONTENT_TYPE, |
| 'User-agent': client.DEFAULT_USER_AGENT |
| }) |
| self.assertEqual(r.get_method(), 'POST') |
| self.assertEqual(r.get_full_url(), 'http://example.org/amf-gateway') |
| |
| self.assertEqual(r.get_data(), '\x00\x00\x00\x00\x00\x01\x00\x07' |
| 'baz.gak\x00\x02/1\x00\x00\x00\x00\x0a\x00\x00\x00\x00') |
| |
| self.assertEqual(response.status, remoting.STATUS_OK) |
| self.assertEqual(response.body, [1, 2, 3]) |
| |
| def test_execute(self): |
| baz = self.gw.getService('baz', auto_execute=False) |
| spam = self.gw.getService('spam', auto_execute=False) |
| wrapper = baz.gak() |
| wrapper2 = spam.eggs() |
| |
| self.assertTrue(wrapper) |
| self.assertTrue(wrapper2) |
| |
| response = self.gw.execute() |
| self.assertTrue(response) |
| self.assertEqual(self.gw.requests, []) |
| |
| r = self.opener.request |
| |
| self.assertEqual(r.headers, { |
| 'Content-type': remoting.CONTENT_TYPE, |
| 'User-agent': client.DEFAULT_USER_AGENT |
| }) |
| self.assertEqual(r.get_method(), 'POST') |
| self.assertEqual(r.get_full_url(), 'http://example.org/amf-gateway') |
| |
| self.assertEqual(r.get_data(), '\x00\x00\x00\x00\x00\x02\x00\x07' |
| 'baz.gak\x00\x02/1\x00\x00\x00\x00\n\x00\x00\x00\x00\x00\tspam.' |
| 'eggs\x00\x02/2\x00\x00\x00\x00\n\x00\x00\x00\x00') |
| |
| def test_get_response(self): |
| self.setResponse(200, '\x00\x00\x00\x00\x00\x00\x00\x00') |
| |
| self.gw._getResponse(None) |
| |
| self.setResponse(404, '', {}) |
| |
| self.assertRaises(remoting.RemotingError, self.gw._getResponse, None) |
| |
| # bad content type |
| self.setResponse(200, '<html></html>', {'Content-Type': 'text/html'}) |
| |
| self.assertRaises(remoting.RemotingError, self.gw._getResponse, None) |
| |
| def test_credentials(self): |
| self.assertFalse('Credentials' in self.gw.headers) |
| self.gw.setCredentials('spam', 'eggs') |
| self.assertTrue('Credentials' in self.gw.headers) |
| self.assertEqual(self.gw.headers['Credentials'], |
| {'userid': u'spam', 'password': u'eggs'}) |
| |
| envelope = self.gw.getAMFRequest([]) |
| self.assertTrue('Credentials' in envelope.headers) |
| |
| cred = envelope.headers['Credentials'] |
| |
| self.assertEqual(cred, self.gw.headers['Credentials']) |
| |
| def test_append_url_header(self): |
| self.setResponse(200, '\x00\x00\x00\x01\x00\x12AppendToGatewayUrl' |
| '\x01\x00\x00\x00\x00\x02\x00\x05hello\x00\x00\x00\x00', { |
| 'Content-Type': 'application/x-amf'}) |
| |
| response = self.gw._getResponse(None) |
| self.assertTrue(response) |
| |
| self.assertEqual(self.gw.original_url, |
| 'http://example.org/amf-gatewayhello') |
| |
| def test_replace_url_header(self): |
| self.setResponse(200, '\x00\x00\x00\x01\x00\x11ReplaceGatewayUrl\x01' |
| '\x00\x00\x00\x00\x02\x00\x10http://spam.eggs\x00\x00\x00\x00', |
| {'Content-Type': 'application/x-amf'}) |
| |
| response = self.gw._getResponse(None) |
| self.assertTrue(response) |
| self.assertEqual(self.gw.original_url, 'http://spam.eggs') |
| |
| def test_add_http_header(self): |
| self.assertEqual(self.gw.http_headers, {}) |
| |
| self.gw.addHTTPHeader('ETag', '29083457239804752309485') |
| |
| self.assertEqual(self.gw.http_headers, { |
| 'ETag': '29083457239804752309485' |
| }) |
| |
| def test_remove_http_header(self): |
| self.gw.http_headers = { |
| 'Set-Cookie': 'foo-bar' |
| } |
| |
| self.gw.removeHTTPHeader('Set-Cookie') |
| |
| self.assertEqual(self.gw.http_headers, {}) |
| self.assertRaises(KeyError, self.gw.removeHTTPHeader, 'foo-bar') |
| |
| def test_http_request_headers(self): |
| self.gw.addHTTPHeader('ETag', '29083457239804752309485') |
| |
| expected_headers = { |
| 'Etag': '29083457239804752309485', |
| 'Content-type': 'application/x-amf', |
| 'User-agent': self.gw.user_agent |
| } |
| |
| self.setResponse(200, '\x00\x00\x00\x01\x00\x11ReplaceGatewayUrl' |
| '\x01\x00\x00\x00\x00\x02\x00\x10http://spam.eggs\x00\x00\x00\x00') |
| |
| self.gw.execute() |
| |
| request = self.opener.request |
| |
| self.assertEqual(expected_headers, request.headers) |
| |
| def test_empty_content_length(self): |
| self.setResponse(200, '\x00\x00\x00\x01\x00\x11ReplaceGatewayUrl\x01' |
| '\x00\x00\x00\x00\x02\x00\x10http://spam.eggs\x00\x00\x00\x00', { |
| 'Content-Type': 'application/x-amf', |
| 'Content-Length': '' |
| }) |
| |
| response = self.gw._getResponse(None) |
| self.assertTrue(response) |
| |
| def test_bad_content_length(self): |
| # test a really borked content-length header |
| self.setResponse(200, self.canned_response, { |
| 'Content-Type': 'application/x-amf', |
| 'Content-Length': 'asdfasdf' |
| }) |
| |
| self.assertRaises(ValueError, self.gw._getResponse, None) |
| |
| |
| class GZipTestCase(BaseServiceTestCase): |
| """ |
| Tests for gzipping responses |
| """ |
| |
| def setUp(self): |
| import gzip |
| |
| env = remoting.Envelope(pyamf.AMF3) |
| r = remoting.Response(['foo' * 50000] * 200) |
| |
| env['/1'] = r |
| |
| response = remoting.encode(env).getvalue() |
| |
| buf = util.BufferedByteStream() |
| x = gzip.GzipFile(fileobj=buf, mode='wb') |
| |
| x.write(response) |
| |
| x.close() |
| |
| self.canned_response = buf.getvalue() |
| |
| BaseServiceTestCase.setUp(self) |
| |
| self.headers['Content-Encoding'] = 'gzip' |
| |
| def test_good_response(self): |
| self.gw._getResponse(None) |
| |
| def test_bad_response(self): |
| self.headers['Content-Length'] = len('foobar') |
| self.setResponse(200, 'foobar', self.headers) |
| |
| self.assertRaises(IOError, self.gw._getResponse, None) |
