Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
348 lines (279 sloc) 10.801 kB
import webtest
import time
import threading
import web
import urllib
data = """
import web
urls = ("/", "%(classname)s")
app = web.application(urls, globals(), autoreload=True)
class %(classname)s:
def GET(self):
return "%(output)s"
"""
urls = (
"/iter", "do_iter",
)
app = web.application(urls, globals())
class do_iter:
def GET(self):
yield 'hello, '
yield web.input(name='world').name
POST = GET
def write(filename, data):
f = open(filename, 'w')
f.write(data)
f.close()
class ApplicationTest(webtest.TestCase):
def test_reloader(self):
write('foo.py', data % dict(classname='a', output='a'))
import foo
app = foo.app
self.assertEquals(app.request('/').data, 'a')
# test class change
time.sleep(1)
write('foo.py', data % dict(classname='a', output='b'))
self.assertEquals(app.request('/').data, 'b')
# test urls change
time.sleep(1)
write('foo.py', data % dict(classname='c', output='c'))
self.assertEquals(app.request('/').data, 'c')
def testUppercaseMethods(self):
urls = ("/", "hello")
app = web.application(urls, locals())
class hello:
def GET(self): return "hello"
def internal(self): return "secret"
response = app.request('/', method='internal')
self.assertEquals(response.status, '405 Method Not Allowed')
def testRedirect(self):
urls = (
"/a", "redirect /hello/",
"/b/(.*)", r"redirect /hello/\1",
"/hello/(.*)", "hello"
)
app = web.application(urls, locals())
class hello:
def GET(self, name):
name = name or 'world'
return "hello " + name
response = app.request('/a')
self.assertEquals(response.status, '301 Moved Permanently')
self.assertEquals(response.headers['Location'], 'http://0.0.0.0:8080/hello/')
response = app.request('/a?x=2')
self.assertEquals(response.status, '301 Moved Permanently')
self.assertEquals(response.headers['Location'], 'http://0.0.0.0:8080/hello/?x=2')
response = app.request('/b/foo?x=2')
self.assertEquals(response.status, '301 Moved Permanently')
self.assertEquals(response.headers['Location'], 'http://0.0.0.0:8080/hello/foo?x=2')
def test_subdirs(self):
urls = (
"/(.*)", "blog"
)
class blog:
def GET(self, path):
return "blog " + path
app_blog = web.application(urls, locals())
urls = (
"/blog", app_blog,
"/(.*)", "index"
)
class index:
def GET(self, path):
return "hello " + path
app = web.application(urls, locals())
self.assertEquals(app.request('/blog/foo').data, 'blog foo')
self.assertEquals(app.request('/foo').data, 'hello foo')
def processor(handler):
return web.ctx.path + ":" + handler()
app.add_processor(processor)
self.assertEquals(app.request('/blog/foo').data, '/blog/foo:blog foo')
def test_subdomains(self):
def create_app(name):
urls = ("/", "index")
class index:
def GET(self):
return name
return web.application(urls, locals())
urls = (
"a.example.com", create_app('a'),
"b.example.com", create_app('b'),
".*.example.com", create_app('*')
)
app = web.subdomain_application(urls, locals())
def test(host, expected_result):
result = app.request('/', host=host)
self.assertEquals(result.data, expected_result)
test('a.example.com', 'a')
test('b.example.com', 'b')
test('c.example.com', '*')
test('d.example.com', '*')
def test_redirect(self):
urls = (
"/(.*)", "blog"
)
class blog:
def GET(self, path):
if path == 'foo':
raise web.seeother('/login', absolute=True)
else:
raise web.seeother('/bar')
app_blog = web.application(urls, locals())
urls = (
"/blog", app_blog,
"/(.*)", "index"
)
class index:
def GET(self, path):
return "hello " + path
app = web.application(urls, locals())
response = app.request('/blog/foo')
self.assertEquals(response.headers['Location'], 'http://0.0.0.0:8080/login')
response = app.request('/blog/foo', env={'SCRIPT_NAME': '/x'})
self.assertEquals(response.headers['Location'], 'http://0.0.0.0:8080/x/login')
response = app.request('/blog/foo2')
self.assertEquals(response.headers['Location'], 'http://0.0.0.0:8080/blog/bar')
response = app.request('/blog/foo2', env={'SCRIPT_NAME': '/x'})
self.assertEquals(response.headers['Location'], 'http://0.0.0.0:8080/x/blog/bar')
def test_processors(self):
urls = (
"/(.*)", "blog"
)
class blog:
def GET(self, path):
return 'blog ' + path
state = web.storage(x=0, y=0)
def f():
state.x += 1
app_blog = web.application(urls, locals())
app_blog.add_processor(web.loadhook(f))
urls = (
"/blog", app_blog,
"/(.*)", "index"
)
class index:
def GET(self, path):
return "hello " + path
app = web.application(urls, locals())
def g():
state.y += 1
app.add_processor(web.loadhook(g))
app.request('/blog/foo')
assert state.x == 1 and state.y == 1, repr(state)
app.request('/foo')
assert state.x == 1 and state.y == 2, repr(state)
def testUnicodeInput(self):
urls = (
"(/.*)", "foo"
)
class foo:
def GET(self, path):
i = web.input(name='')
return repr(i.name)
def POST(self, path):
if path == '/multipart':
i = web.input(file={})
return i.file.value
else:
i = web.input()
return repr(dict(i))
app = web.application(urls, locals())
def f(name):
path = '/?' + urllib.urlencode({"name": name.encode('utf-8')})
self.assertEquals(app.request(path).data, repr(name))
f(u'\u1234')
f(u'foo')
response = app.request('/', method='POST', data=dict(name='foo'))
self.assertEquals(response.data, "{'name': u'foo'}")
data = '--boundary\r\nContent-Disposition: form-data; name="x"\r\nfoo\r\n--boundary\r\nContent-Disposition: form-data; name="file"; filename="a.txt"\r\nContent-Type: text/plain\r\n\r\na\r\n--boundary--\r\n'
headers = {'Content-Type': 'multipart/form-data; boundary=boundary'}
response = app.request('/multipart', method="POST", data=data, headers=headers)
self.assertEquals(response.data, 'a')
def testCustomNotFound(self):
urls_a = ("/", "a")
urls_b = ("/", "b")
app_a = web.application(urls_a, locals())
app_b = web.application(urls_b, locals())
app_a.notfound = lambda: web.HTTPError("404 Not Found", {}, "not found 1")
urls = (
"/a", app_a,
"/b", app_b
)
app = web.application(urls, locals())
def assert_notfound(path, message):
response = app.request(path)
self.assertEquals(response.status.split()[0], "404")
self.assertEquals(response.data, message)
assert_notfound("/a/foo", "not found 1")
assert_notfound("/b/foo", "not found")
app.notfound = lambda: web.HTTPError("404 Not Found", {}, "not found 2")
assert_notfound("/a/foo", "not found 1")
assert_notfound("/b/foo", "not found 2")
def testIter(self):
self.assertEquals(app.request('/iter').data, 'hello, world')
self.assertEquals(app.request('/iter?name=web').data, 'hello, web')
self.assertEquals(app.request('/iter', method='POST').data, 'hello, world')
self.assertEquals(app.request('/iter', method='POST', data='name=web').data, 'hello, web')
def testUnload(self):
x = web.storage(a=0)
urls = (
"/foo", "foo",
"/bar", "bar"
)
class foo:
def GET(self):
return "foo"
class bar:
def GET(self):
raise web.notfound()
app = web.application(urls, locals())
def unload():
x.a += 1
app.add_processor(web.unloadhook(unload))
app.request('/foo')
self.assertEquals(x.a, 1)
app.request('/bar')
self.assertEquals(x.a, 2)
def test_changequery(self):
urls = (
'/', 'index',
)
class index:
def GET(self):
return web.changequery(x=1)
app = web.application(urls, locals())
def f(path):
return app.request(path).data
self.assertEquals(f('/?x=2'), '/?x=1')
self.assertEquals(f('/?y=1&y=2&x=2'), '/?y=1&y=2&x=1')
def test_setcookie(self):
urls = (
'/', 'index',
)
class index:
def GET(self):
web.setcookie("foo", "bar")
return "hello"
app = web.application(urls, locals())
def f(script_name=""):
response = app.request("/", env={"SCRIPT_NAME": script_name})
return response.headers['Set-Cookie']
self.assertEquals(f(''), 'foo=bar; Path=/')
self.assertEquals(f('/admin'), 'foo=bar; Path=/admin/')
def test_stopsimpleserver(self):
urls = (
'/', 'index',
)
class index:
def GET(self):
pass
app = web.application(urls, locals())
thread = threading.Thread(target=app.run)
thread.start()
time.sleep(1)
self.assertTrue(thread.isAlive())
app.stop()
thread.join(timeout=1)
self.assertFalse(thread.isAlive())
if __name__ == '__main__':
webtest.main()
Jump to Line
Something went wrong with that request. Please try again.