[py-svn] r48136 - in py/branch/reporter-merge/py/test: . rsession rsession/testing testing
fijal at codespeak.net
fijal at codespeak.net
Sun Oct 28 19:16:48 CET 2007
Author: fijal
Date: Sun Oct 28 19:16:47 2007
New Revision: 48136
Added:
py/branch/reporter-merge/py/test/box.py
- copied unchanged from r47660, py/branch/reporter-merge/py/test/rsession/box.py
py/branch/reporter-merge/py/test/executor.py
- copied unchanged from r47678, py/branch/reporter-merge/py/test/rsession/executor.py
py/branch/reporter-merge/py/test/test_executor.py
- copied, changed from r47660, py/branch/reporter-merge/py/test/rsession/testing/test_executor.py
py/branch/reporter-merge/py/test/testing/test_boxing.py
- copied unchanged from r47660, py/branch/reporter-merge/py/test/rsession/testing/test_boxing.py
Removed:
py/branch/reporter-merge/py/test/rsession/box.py
py/branch/reporter-merge/py/test/rsession/executor.py
py/branch/reporter-merge/py/test/rsession/testing/test_boxing.py
py/branch/reporter-merge/py/test/rsession/testing/test_executor.py
Modified:
py/branch/reporter-merge/py/test/rsession/rsession.py
Log:
Intermediate checkin to start moving things
Deleted: /py/branch/reporter-merge/py/test/rsession/box.py
==============================================================================
--- /py/branch/reporter-merge/py/test/rsession/box.py Sun Oct 28 19:16:47 2007
+++ (empty file)
@@ -1,120 +0,0 @@
-
-""" boxing - wrapping process with another process so we can run
-a process inside and see if it crashes
-"""
-
-import py
-import os
-import sys
-import marshal
-from py.__.test import config as pytestconfig
-
-PYTESTSTDOUT = "pyteststdout"
-PYTESTSTDERR = "pyteststderr"
-PYTESTRETVAL = "pytestretval"
-
-import tempfile
-import itertools
-from StringIO import StringIO
-
-counter = itertools.count().next
-
-class FileBox(object):
- def __init__(self, fun, args=None, kwargs=None, config=None):
- if args is None:
- args = []
- if kwargs is None:
- kwargs = {}
- self.fun = fun
- self.config = config
- assert self.config
- self.args = args
- self.kwargs = kwargs
-
- def run(self, continuation=False):
- # XXX we should not use py.test.ensuretemp here
- count = counter()
- tempdir = py.test.ensuretemp("box%d" % count)
- self.tempdir = tempdir
- self.PYTESTRETVAL = tempdir.join('retval')
- self.PYTESTSTDOUT = tempdir.join('stdout')
- self.PYTESTSTDERR = tempdir.join('stderr')
-
- nice_level = self.config.getvalue('dist_nicelevel')
- pid = os.fork()
- if pid:
- if not continuation:
- self.parent(pid)
- else:
- return self.parent, pid
- else:
- try:
- outcome = self.children(nice_level)
- except:
- excinfo = py.code.ExceptionInfo()
- x = open("/tmp/traceback", "w")
- print >>x, "Internal box error"
- for i in excinfo.traceback:
- print >>x, str(i)[2:-1]
- print >>x, excinfo
- x.close()
- os._exit(1)
- os.close(1)
- os.close(2)
- os._exit(0)
- return pid
-
- def children(self, nice_level):
- # right now we need to call a function, but first we need to
- # map all IO that might happen
- # make sure sys.stdout points to file descriptor one
- sys.stdout = stdout = self.PYTESTSTDOUT.open('w')
- sys.stdout.flush()
- fdstdout = stdout.fileno()
- if fdstdout != 1:
- os.dup2(fdstdout, 1)
- sys.stderr = stderr = self.PYTESTSTDERR.open('w')
- fdstderr = stderr.fileno()
- if fdstderr != 2:
- os.dup2(fdstderr, 2)
- retvalf = self.PYTESTRETVAL.open("w")
- try:
- if nice_level:
- os.nice(nice_level)
- # with fork() we have duplicated py.test's basetemp
- # directory so we want to set it manually here.
- # this may be expensive for some test setups,
- # but that is what you get with boxing.
- # XXX but we are called in more than strict boxing
- # mode ("AsyncExecutor") so we can't do the following without
- # inflicting on --dist speed, hum:
- # pytestconfig.basetemp = self.tempdir.join("childbasetemp")
- retval = self.fun(*self.args, **self.kwargs)
- retvalf.write(marshal.dumps(retval))
- finally:
- stdout.close()
- stderr.close()
- retvalf.close()
- os._exit(0)
-
- def parent(self, pid, waiter=os.waitpid):
- pid, exitstat = waiter(pid, 0)
- self.signal = exitstat & 0x7f
- self.exitstat = exitstat & 0xff00
-
-
- if not exitstat:
- retval = self.PYTESTRETVAL.open()
- try:
- retval_data = retval.read()
- finally:
- retval.close()
- self.retval = marshal.loads(retval_data)
- else:
- self.retval = None
-
- self.stdoutrepr = self.PYTESTSTDOUT.read()
- self.stderrrepr = self.PYTESTSTDERR.read()
- return self.stdoutrepr, self.stderrrepr
-
-Box = FileBox
Deleted: /py/branch/reporter-merge/py/test/rsession/executor.py
==============================================================================
--- /py/branch/reporter-merge/py/test/rsession/executor.py Sun Oct 28 19:16:47 2007
+++ (empty file)
@@ -1,136 +0,0 @@
-""" Remote executor
-"""
-
-import py, os, sys
-
-from py.__.test.outcome import SerializableOutcome, ReprOutcome
-from py.__.test.rsession.box import Box
-from py.__.test import repevent
-from py.__.test.outcome import Skipped, Failed
-import py.__.test.custompdb
-
-class RunExecutor(object):
- """ Same as in executor, but just running run
- """
- wraps = False
-
- def __init__(self, item, usepdb=False, reporter=None, config=None):
- self.item = item
- self.usepdb = usepdb
- self.reporter = reporter
- self.config = config
- assert self.config
-
- def run(self, capture=True):
- if capture:
- self.item.startcapture()
- try:
- self.item.run()
- finally:
- self.item.finishcapture()
- else:
- self.item.run()
-
- def execute(self, capture=True):
- try:
- self.run(capture)
- outcome = SerializableOutcome()
- except Skipped:
- e = py.code.ExceptionInfo()
- outcome = SerializableOutcome(skipped=e)
- except (SystemExit, KeyboardInterrupt):
- raise
- except:
- e = sys.exc_info()[1]
- if isinstance(e, Failed) and e.excinfo:
- excinfo = e.excinfo
- else:
- excinfo = py.code.ExceptionInfo()
- if isinstance(self.item, py.test.collect.Function):
- fun = self.item.obj # hope this is stable
- code = py.code.Code(fun)
- excinfo.traceback = excinfo.traceback.cut(
- path=code.path, firstlineno=code.firstlineno)
- outcome = SerializableOutcome(excinfo=excinfo, setupfailure=False)
- if self.usepdb:
- if self.reporter is not None:
- self.reporter(repevent.ImmediateFailure(self.item,
- ReprOutcome(outcome.make_repr
- (self.config.option.tbstyle))))
- py.__.test.custompdb.post_mortem(excinfo._excinfo[2])
- # XXX hmm, we probably will not like to continue from that
- # point
- raise SystemExit()
- outcome.stdout, outcome.stderr = self.item._getouterr()
- return outcome
-
-class ApigenExecutor(RunExecutor):
- """ Same as RunExecutor, but takes tracer to trace calls as
- an argument to execute
- """
- def execute(self, tracer):
- self.tracer = tracer
- return super(ApigenExecutor, self).execute()
-
- def wrap_underlaying(self, target, *args):
- try:
- self.tracer.start_tracing()
- return target(*args)
- finally:
- self.tracer.end_tracing()
-
- def run(self, capture):
- """ We want to trace *only* function objects here. Unsure
- what to do with custom collectors at all
- """
- if hasattr(self.item, 'obj') and type(self.item) is py.test.collect.Function:
- self.item.execute = self.wrap_underlaying
- self.item.run()
-
-class BoxExecutor(RunExecutor):
- """ Same as RunExecutor, but boxes test instead
- """
- wraps = True
-
- def execute(self):
- def fun():
- outcome = RunExecutor.execute(self, False)
- return outcome.make_repr(self.config.option.tbstyle)
- b = Box(fun, config=self.config)
- pid = b.run()
- assert pid
- if b.retval is not None:
- passed, setupfailure, excinfo, skipped, critical, _, _, _\
- = b.retval
- return (passed, setupfailure, excinfo, skipped, critical, 0,
- b.stdoutrepr, b.stderrrepr)
- else:
- return (False, False, None, False, False, b.signal,
- b.stdoutrepr, b.stderrrepr)
-
-class AsyncExecutor(RunExecutor):
- """ same as box executor, but instead it returns function to continue
- computations (more async mode)
- """
- wraps = True
-
- def execute(self):
- def fun():
- outcome = RunExecutor.execute(self, False)
- return outcome.make_repr(self.config.option.tbstyle)
-
- b = Box(fun, config=self.config)
- parent, pid = b.run(continuation=True)
-
- def cont(waiter=os.waitpid):
- parent(pid, waiter=waiter)
- if b.retval is not None:
- passed, setupfailure, excinfo, skipped,\
- critical, _, _, _ = b.retval
- return (passed, setupfailure, excinfo, skipped, critical, 0,
- b.stdoutrepr, b.stderrrepr)
- else:
- return (False, False, None, False, False,
- b.signal, b.stdoutrepr, b.stderrrepr)
-
- return cont, pid
Modified: py/branch/reporter-merge/py/test/rsession/rsession.py
==============================================================================
--- py/branch/reporter-merge/py/test/rsession/rsession.py (original)
+++ py/branch/reporter-merge/py/test/rsession/rsession.py Sun Oct 28 19:16:47 2007
@@ -85,72 +85,3 @@
keyword = self.config.option.keyword
itemgenerator = itemgen(self, colitems, reporter, keyword)
all_tests = dispatch_loop(nodes, itemgenerator, checkfun)
-
-class LSession(AbstractSession):
- """ Local version of session
- """
- def main(self, reporter=None, runner=None):
- # check out if used options makes any sense
- config = self.config
- hosts = [HostInfo('localhost')]
-
- #reporter, checkfun = self.init_reporter(reporter, config, hosts)
- reporter = self.reporter
- checkfun = self.shouldstop
-
- reporter(repevent.TestStarted(hosts, config, []))
- colitems = self.config.getcolitems()
- reporter(repevent.RsyncFinished())
-
- if runner is None:
- runner = self.init_runner()
-
- keyword = self.config.option.keyword
-
- itemgenerator = itemgen(self, colitems, reporter, keyword)
- local_loop(self, reporter, itemgenerator, checkfun, self.config, runner=runner)
-
- retval = reporter(repevent.TestFinished())
-
- if not self.config.option.nomagic:
- py.magic.revoke(assertion=1)
-
- self.write_docs()
- return retval
-
- def write_docs(self):
- if self.config.option.apigen:
- from py.__.apigen.tracer.docstorage import DocStorageAccessor
- apigen = py.path.local(self.config.option.apigen).pyimport()
- if not hasattr(apigen, 'build'):
- raise NotImplementedError("%s does not contain 'build' "
- "function" %(apigen,))
- print >>sys.stderr, 'building documentation'
- capture = py.io.StdCaptureFD()
- try:
- pkgdir = py.path.local(self.config.args[0]).pypkgpath()
- apigen.build(pkgdir,
- DocStorageAccessor(self.docstorage),
- capture)
- finally:
- capture.reset()
- print >>sys.stderr, '\ndone'
-
- def init_runner(self):
- if self.config.option.apigen:
- from py.__.apigen.tracer.tracer import Tracer, DocStorage
- pkgdir = py.path.local(self.config.args[0]).pypkgpath()
- apigen = py.path.local(self.config.option.apigen).pyimport()
- if not hasattr(apigen, 'get_documentable_items'):
- raise NotImplementedError("Provided script does not seem "
- "to contain get_documentable_items")
- pkgname, items = apigen.get_documentable_items(pkgdir)
- self.docstorage = DocStorage().from_dict(items,
- module_name=pkgname)
- self.tracer = Tracer(self.docstorage)
- return apigen_runner
- elif self.config.option.boxed:
- return box_runner
- else:
- return plain_runner
-
Deleted: /py/branch/reporter-merge/py/test/rsession/testing/test_boxing.py
==============================================================================
--- /py/branch/reporter-merge/py/test/rsession/testing/test_boxing.py Sun Oct 28 19:16:47 2007
+++ (empty file)
@@ -1,95 +0,0 @@
-
-""" test boxing functionality
-"""
-
-import py, sys, os
-
-if sys.platform == 'win32':
- py.test.skip("rsession is unsupported on Windows.")
-
-from py.__.test.rsession.box import Box
-from py.__.test.rsession.testing import example2
-
-def setup_module(mod):
- tmpdir = py.test.ensuretemp("boxtests")
- mod.config = py.test.config._reparse([tmpdir])
-
-def test_basic_boxing():
- # XXX: because we do not have option transfer
-## if not hasattr(option, 'nocapture') or not option.nocapture:
-## py.test.skip("Interacts with pylib i/o skipping which is bad actually")
- b = Box(example2.boxf1, config=config)
- b.run()
- assert b.stdoutrepr == "some out\n"
- assert b.stderrrepr == "some err\n"
- assert b.exitstat == 0
- assert b.signal == 0
- assert b.retval == 1
-
-def test_boxing_on_fds():
- b = Box(example2.boxf2, config=config)
- b.run()
- assert b.stdoutrepr == "someout"
- assert b.stderrrepr == "someerr"
- assert b.exitstat == 0
- assert b.signal == 0
- assert b.retval == 2
-
-def test_boxing_signal():
- b = Box(example2.boxseg, config=config)
- b.run()
- assert b.retval is None
- if py.std.sys.version_info < (2,4):
- py.test.skip("signal detection does not work with python prior 2.4")
- assert b.signal == 11
-
-def test_boxing_huge_data():
- b = Box(example2.boxhuge, config=config)
- b.run()
- assert b.stdoutrepr
- assert b.exitstat == 0
- assert b.signal == 0
- assert b.retval == 3
-
-def test_box_seq():
- # we run many boxes with huge data, just one after another
- for i in xrange(100):
- b = Box(example2.boxhuge, config=config)
- b.run()
- assert b.stdoutrepr
- assert b.exitstat == 0
- assert b.signal == 0
- assert b.retval == 3
-
-def test_box_in_a_box():
- def boxfun():
- b = Box(example2.boxf2, config=config)
- b.run()
- print b.stdoutrepr
- print >>sys.stderr, b.stderrrepr
- return b.retval
-
- b = Box(boxfun, config=config)
- b.run()
- assert b.stdoutrepr == "someout\n"
- assert b.stderrrepr == "someerr\n"
- assert b.exitstat == 0
- assert b.signal == 0
- assert b.retval == 2
-
-def test_box_killer():
- class A:
- pass
- info = A()
- import time
-
- def box_fun():
- time.sleep(10) # we don't want to last forever here
-
- b = Box(box_fun, config=config)
- par, pid = b.run(continuation=True)
- os.kill(pid, 15)
- par(pid)
- if py.std.sys.version_info < (2,4):
- py.test.skip("signal detection does not work with python prior 2.4")
- assert b.signal == 15
Deleted: /py/branch/reporter-merge/py/test/rsession/testing/test_executor.py
==============================================================================
--- /py/branch/reporter-merge/py/test/rsession/testing/test_executor.py Sun Oct 28 19:16:47 2007
+++ (empty file)
@@ -1,171 +0,0 @@
-
-import py
-import example1
-
-from py.__.test.rsession.executor import RunExecutor, BoxExecutor,\
- AsyncExecutor, ApigenExecutor
-from py.__.test.outcome import ReprOutcome
-from py.__.test.rsession.testing.basetest import BasicRsessionTest
-from py.__.test.outcome import Failed
-
-def setup_module(mod):
- if py.std.sys.platform == "win32":
- py.test.skip("skipping executor tests (some require os.fork)")
-
-class Item(py.test.collect.Item):
- def __init__(self, name, config):
- super(Item, self).__init__(name)
- self._config = config
-
-class ItemTestPassing(Item):
- def run(self):
- return None
-
-class ItemTestFailing(Item):
- def run(self):
- assert 0 == 1
-
-class ItemTestSkipping(Item):
- def run(self):
- py.test.skip("hello")
-
-class ItemTestPrinting(Item):
- def run(self):
- print "hello"
-
-class ItemTestFailingExplicit(Item):
- def run(self):
- raise Failed(excinfo="3")
-
-class ItemTestFailingExplicitEmpty(Item):
- def run(self):
- py.test.raises(ValueError, lambda : 123)
-
-class TestExecutor(BasicRsessionTest):
- def test_run_executor(self):
- ex = RunExecutor(ItemTestPassing("pass", self.config), config=self.config)
- outcome = ex.execute()
- assert outcome.passed
-
- ex = RunExecutor(ItemTestFailing("fail", self.config), config=self.config)
- outcome = ex.execute()
- assert not outcome.passed
-
- ex = RunExecutor(ItemTestSkipping("skip", self.config), config=self.config)
- outcome = ex.execute()
- assert outcome.skipped
- assert not outcome.passed
- assert not outcome.excinfo
-
- def test_run_executor_capture(self):
- ex = RunExecutor(ItemTestPrinting("print", self.config), config=self.config)
- outcome = ex.execute()
- assert outcome.stdout == "hello\n"
-
- def test_box_executor(self):
- ex = BoxExecutor(ItemTestPassing("pass", self.config), config=self.config)
- outcome_repr = ex.execute()
- outcome = ReprOutcome(outcome_repr)
- assert outcome.passed
-
- ex = BoxExecutor(ItemTestFailing("fail", self.config), config=self.config)
- outcome_repr = ex.execute()
- outcome = ReprOutcome(outcome_repr)
- assert not outcome.passed
-
- ex = BoxExecutor(ItemTestSkipping("skip", self.config), config=self.config)
- outcome_repr = ex.execute()
- outcome = ReprOutcome(outcome_repr)
- assert outcome.skipped
- assert not outcome.passed
- assert not outcome.excinfo
-
- def test_box_executor_stdout(self):
- item = self.getexample("print")
- ex = BoxExecutor(item, config=self.config)
- outcome_repr = ex.execute()
- outcome = ReprOutcome(outcome_repr)
- assert outcome.passed
- assert outcome.stdout.find("samfing") != -1
-
- def test_box_executor_stdout_error(self):
- item = self.getexample("printfail")
- ex = BoxExecutor(item, config=self.config)
- outcome_repr = ex.execute()
- outcome = ReprOutcome(outcome_repr)
- assert not outcome.passed
- assert outcome.stdout.find("samfing elz") != -1
-
- def test_cont_executor(self):
- item = self.getexample("printfail")
- ex = AsyncExecutor(item, config=self.config)
- cont, pid = ex.execute()
- assert pid
- outcome_repr = cont()
- outcome = ReprOutcome(outcome_repr)
- assert not outcome.passed
- assert outcome.stdout.find("samfing elz") != -1
-
- def test_apigen_executor(self):
- class Tracer(object):
- def __init__(self):
- self.starts = 0
- self.ends = 0
-
- def start_tracing(self):
- self.starts += 1
-
- def end_tracing(self):
- self.ends += 1
-
- tmpdir = py.test.ensuretemp("apigen_executor")
- tmpdir.ensure("__init__.py")
- tmpdir.ensure("test_one.py").write(py.code.Source("""
- def g():
- pass
-
- def test_1():
- g()
-
- class TestX(object):
- def setup_method(self, m):
- self.ttt = 1
-
- def test_one(self):
- self.ttt += 1
-
- def test_raise(self):
- 1/0
- """))
- config = py.test.config._reparse([tmpdir])
- rootcol = config._getcollector(tmpdir)
- tracer = Tracer()
- item = rootcol._getitembynames("test_one.py/test_1")
- ex = ApigenExecutor(item, config=config)
- out1 = ex.execute(tracer)
- item = rootcol._getitembynames("test_one.py/TestX/()/test_one")
- ex = ApigenExecutor(item, config=config)
- out2 = ex.execute(tracer)
- item = rootcol._getitembynames("test_one.py/TestX/()/test_raise")
- ex = ApigenExecutor(item, config=config)
- out3 = ex.execute(tracer)
- assert tracer.starts == 3
- assert tracer.ends == 3
- assert out1.passed
- assert out2.passed
- assert not out3.passed
-
- def test_executor_explicit_Failed(self):
- ex = RunExecutor(ItemTestFailingExplicit("failex", self.config),
- config=self.config)
-
- outcome = ex.execute()
- assert not outcome.passed
- assert outcome.excinfo == "3"
-
- def test_executor_explicit_Faile_no_excinfo(self):
- ex = RunExecutor(ItemTestFailingExplicitEmpty("failexx", self.config),
- config=self.config)
- outcome = ex.execute()
- assert not outcome.passed
-
Copied: py/branch/reporter-merge/py/test/test_executor.py (from r47660, py/branch/reporter-merge/py/test/rsession/testing/test_executor.py)
==============================================================================
--- py/branch/reporter-merge/py/test/rsession/testing/test_executor.py (original)
+++ py/branch/reporter-merge/py/test/test_executor.py Sun Oct 28 19:16:47 2007
@@ -2,7 +2,7 @@
import py
import example1
-from py.__.test.rsession.executor import RunExecutor, BoxExecutor,\
+from py.__.test.executor import RunExecutor, BoxExecutor,\
AsyncExecutor, ApigenExecutor
from py.__.test.outcome import ReprOutcome
from py.__.test.rsession.testing.basetest import BasicRsessionTest
More information about the pytest-commit
mailing list