Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
G
gevent
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
Analytics
Analytics
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Commits
Issue Boards
Open sidebar
Kirill Smelkov
gevent
Commits
a1d02f2a
Commit
a1d02f2a
authored
Nov 24, 2010
by
Denis Bilenko
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
slight improvements to test__pywsgi.py
parent
47915375
Changes
1
Hide whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
23 additions
and
17 deletions
+23
-17
greentest/test__pywsgi.py
greentest/test__pywsgi.py
+23
-17
No files found.
greentest/test__pywsgi.py
View file @
a1d02f2a
...
@@ -96,11 +96,11 @@ def iread_chunks(fd):
...
@@ -96,11 +96,11 @@ def iread_chunks(fd):
class
Response
(
object
):
class
Response
(
object
):
def
__init__
(
self
,
status_line
,
headers
,
body
=
None
,
chunks
=
None
):
def
__init__
(
self
,
status_line
,
headers
):
self
.
status_line
=
status_line
self
.
status_line
=
status_line
self
.
headers
=
headers
self
.
headers
=
headers
self
.
body
=
body
self
.
body
=
None
self
.
chunks
=
chunks
self
.
chunks
=
False
try
:
try
:
version
,
code
,
self
.
reason
=
status_line
[:
-
2
].
split
(
' '
,
2
)
version
,
code
,
self
.
reason
=
status_line
[:
-
2
].
split
(
' '
,
2
)
except
Exception
:
except
Exception
:
...
@@ -133,16 +133,15 @@ class Response(object):
...
@@ -133,16 +133,15 @@ class Response(object):
assert
self
.
version
==
version
,
'Unexpected version: %r (expected %r)
\
n
%s'
%
(
self
.
version
,
version
,
self
)
assert
self
.
version
==
version
,
'Unexpected version: %r (expected %r)
\
n
%s'
%
(
self
.
version
,
version
,
self
)
def
assertHeader
(
self
,
header
,
value
):
def
assertHeader
(
self
,
header
,
value
):
real_value
=
self
.
headers
.
get
(
header
)
real_value
=
self
.
headers
.
get
(
header
,
False
)
assert
real_value
==
value
,
\
assert
real_value
==
value
,
\
'Unexpected header %r: %r (expected %r)
\
n
%s'
%
(
header
,
real_value
,
value
,
self
)
'Unexpected header %r: %r (expected %r)
\
n
%s'
%
(
header
,
real_value
,
value
,
self
)
def
assertBody
(
self
,
body
):
def
assertBody
(
self
,
body
):
assert
self
.
body
==
body
,
\
assert
self
.
body
==
body
,
'Unexpected body: %r (expected %r)
\
n
%s'
%
(
self
.
body
,
body
,
self
)
'Unexpected body: %r (expected %r)
\
n
%s'
%
(
self
.
body
,
body
,
self
)
@
classmethod
@
classmethod
def
read
(
cls
,
fd
,
code
=
200
,
reason
=
'default'
,
version
=
'1.1'
,
body
=
None
):
def
read
(
cls
,
fd
,
code
=
200
,
reason
=
'default'
,
version
=
'1.1'
,
body
=
None
,
chunks
=
None
,
content_length
=
None
):
_status_line
,
headers
=
read_headers
(
fd
)
_status_line
,
headers
=
read_headers
(
fd
)
self
=
cls
(
_status_line
,
headers
)
self
=
cls
(
_status_line
,
headers
)
if
code
is
not
None
:
if
code
is
not
None
:
...
@@ -155,6 +154,10 @@ class Response(object):
...
@@ -155,6 +154,10 @@ class Response(object):
self
.
assertVersion
(
version
)
self
.
assertVersion
(
version
)
if
self
.
code
==
100
:
if
self
.
code
==
100
:
return
self
return
self
if
content_length
is
not
None
:
if
isinstance
(
content_length
,
int
):
content_length
=
str
(
content_length
)
self
.
assertHeader
(
'Content-Length'
,
content_length
)
try
:
try
:
if
'chunked'
in
headers
.
get
(
'Transfer-Encoding'
,
''
):
if
'chunked'
in
headers
.
get
(
'Transfer-Encoding'
,
''
):
if
CONTENT_LENGTH
in
headers
:
if
CONTENT_LENGTH
in
headers
:
...
@@ -171,6 +174,8 @@ class Response(object):
...
@@ -171,6 +174,8 @@ class Response(object):
raise
raise
if
body
is
not
None
:
if
body
is
not
None
:
self
.
assertBody
(
body
)
self
.
assertBody
(
body
)
if
chunks
is
not
None
:
assert
chunks
==
self
.
chunks
,
(
chunks
,
self
.
chunks
)
return
self
return
self
read_http
=
Response
.
read
read_http
=
Response
.
read
...
@@ -326,7 +331,7 @@ class TestNoChunks(CommonTests):
...
@@ -326,7 +331,7 @@ class TestNoChunks(CommonTests):
fd
=
self
.
makefile
()
fd
=
self
.
makefile
()
fd
.
write
(
'GET / HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
\
r
\
n
'
)
fd
.
write
(
'GET / HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
\
r
\
n
'
)
response
=
read_http
(
fd
,
body
=
'hello world'
)
response
=
read_http
(
fd
,
body
=
'hello world'
)
assert
response
.
chunks
is
Non
e
,
response
.
chunks
assert
response
.
chunks
is
Fals
e
,
response
.
chunks
response
.
assertHeader
(
'Content-Length'
,
'11'
)
response
.
assertHeader
(
'Content-Length'
,
'11'
)
if
not
server_implements_pipeline
:
if
not
server_implements_pipeline
:
...
@@ -334,7 +339,7 @@ class TestNoChunks(CommonTests):
...
@@ -334,7 +339,7 @@ class TestNoChunks(CommonTests):
fd
.
write
(
'GET /not-found HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
\
r
\
n
'
)
fd
.
write
(
'GET /not-found HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
\
r
\
n
'
)
response
=
read_http
(
fd
,
code
=
404
,
reason
=
'Not Found'
,
body
=
'not found'
)
response
=
read_http
(
fd
,
code
=
404
,
reason
=
'Not Found'
,
body
=
'not found'
)
assert
response
.
chunks
is
Non
e
,
response
.
chunks
assert
response
.
chunks
is
Fals
e
,
response
.
chunks
response
.
assertHeader
(
'Content-Length'
,
'9'
)
response
.
assertHeader
(
'Content-Length'
,
'9'
)
...
@@ -406,12 +411,12 @@ class TestChunkedApp(TestCase):
...
@@ -406,12 +411,12 @@ class TestChunkedApp(TestCase):
def
test_chunked_response
(
self
):
def
test_chunked_response
(
self
):
fd
=
self
.
makefile
()
fd
=
self
.
makefile
()
fd
.
write
(
'GET / HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
fd
.
write
(
'GET / HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
response
=
read_http
(
fd
,
body
=
self
.
body
())
response
=
read_http
(
fd
,
body
=
self
.
body
()
,
chunks
=
None
)
if
server_implements_chunked
:
if
server_implements_chunked
:
response
.
assertHeader
(
'Transfer-Encoding'
,
'chunked'
)
response
.
assertHeader
(
'Transfer-Encoding'
,
'chunked'
)
self
.
assertEqual
(
response
.
chunks
,
self
.
chunks
)
self
.
assertEqual
(
response
.
chunks
,
self
.
chunks
)
else
:
else
:
response
.
assertHeader
(
'Transfer-Encoding'
,
Non
e
)
response
.
assertHeader
(
'Transfer-Encoding'
,
Fals
e
)
response
.
assertHeader
(
'Content-Length'
,
str
(
len
(
self
.
body
())))
response
.
assertHeader
(
'Content-Length'
,
str
(
len
(
self
.
body
())))
self
.
assertEqual
(
response
.
chunks
,
None
)
self
.
assertEqual
(
response
.
chunks
,
None
)
...
@@ -490,14 +495,14 @@ class TestUseWrite(TestCase):
...
@@ -490,14 +495,14 @@ class TestUseWrite(TestCase):
fd
.
write
(
'GET /explicit-content-length HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
fd
.
write
(
'GET /explicit-content-length HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
response
=
read_http
(
fd
,
body
=
self
.
body
+
self
.
end
)
response
=
read_http
(
fd
,
body
=
self
.
body
+
self
.
end
)
response
.
assertHeader
(
'Content-Length'
,
self
.
content_length
)
response
.
assertHeader
(
'Content-Length'
,
self
.
content_length
)
response
.
assertHeader
(
'Transfer-Encoding'
,
Non
e
)
response
.
assertHeader
(
'Transfer-Encoding'
,
Fals
e
)
def
test_no_content_length
(
self
):
def
test_no_content_length
(
self
):
fd
=
self
.
makefile
()
fd
=
self
.
makefile
()
fd
.
write
(
'GET /no-content-length HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
fd
.
write
(
'GET /no-content-length HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
response
=
read_http
(
fd
,
body
=
self
.
body
+
self
.
end
)
response
=
read_http
(
fd
,
body
=
self
.
body
+
self
.
end
)
if
server_implements_chunked
:
if
server_implements_chunked
:
response
.
assertHeader
(
'Content-Length'
,
Non
e
)
response
.
assertHeader
(
'Content-Length'
,
Fals
e
)
response
.
assertHeader
(
'Transfer-Encoding'
,
'chunked'
)
response
.
assertHeader
(
'Transfer-Encoding'
,
'chunked'
)
else
:
else
:
response
.
assertHeader
(
'Content-Length'
,
self
.
content_length
)
response
.
assertHeader
(
'Content-Length'
,
self
.
content_length
)
...
@@ -507,7 +512,7 @@ class TestUseWrite(TestCase):
...
@@ -507,7 +512,7 @@ class TestUseWrite(TestCase):
fd
.
write
(
'GET /no-content-length-twice HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
fd
.
write
(
'GET /no-content-length-twice HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
response
=
read_http
(
fd
,
body
=
self
.
body
+
self
.
body
+
self
.
end
)
response
=
read_http
(
fd
,
body
=
self
.
body
+
self
.
body
+
self
.
end
)
if
server_implements_chunked
:
if
server_implements_chunked
:
response
.
assertHeader
(
'Content-Length'
,
Non
e
)
response
.
assertHeader
(
'Content-Length'
,
Fals
e
)
response
.
assertHeader
(
'Transfer-Encoding'
,
'chunked'
)
response
.
assertHeader
(
'Transfer-Encoding'
,
'chunked'
)
assert
response
.
chunks
==
[
self
.
body
,
self
.
body
,
self
.
end
],
response
.
chunks
assert
response
.
chunks
==
[
self
.
body
,
self
.
body
,
self
.
end
],
response
.
chunks
else
:
else
:
...
@@ -557,6 +562,7 @@ class TestHttps(HttpsTestCase):
...
@@ -557,6 +562,7 @@ class TestHttps(HttpsTestCase):
class
TestInternational
(
TestCase
):
class
TestInternational
(
TestCase
):
validator
=
None
# wsgiref.validate.IteratorWrapper([]) does not have __len__
def
application
(
self
,
environ
,
start_response
):
def
application
(
self
,
environ
,
start_response
):
assert
environ
[
'PATH_INFO'
]
==
'/
\
xd0
\
xbf
\
xd1
\
x80
\
xd0
\
xb8
\
xd0
\
xb2
\
xd0
\
xb5
\
xd1
\
x82
'
,
environ
[
'PATH_INFO'
]
assert
environ
[
'PATH_INFO'
]
==
'/
\
xd0
\
xbf
\
xd1
\
x80
\
xd0
\
xb8
\
xd0
\
xb2
\
xd0
\
xb5
\
xd1
\
x82
'
,
environ
[
'PATH_INFO'
]
...
@@ -567,7 +573,7 @@ class TestInternational(TestCase):
...
@@ -567,7 +573,7 @@ class TestInternational(TestCase):
def
test
(
self
):
def
test
(
self
):
sock
=
self
.
connect
()
sock
=
self
.
connect
()
sock
.
sendall
(
'GET /%D0%BF%D1%80%D0%B8%D0%B2%D0%B5%D1%82?%D0%B2%D0%BE%D0%BF%D1%80%D0%BE%D1%81=%D0%BE%D1%82%D0%B2%D0%B5%D1%82 HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
sock
.
sendall
(
'GET /%D0%BF%D1%80%D0%B8%D0%B2%D0%B5%D1%82?%D0%B2%D0%BE%D0%BF%D1%80%D0%BE%D1%81=%D0%BE%D1%82%D0%B2%D0%B5%D1%82 HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
read_http
(
sock
.
makefile
(),
reason
=
'PASSED'
)
read_http
(
sock
.
makefile
(),
reason
=
'PASSED'
,
chunks
=
False
,
body
=
''
,
content_length
=
0
)
class
TestInputReadline
(
TestCase
):
class
TestInputReadline
(
TestCase
):
...
@@ -649,14 +655,14 @@ class TestEmptyYield(TestCase):
...
@@ -649,14 +655,14 @@ class TestEmptyYield(TestCase):
def
test_err
(
self
):
def
test_err
(
self
):
fd
=
self
.
connect
().
makefile
(
bufsize
=
1
)
fd
=
self
.
connect
().
makefile
(
bufsize
=
1
)
fd
.
write
(
'GET / HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
fd
.
write
(
'GET / HTTP/1.1
\
r
\
n
Host: localhost
\
r
\
n
Connection: close
\
r
\
n
\
r
\
n
'
)
st
,
h
,
body
=
read_http
(
fd
)
read_http
(
fd
,
body
=
''
,
chunks
=
[])
assert
body
==
""
garbage
=
fd
.
read
()
garbage
=
fd
.
read
()
self
.
assert_
(
garbage
==
""
,
"got garbage: %r"
%
garbage
)
self
.
assert_
(
garbage
==
""
,
"got garbage: %r"
%
garbage
)
class
TestEmptyWrite
(
TestEmptyYield
):
class
TestEmptyWrite
(
TestEmptyYield
):
@
staticmethod
@
staticmethod
def
application
(
env
,
start_response
):
def
application
(
env
,
start_response
):
write
=
start_response
(
'200 OK'
,
[(
'Content-Type'
,
'text/plain'
)])
write
=
start_response
(
'200 OK'
,
[(
'Content-Type'
,
'text/plain'
)])
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment