[py-svn] r31751 - in py/branch/distributed/py/test/rsession: . webdata
fijal at codespeak.net
fijal at codespeak.net
Sun Aug 27 20:22:54 CEST 2006
Author: fijal
Date: Sun Aug 27 20:22:51 2006
New Revision: 31751
Modified:
py/branch/distributed/py/test/rsession/web.py
py/branch/distributed/py/test/rsession/webdata/index.html
py/branch/distributed/py/test/rsession/webjs.py
Log:
Added some niclets. Still does not work always like py.test ../../pypy/sth, needs definitely more tests for pylib magic.
Modified: py/branch/distributed/py/test/rsession/web.py
==============================================================================
--- py/branch/distributed/py/test/rsession/web.py (original)
+++ py/branch/distributed/py/test/rsession/web.py Sun Aug 27 20:22:51 2006
@@ -18,7 +18,13 @@
described
from py.__.test.rsession.rsession import RSession
from py.__.test.rsession import report
+from py.__.test import collect
from pypy.translator.js.main import rpython2javascript
+import os
+
+from pypy.translator.js import json
+
+import Queue
DATADIR = "webdata"
@@ -30,42 +36,87 @@
class ExportedMethods(BasicExternal):
_render_xmlhttp = True
def __init__(self):
- self.wait_flag = threading.Condition()
- self.pending_events = []
+ self.pending_events = Queue.Queue()
+ self.start_event = threading.Event()
+
+ def findmodule(self, item):
+ # find the most outwards parent which is module
+ current = item
+ while current:
+ if isinstance(current, collect.Module):
+ print current.name
+ return current.name
+ current = current.parent
+ return "unknown"
+
+ @described(retval=["aa"])
+ def show_hosts(self):
+ self.start_event.wait()
+ return json.write(self.hosts)
@described(retval={"aa":"aa"})
def show_status_change(self):
- while 1:
- try:
- self.wait_flag.acquire()
- self.wait_flag.wait(1)
- if self.pending_events:
- event = self.pending_events.pop()
- # some dispatcher here
- if isinstance(event, report.ReceivedItemOutcome):
- args = ["%s: '%s'" % (key, escape(str(value))) \
- for key, value in event.outcome.__dict__.iteritems()]
- elif isinstance(event, report.ItemStart):
- item = event.item
- item_tuple = "itemtype: '%s', itemname: '%s'" % (\
- item.__class__.__name__, item.name)
- args = [item_tuple]
- else:
- args = []
- args.append("event: '%s'" % escape(str(event)))
-
- type_tuple = "type: '%s'" % event.__class__.__name__
- return "{%s}" % ",".join(args + [type_tuple])
- finally:
- self.wait_flag.release()
-
- def report(self, event):
+ def add_item(event):
+ item = event.item
+ itemtype = item.__class__.__name__
+ itemname = item.name
+ return {'itemtype':itemtype, 'itemname':itemname}
+
+ event = self.pending_events.get()
+ if event is None:
+ os._exit(0)
+ # some dispatcher here
+ if isinstance(event, report.ReceivedItemOutcome):
+ args = {}
+ for key, val in event.outcome.__dict__.iteritems():
+ args[key] = str(val)
+ args.update(add_item(event))
+ args['moduleitem'] = str(self.findmodule(event.item))
+ elif isinstance(event, report.ItemStart):
+ args = add_item(event)
+ elif isinstance(event, report.HostReady):
+ args = {'hostname' : event.hostname}
+ else:
+ args = {}
+ args['event'] = escape(str(event))
+ args['type'] = event.__class__.__name__
+ return json.write(args)
+
+ def report_ReceivedItemOutcome(self, event):
+ self.pending_events.put(event)
+
+ def report_ItemStart(self, event):
+ self.pending_events.put(event)
+
+ def report_unknown(self, event):
+ # XXX: right now, we just pass it for showing
+ self.pending_events.put(event)
+
+ def report_TestStarted(self, event):
+ self.hosts = event.hosts
+ self.start_event.set()
+ self.pending_events.put(event)
+
+ def report(self, what):
+ repfun = getattr(self, "report_" + what.__class__.__name__,
+ self.report_unknown)
try:
- self.wait_flag.acquire()
- self.pending_events.insert(0, event)
- self.wait_flag.notify()
- finally:
- self.wait_flag.release()
+ repfun(what)
+ except (KeyboardInterrupt, SystemExit):
+ raise
+ except:
+ print "Internal reporting problem"
+ excinfo = py.code.ExceptionInfo()
+ for i in excinfo.traceback:
+ print str(i)[2:-1]
+ print excinfo
+
+## try:
+## self.wait_flag.acquire()
+## self.pending_events.insert(0, event)
+## self.wait_flag.notify()
+## finally:
+## self.wait_flag.release()
exported_methods = ExportedMethods()
@@ -111,10 +162,13 @@
server_address = ('', 8000)
# do some options parsing
config, args = py.test.Config.parse(args)
- thread.start_new_thread(RSession(config).main, (args, \
- exported_methods.report))
httpd = HTTPServer(server_address, TestHandler)
- httpd.serve_forever()
+ thread.start_new_thread(httpd.serve_forever, ())
+ try:
+ RSession(config).main(args, exported_methods.report)
+ finally:
+ exported_methods.pending_events.put(None)
+ os.wait()
if __name__ == '__main__':
_main(sys.argv)
Modified: py/branch/distributed/py/test/rsession/webdata/index.html
==============================================================================
--- py/branch/distributed/py/test/rsession/webdata/index.html (original)
+++ py/branch/distributed/py/test/rsession/webdata/index.html Sun Aug 27 20:22:51 2006
@@ -5,6 +5,9 @@
</head>
<body onload="main()">
<h3>Tests</h3>
+<table id="hosts">
+ <tr id="hostrow"></tr>
+</table>
<table id="main_table">
</table>
<div id="testmain"></div>
Modified: py/branch/distributed/py/test/rsession/webjs.py
==============================================================================
--- py/branch/distributed/py/test/rsession/webjs.py (original)
+++ py/branch/distributed/py/test/rsession/webjs.py Sun Aug 27 20:22:51 2006
@@ -8,19 +8,54 @@
def create_elem(s):
return dom.get_document().createElement(s)
+def get_elem(el):
+ return dom.get_document().getElementById(el)
+
def comeback(msg):
elem = dom.get_document().getElementById("testmain")
elem.innerHTML += '%s<br/>' % msg['event']
main_t = dom.get_document().getElementById("main_table")
if msg['type'] == 'ItemStart':
# we start a new directory or what
- tr = create_elem("tr")
- td = create_elem("td")
- tr.appendChild(td)
- td.innerHTML = msg['itemtype'] + " " + msg['itemname']
- main_t.appendChild(tr)
+ if msg['itemtype'] == 'Module':
+ tr = create_elem("tr")
+ td = create_elem("td")
+ tr.appendChild(td)
+ td.innerHTML = msg['itemname']
+ # XXX: Hack, which asserts that module is unique, which is obviously
+ # not true, we should perform some cracking, but probably
+ # on different side than
+ tr.id = msg['itemname']
+ main_t.appendChild(tr)
+ elif msg['type'] == 'HostReady':
+ dom.get_document().getElementById(msg['hostname']).style.background = \
+ "#00ff00"
+ elif msg['type'] == 'ReceivedItemOutcome':
+ try:
+ module_part = get_elem(msg['moduleitem'])
+ td = create_elem("td")
+ # TODO: dispatch output
+ if msg["passed"] == 'True':
+ td.innerHTML = "."
+ else:
+ td.innerHTML = 'F'
+ module_part.appendChild(td)
+ except:
+ dom.get_document().getElementById("testmain").innerHTML += "some error"
exported_methods.show_status_change(comeback)
+def host_init(host_list):
+ elem = dom.get_document().getElementById("hostrow")
+ for host in host_list:
+ td = create_elem("td")
+ td.style.background = "#ff0000"
+ td.innerHTML = host
+ td.id = host
+ elem.appendChild(td)
+## elem = dom.get_document().getElementById("testmain")
+## elem.innerHTML = str(host_list.keys())
+
def main():
+ exported_methods.show_hosts(host_init)
exported_methods.show_status_change(comeback)
More information about the pytest-commit
mailing list