Implement and unit test not-after-connect option.

This commit is contained in:
Aldo Cortesi 2013-01-05 20:36:06 +13:00
parent 5493925ea0
commit ed7ef84557
3 changed files with 22 additions and 10 deletions

View File

@ -70,15 +70,16 @@ class PathodHandler(tcp.BaseHandler):
'\r\n' '\r\n'
) )
self.wfile.flush() self.wfile.flush()
try: if not self.server.ssloptions.not_after_connect:
self.convert_to_ssl( try:
self.server.ssloptions.certfile, self.convert_to_ssl(
self.server.ssloptions.keyfile, self.server.ssloptions.certfile,
) self.server.ssloptions.keyfile,
except tcp.NetLibError, v: )
s = str(v) except tcp.NetLibError, v:
self.info(s) s = str(v)
return False, dict(type = "error", msg = s) self.info(s)
return False, dict(type = "error", msg = s)
return True, None return True, None
elif m(http.parse_init_proxy(line)): elif m(http.parse_init_proxy(line)):
method, _, _, _, path, httpversion = m.v method, _, _, _, path, httpversion = m.v

View File

@ -52,6 +52,14 @@ class TestNoApi(tutils.DaemonTests):
assert not "Log" in r.content assert not "Log" in r.content
class TestNotAfterConnect(tutils.DaemonTests):
ssl = False
not_after_connect = True
def test_connect(self):
v = self.pathoc(r"get:'http://foo.com/p/202':da", connect_to=("localhost", self.d.port))
assert v[1] == 202
class TestNohang(tutils.DaemonTests): class TestNohang(tutils.DaemonTests):
nohang = True nohang = True
def test_nohang(self): def test_nohang(self):

View File

@ -1,6 +1,6 @@
import tempfile, os, shutil import tempfile, os, shutil
from contextlib import contextmanager from contextlib import contextmanager
from libpathod import utils, test, pathoc from libpathod import utils, test, pathoc, pathod
import requests import requests
class DaemonTests: class DaemonTests:
@ -10,12 +10,15 @@ class DaemonTests:
ssl = False ssl = False
timeout = None timeout = None
hexdump = False hexdump = False
not_after_connect = False
@classmethod @classmethod
def setUpAll(self): def setUpAll(self):
so = pathod.SSLOptions(not_after_connect = self.not_after_connect)
self.d = test.Daemon( self.d = test.Daemon(
staticdir=test_data.path("data"), staticdir=test_data.path("data"),
anchors=[("/anchor/.*", "202:da")], anchors=[("/anchor/.*", "202:da")],
ssl = self.ssl, ssl = self.ssl,
ssloptions = so,
sizelimit=1*1024*1024, sizelimit=1*1024*1024,
noweb = self.noweb, noweb = self.noweb,
noapi = self.noapi, noapi = self.noapi,