jiuyuan
/
test_ok1
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

* remove unused py._thread namespace, rewrite the one usage 

* remove unused py/test/web directory

--HG--
branch : trunk
This commit is contained in:
holger krekel 2009-10-03 19:57:48 +02:00
parent 5791c06bf2
commit 6e11f8cd2a
12 changed files with 8 additions and 577 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
_py/test/web/__init__.py
View File

6
_py/test/web/exception.py
View File

@ -1,6 +0,0 @@
class CSSError(Exception):
"""raised when there's a problem with the CSS"""
class HTMLError(Exception):
"""raised when there's a problem with the HTML"""

58
_py/test/web/post_multipart.py
View File

@ -1,58 +0,0 @@
import httplib, mimetypes
"""Copied from the cookbook
see ActiveState's ASPN
http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/146306
"""
def post_multipart(host, selector, fields, files):
"""
Post fields and files to an http host as multipart/form-data.
fields is a sequence of (name, value) elements for regular form fields.
files is a sequence of (name, filename, value) elements for data to be
uploaded as files
Return the server's response page.
"""
content_type, body = encode_multipart_formdata(fields, files)
h = httplib.HTTP(host)
h.putrequest('POST', selector)
h.putheader('content-type', content_type)
h.putheader('content-length', str(len(body)))
h.endheaders()
h.send(body)
errcode, errmsg, headers = h.getreply()
return h.file.read()
def encode_multipart_formdata(fields, files):
"""
fields is a sequence of (name, value) elements for regular form fields.
files is a sequence of (name, filename, value) elements for data to be
uploaded as files
Return (content_type, body) ready for httplib.HTTP instance
"""
BOUNDARY = '----------ThIs_Is_tHe_bouNdaRY_$'
CRLF = '\r\n'
L = []
for (key, value) in fields:
L.append('--' + BOUNDARY)
L.append('Content-Disposition: form-data; name="%s"' % key)
L.append('')
L.append(value)
for (key, filename, value) in files:
L.append('--' + BOUNDARY)
L.append('Content-Disposition: form-data; name="%s"; filename="%s"' %
(key, filename))
L.append('Content-Type: %s' % get_content_type(filename))
L.append('')
L.append(value)
L.append('--' + BOUNDARY + '--')
L.append('')
body = CRLF.join(L)
content_type = 'multipart/form-data; boundary=%s' % BOUNDARY
return content_type, body
def get_content_type(filename):
return mimetypes.guess_type(filename)[0] or 'application/octet-stream'

41
_py/test/web/webcheck.py
View File

@ -1,41 +0,0 @@
import py
import re
from exception import *
from post_multipart import post_multipart
#import css_checker
def check_html(string):
"""check an HTML string for wellformedness and validity"""
tempdir = py.test.ensuretemp('check_html')
filename = 'temp%s.html' % (hash(string), )
tempfile = tempdir.join(filename)
tempfile.write(string)
ret = post_multipart('validator.w3.org', '/check', [],
[('uploaded_file', 'somehtml.html', string)])
is_valid = get_validation_result_from_w3_html(ret)
return is_valid
reg_validation_result = re.compile(
'<(h2|td)[^>]*class="(in)?valid"[^>]*>([^<]*)<', re.M | re.S)
def get_validation_result_from_w3_html(html):
match = reg_validation_result.search(html)
valid = match.group(1) is None
text = match.group(2).strip()
if not valid:
temp = py.test.ensuretemp('/w3_results_%s.html' % hash(html), dir=0)
temp.write(html)
raise HTMLError(
"The html is not valid. See the report file at '%s'" % temp)
return valid
#def check_css(string, basepath, htmlpath='/'):
# """check the CSS of an HTML string
#
# check whether an HTML string contains CSS rels, and if so check whether
# any classes defined in the HTML actually have a matching CSS selector
# """
# c = css_checker.css_checker(string, basepath, htmlpath)
# # raises a CSSError when failing, this is done from the tester class to
# # allow being more verbose than just 'something went wrong'
# return c.check()

1
_py/thread/__init__.py
View File

@ -1 +0,0 @@
#

83
_py/thread/io.py
View File

@ -1,83 +0,0 @@
try:
from _thread import get_ident
except ImportError:
from thread import get_ident
class ThreadOut(object):
""" A file like object that diverts writing operations
to per-thread writefuncs.
This is a py lib internal class and not meant for outer use
or modification.
"""
def __new__(cls, obj, attrname):
""" Divert file output to per-thread writefuncs.
the given obj and attrname describe the destination
of the file.
"""
current = getattr(obj, attrname)
if isinstance(current, cls):
current._used += 1
return current
self = object.__new__(cls)
self._tid2out = {}
self._used = 1
self._oldout = getattr(obj, attrname)
self._defaultwriter = self._oldout.write
self._address = (obj, attrname)
setattr(obj, attrname, self)
return self
def isatty(self):
# XXX
return False
def setdefaultwriter(self, writefunc):
self._defaultwriter = writefunc
def resetdefault(self):
self._defaultwriter = self._oldout.write
def softspace():
def fget(self):
return self._get()[0]
def fset(self, value):
self._get()[0] = value
return property(fget, fset, None, "software attribute")
softspace = softspace()
def deinstall(self):
self._used -= 1
x = self._used
if x <= 0:
obj, attrname = self._address
setattr(obj, attrname, self._oldout)
def setwritefunc(self, writefunc, tid=None):
if tid is None:
tid = get_ident()
self._tid2out[tid] = [0, writefunc]
def delwritefunc(self, tid=None, ignoremissing=True):
if tid is None:
tid = get_ident()
try:
del self._tid2out[tid]
except KeyError:
if not ignoremissing:
raise
def _get(self):
tid = get_ident()
try:
return self._tid2out[tid]
except KeyError:
return getattr(self._defaultwriter, 'softspace', 0), self._defaultwriter
def write(self, data):
softspace, out = self._get()
out(data)
def flush(self):
pass

208
_py/thread/pool.py
View File

@ -1,208 +0,0 @@
import threading
import time
import sys
import py
queue = py.builtin._tryimport('queue', 'Queue')
ERRORMARKER = object()
class Reply(object):
""" reply instances provide access to the result
of a function execution that got dispatched
through WorkerPool.dispatch()
"""
_excinfo = None
def __init__(self, task):
self.task = task
self._queue = queue.Queue()
def _set(self, result):
self._queue.put(result)
def _setexcinfo(self, excinfo):
self._excinfo = excinfo
self._queue.put(ERRORMARKER)
def _get_with_timeout(self, timeout):
# taken from python2.3's Queue.get()
# we want to run on python2.2 here
delay = 0.0005 # 500 us -> initial delay of 1 ms
endtime = time.time() + timeout
while 1:
try:
return self._queue.get_nowait()
except queue.Empty:
remaining = endtime - time.time()
if remaining <= 0: #time is over and no element arrived
raise IOError("timeout waiting for task %r" %(self.task,))
delay = min(delay * 2, remaining, .05)
time.sleep(delay) #reduce CPU usage by using a sleep
def get(self, timeout=None):
""" get the result object from an asynchronous function execution.
if the function execution raised an exception,
then calling get() will reraise that exception
including its traceback.
"""
if self._queue is None:
raise EOFError("reply has already been delivered")
if timeout is not None:
result = self._get_with_timeout(timeout)
else:
result = self._queue.get()
if result is ERRORMARKER:
self._queue = None
excinfo = self._excinfo
py.builtin._reraise(excinfo[0], excinfo[1], excinfo[2])
return result
class WorkerThread(threading.Thread):
def __init__(self, pool):
threading.Thread.__init__(self)
self._queue = queue.Queue()
self._pool = pool
self.setDaemon(1)
def _run_once(self):
reply = self._queue.get()
if reply is SystemExit:
return False
assert self not in self._pool._ready
task = reply.task
try:
func, args, kwargs = task
result = func(*args, **kwargs)
except (SystemExit, KeyboardInterrupt):
return False
except:
reply._setexcinfo(sys.exc_info())
else:
reply._set(result)
# at this point, reply, task and all other local variables go away
return True
def run(self):
try:
while self._run_once():
self._pool._ready[self] = True
finally:
del self._pool._alive[self]
try:
del self._pool._ready[self]
except KeyError:
pass
def send(self, task):
reply = Reply(task)
self._queue.put(reply)
return reply
def stop(self):
self._queue.put(SystemExit)
class WorkerPool(object):
""" A WorkerPool allows to dispatch function executions
to threads. Each Worker Thread is reused for multiple
function executions. The dispatching operation
takes care to create and dispatch to existing
threads.
You need to call shutdown() to signal
the WorkerThreads to terminate and join()
in order to wait until all worker threads
have terminated.
"""
_shuttingdown = False
def __init__(self, maxthreads=None):
""" init WorkerPool instance which may
create up to `maxthreads` worker threads.
"""
self.maxthreads = maxthreads
self._ready = {}
self._alive = {}
def dispatch(self, func, *args, **kwargs):
""" return Reply object for the asynchronous dispatch
of the given func(*args, **kwargs) in a
separate worker thread.
"""
if self._shuttingdown:
raise IOError("WorkerPool is already shutting down")
try:
thread, _ = self._ready.popitem()
except KeyError: # pop from empty list
if self.maxthreads and len(self._alive) >= self.maxthreads:
raise IOError("can't create more than %d threads." %
(self.maxthreads,))
thread = self._newthread()
return thread.send((func, args, kwargs))
def _newthread(self):
thread = WorkerThread(self)
self._alive[thread] = True
thread.start()
return thread
def shutdown(self):
""" signal all worker threads to terminate.
call join() to wait until all threads termination.
"""
if not self._shuttingdown:
self._shuttingdown = True
for t in list(self._alive):
t.stop()
def join(self, timeout=None):
""" wait until all worker threads have terminated. """
current = threading.currentThread()
deadline = delta = None
if timeout is not None:
deadline = time.time() + timeout
for thread in list(self._alive):
if deadline:
delta = deadline - time.time()
if delta <= 0:
raise IOError("timeout while joining threads")
thread.join(timeout=delta)
if thread.isAlive():
raise IOError("timeout while joining threads")
class NamedThreadPool:
def __init__(self, **kw):
self._namedthreads = {}
for name, value in kw.items():
self.start(name, value)
def __repr__(self):
return "<NamedThreadPool %r>" %(self._namedthreads)
def get(self, name=None):
if name is None:
l = []
for x in self._namedthreads.values():
l.extend(x)
return l
else:
return self._namedthreads.get(name, [])
def getstarted(self, name=None):
return [t for t in self.get(name) if t.isAlive()]
def prunestopped(self, name=None):
if name is None:
for name in self.names():
self.prunestopped(name)
else:
self._namedthreads[name] = self.getstarted(name)
def names(self):
return self._namedthreads.keys()
def start(self, name, func):
l = self._namedthreads.setdefault(name, [])
thread = threading.Thread(name="%s%d" % (name, len(l)),
target=func)
thread.start()
l.append(thread)

8
setup.py
View File

@ -65,9 +65,7 @@ def main():
'_py.test',
'_py.test.dist',
'_py.test.looponfail',
'_py.test.plugin',
'_py.test.web',
'_py.thread'],
'_py.test.plugin',],
package_data={'py': ['bin/_findpy.py',
'bin/env.cmd',
'bin/env.py',
@ -86,8 +84,8 @@ def main():
'bin/win32/py.rest.cmd',
'bin/win32/py.svnwcrevert.cmd',
'bin/win32/py.test.cmd',
'bin/win32/py.which.cmd',
'rest/rest.sty.template']},
'bin/win32/py.which.cmd',],
'_py': ['rest/rest.sty.template']},
zip_safe=True,
)

13
testing/pytest/looponfail/test_util.py
View File

@ -51,14 +51,11 @@ def test_pycremoval(tmpdir):
assert not pycfile.check()
def test_waitonchange(tmpdir):
def test_waitonchange(tmpdir, monkeypatch):
tmp = tmpdir
sd = StatRecorder([tmp])
wp = py._thread.WorkerPool(1)
reply = wp.dispatch(sd.waitonchange, checkinterval=0.2)
py.std.time.sleep(0.05)
tmp.ensure("newfile.py")
reply.get(timeout=0.5)
wp.shutdown()
l = [True, False]
monkeypatch.setattr(StatRecorder, 'check', lambda self: l.pop())
sd.waitonchange(checkinterval=0.2)
assert not l

1
testing/thread/__init__.py
View File

@ -1 +0,0 @@
#

72
testing/thread/test_io.py
View File

@ -1,72 +0,0 @@
import py
import sys
WorkerPool = py._thread.WorkerPool
ThreadOut = py._thread.ThreadOut
def test_threadout_install_deinstall():
old = sys.stdout
out = ThreadOut(sys, 'stdout')
out.deinstall()
assert old == sys.stdout
class TestThreadOut:
def test_threadout_one(self):
out = ThreadOut(sys, 'stdout')
try:
l = []
out.setwritefunc(l.append)
py.builtin.print_(42,13)
x = l.pop(0)
assert x == '42'
x = l.pop(0)
assert x == ' '
x = l.pop(0)
assert x == '13'
finally:
out.deinstall()
def test_threadout_multi_and_default(self):
out = ThreadOut(sys, 'stdout')
try:
num = 3
defaults = []
def f(l):
out.setwritefunc(l.append)
sys.stdout.write(str(id(l)))
out.delwritefunc()
print(1)
out.setdefaultwriter(defaults.append)
pool = WorkerPool()
listlist = []
for x in range(num):
l = []
listlist.append(l)
pool.dispatch(f, l)
pool.shutdown()
for name, value in out.__dict__.items():
sys.stderr.write("%s: %s" %(name, value))
pool.join(2.0)
for i in range(num):
item = listlist[i]
assert item ==[str(id(item))]
assert not out._tid2out
assert defaults
expect = ['1' for x in range(num)]
defaults = [x for x in defaults if x.strip()]
assert defaults == expect
finally:
out.deinstall()
def test_threadout_nested(self):
out1 = ThreadOut(sys, 'stdout')
try:
# we want ThreadOuts to coexist
last = sys.stdout
out = ThreadOut(sys, 'stdout')
assert last == sys.stdout
out.deinstall()
assert last == sys.stdout
finally:
out1.deinstall()

94
testing/thread/test_pool.py
View File

@ -1,94 +0,0 @@
import py
import sys
from _py.thread.pool import queue
WorkerPool = py._thread.WorkerPool
ThreadOut = py._thread.ThreadOut
def test_some():
pool = WorkerPool()
q = queue.Queue()
num = 4
def f(i):
q.put(i)
while q.qsize():
py.std.time.sleep(0.01)
for i in range(num):
pool.dispatch(f, i)
for i in range(num):
q.get()
assert len(pool._alive) == 4
pool.shutdown()
# XXX I replaced the following join() with a time.sleep(1), which seems
# to fix the test on Windows, and doesn't break it on Linux... Completely
# unsure what the idea is, though, so it would be nice if someone with some
# more understanding of what happens here would either fix this better, or
# remove this comment...
# pool.join(timeout=1.0)
py.std.time.sleep(1)
assert len(pool._alive) == 0
assert len(pool._ready) == 0
def test_get():
pool = WorkerPool()
def f():
return 42
reply = pool.dispatch(f)
result = reply.get()
assert result == 42
def test_get_timeout():
pool = WorkerPool()
def f():
py.std.time.sleep(0.2)
return 42
reply = pool.dispatch(f)
py.test.raises(IOError, "reply.get(timeout=0.01)")
def test_get_excinfo():
pool = WorkerPool()
def f():
raise ValueError("42")
reply = pool.dispatch(f)
excinfo = py.test.raises(ValueError, "reply.get(1.0)")
py.test.raises(EOFError, "reply.get(1.0)")
def test_maxthreads():
pool = WorkerPool(maxthreads=1)
def f():
py.std.time.sleep(0.5)
try:
pool.dispatch(f)
py.test.raises(IOError, pool.dispatch, f)
finally:
pool.shutdown()
def test_join_timeout():
pool = WorkerPool()
q = queue.Queue()
def f():
q.get()
reply = pool.dispatch(f)
pool.shutdown()
py.test.raises(IOError, pool.join, 0.01)
q.put(None)
reply.get(timeout=1.0)
pool.join(timeout=0.1)
def test_pool_clean_shutdown():
capture = py.io.StdCaptureFD()
pool = WorkerPool()
def f():
pass
pool.dispatch(f)
pool.dispatch(f)
pool.shutdown()
pool.join(timeout=1.0)
assert not pool._alive
assert not pool._ready
out, err = capture.reset()
print(out)
sys.stderr.write(err + "\n")
assert err == ''