Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Code duplication from zope.app.wsgi in tests #7

Closed
icemac opened this issue Dec 4, 2019 · 1 comment · Fixed by #8
Closed

Code duplication from zope.app.wsgi in tests #7

icemac opened this issue Dec 4, 2019 · 1 comment · Fixed by #8
Assignees

Comments

@icemac
Copy link
Member

icemac commented Dec 4, 2019

The following code lines are mostly identical to zope.app.wsgi.testlayer:

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)

As zope.app.publisher already depends on zope.app.wsgi. The code there could be used here after zopefoundation/zope.app.wsgi#17 is fixed.

@icemac
Copy link
Member Author

icemac commented Dec 4, 2019

As zope.app.wsgi has no infrastructure to test the XMLRPC server proxy (zopefoundation/zope.app.wsgi#17 is there because the code is completely untested.) #8 suggests to generalize the code here and remove it from zope.app.wsgi.

@icemac icemac self-assigned this Dec 4, 2019
@icemac icemac closed this as completed in #8 Dec 5, 2019
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
Development

Successfully merging a pull request may close this issue.

1 participant