Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Make XMLRPC testing infrastructure reusable:
Move it from `.xmlrpc.tests` to `.xmlrpc.testing`. It now requires the WSGI app to be provided. Use the `testing` extra from `setup.py` to make use of this testing infrastructure.
- Loading branch information
Michael Howitz
committed
Dec 4, 2019
1 parent
5125d8c
commit c2d9163
Showing
6 changed files
with
142 additions
and
115 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| try: | ||
| import xmlrpclib | ||
| except ImportError: | ||
| import xmlrpc.client as xmlrpclib | ||
|
|
||
| try: | ||
| import httplib | ||
| except ImportError: | ||
| import http.client as httplib | ||
|
|
||
| from io import BytesIO | ||
|
|
||
| from zope.app.publisher.testing import AppPublisherLayer | ||
| from zope.app.wsgi.testlayer import http as _http | ||
|
|
||
|
|
||
| class FakeSocket(object): | ||
|
|
||
| def __init__(self, data): | ||
| self.data = data | ||
|
|
||
| def makefile(self, mode, bufsize=None): | ||
| assert 'b' in mode | ||
| data = self.data | ||
| if not isinstance(data, bytes): | ||
| data = data.encode('iso-8859-1') | ||
| return BytesIO(data) | ||
|
|
||
|
|
||
| def http(wsgi_app, query_str, *args, **kwargs): | ||
| wsgi_app = AppPublisherLayer.make_wsgi_app() | ||
| # Strip leading \n | ||
| query_str = query_str.lstrip() | ||
| kwargs.setdefault('handle_errors', True) | ||
| if not isinstance(query_str, bytes): | ||
| query_str = query_str.encode("utf-8") | ||
| return _http(wsgi_app, query_str, *args, **kwargs) | ||
|
|
||
|
|
||
| class ZopeTestTransport(xmlrpclib.Transport): | ||
| """xmlrpclib transport that delegates to | ||
| zope.app.wsgi.testlayer.http | ||
| It can be used like a normal transport, including support for basic | ||
| authentication. | ||
| """ | ||
|
|
||
| verbose = False | ||
| handleErrors = True | ||
|
|
||
| def __init__(self, wsgi_app): | ||
| super(ZopeTestTransport, self).__init__() | ||
| self.wsgi_app = wsgi_app | ||
|
|
||
| def request(self, host, handler, request_body, verbose=0): | ||
| request = "POST %s HTTP/1.0\n" % (handler,) | ||
| request += "Content-Length: %i\n" % len(request_body) | ||
| request += "Content-Type: text/xml\n" | ||
|
|
||
| host, extra_headers, _x509 = self.get_host_info(host) | ||
| if extra_headers: | ||
| request += "Authorization: %s\n" % ( | ||
| dict(extra_headers)["Authorization"],) | ||
|
|
||
| request += "\n" | ||
| if isinstance(request_body, bytes) and str is not bytes: | ||
| # Python 3 | ||
| request = request.encode("ascii") | ||
| request += request_body | ||
| response = http( | ||
| self.wsgi_app, request, handle_errors=self.handleErrors) | ||
|
|
||
| errcode = response.getStatus() | ||
| errmsg = response.getStatusString() | ||
| assert errcode == 200 | ||
|
|
||
| body = response.getBody() | ||
| if not isinstance(body, str): | ||
| # Python 3 | ||
| body = body.decode("utf-8") | ||
| content = 'HTTP/1.0 ' + errmsg + '\n\n' + body | ||
|
|
||
| res = httplib.HTTPResponse(FakeSocket(content)) | ||
| res.begin() | ||
| return self.parse_response(res) | ||
|
|
||
|
|
||
| def ServerProxy(wsgi_app, uri, transport=None, encoding=None, | ||
| verbose=0, allow_none=0, handleErrors=True): | ||
| """A factory that creates a server proxy using the ZopeTestTransport | ||
| by default. | ||
| """ | ||
| if transport is None: | ||
| transport = ZopeTestTransport(wsgi_app) | ||
| if isinstance(transport, ZopeTestTransport): | ||
| transport.handleErrors = handleErrors | ||
| return xmlrpclib.ServerProxy(uri, transport, encoding, verbose, allow_none) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,92 +1,11 @@ | ||
| try: | ||
| import xmlrpclib | ||
| except ImportError: | ||
| import xmlrpc.client as xmlrpclib | ||
|
|
||
| try: | ||
| import httplib | ||
| except ImportError: | ||
| import http.client as httplib | ||
|
|
||
| from io import BytesIO | ||
|
|
||
| from zope.app.publisher.testing import AppPublisherLayer | ||
| from zope.app.wsgi.testlayer import http as _http | ||
|
|
||
| class FakeSocket(object): | ||
|
|
||
| def __init__(self, data): | ||
| self.data = data | ||
|
|
||
| def makefile(self, mode, bufsize=None): | ||
| assert 'b' in mode | ||
| data = self.data | ||
| if not isinstance(data, bytes): | ||
| data = data.encode('iso-8859-1') | ||
| return BytesIO(data) | ||
|
|
||
| def http(query_str, *args, **kwargs): | ||
| wsgi_app = AppPublisherLayer.make_wsgi_app() | ||
| # Strip leading \n | ||
| query_str = query_str.lstrip() | ||
| kwargs.setdefault('handle_errors', True) | ||
| if not isinstance(query_str, bytes): | ||
| query_str = query_str.encode("utf-8") | ||
| return _http(wsgi_app, query_str, *args, **kwargs) | ||
|
|
||
|
|
||
| class ZopeTestTransport(xmlrpclib.Transport): | ||
| """xmlrpclib transport that delegates to | ||
| zope.app.wsgi.testlayer.http | ||
| It can be used like a normal transport, including support for basic | ||
| authentication. | ||
| """ | ||
|
|
||
| verbose = False | ||
| handleErrors = True | ||
|
|
||
| def request(self, host, handler, request_body, verbose=0): | ||
| request = "POST %s HTTP/1.0\n" % (handler,) | ||
| request += "Content-Length: %i\n" % len(request_body) | ||
| request += "Content-Type: text/xml\n" | ||
|
|
||
| host, extra_headers, _x509 = self.get_host_info(host) | ||
| if extra_headers: | ||
| request += "Authorization: %s\n" % ( | ||
| dict(extra_headers)["Authorization"],) | ||
|
|
||
| request += "\n" | ||
| if isinstance(request_body, bytes) and str is not bytes: | ||
| # Python 3 | ||
| request = request.encode("ascii") | ||
| request += request_body | ||
| response = http(request, handle_errors=self.handleErrors) | ||
|
|
||
| errcode = response.getStatus() | ||
| errmsg = response.getStatusString() | ||
| # This is not the same way that the normal transport deals with the | ||
| # headers. | ||
| headers = response.getHeaders() | ||
|
|
||
| assert errcode == 200 | ||
|
|
||
| body = response.getBody() | ||
| if not isinstance(body, str): | ||
| # Python 3 | ||
| body = body.decode("utf-8") | ||
| content = 'HTTP/1.0 ' + errmsg + '\n\n' + body | ||
|
|
||
| res = httplib.HTTPResponse(FakeSocket(content)) | ||
| res.begin() | ||
| return self.parse_response(res) | ||
|
|
||
| def ServerProxy(uri, transport=None, encoding=None, | ||
| verbose=0, allow_none=0, handleErrors=True): | ||
| """A factory that creates a server proxy using the ZopeTestTransport | ||
| by default. | ||
| """ | ||
| if transport is None: | ||
| transport = ZopeTestTransport() | ||
| if isinstance(transport, ZopeTestTransport): | ||
| transport.handleErrors = handleErrors | ||
| return xmlrpclib.ServerProxy(uri, transport, encoding, verbose, allow_none) | ||
| from zope.deferredimport import deprecated | ||
|
|
||
| deprecated( | ||
| "The contents of zope/app/publisher/xmlrpc/tests/__init__.py have been" | ||
| " moved to zope/app/publisher/xmlrpc/testing.py for reusability." | ||
| " Please import from there.", | ||
| FakeSocket='zope.app.publisher.xmlrpc.testing:FakeSocket', | ||
| http='zope.app.publisher.xmlrpc.testing:http', | ||
| ZopeTestTransport='zope.app.publisher.xmlrpc.testing:ZopeTestTransport', | ||
| ServerProxy='zope.app.publisher.xmlrpc.testing:ServerProxy', | ||
| ) |
Oops, something went wrong.