[svn r57758] unify test support, remove basetest.py

--HG--
branch : trunk
This commit is contained in:
hpk 2008-09-02 15:22:10 +02:00
parent 14cfc8d342
commit 7518dcabc2
4 changed files with 50 additions and 52 deletions

View File

@ -1,32 +0,0 @@
""" Support module for running tests
"""
import py
from py.__.test.testing.setupdata import getexamplefile
class DirSetup(object):
def setup_method(self, method):
name = "%s.%s" %(self.__class__.__name__, method.func_name)
self.tmpdir = py.test.ensuretemp(name)
self.source = self.tmpdir.ensure("source", dir=1)
self.dest = self.tmpdir.join("dest")
class BasicRsessionTest(object):
def setup_class(cls):
path = getexamplefile("test_funcexamples.py")
cls.config = py.test.config._reparse([path.dirpath()])
cls.modulecol = cls.config.getfsnode(path)
def setup_method(self, method):
self.session = self.config.initsession()
def getfunc(self, name):
funcname = "test_func" + name
col = self.modulecol.join(funcname)
assert col is not None, funcname
return col
def getdocexample(self):
return getexamplefile("docexample.txt")

View File

@ -6,7 +6,6 @@ import py
from py.__.test import event
from py.__.test.dsession.dsession import DSession
from py.__.test.dsession.hostmanage import HostManager, Host
from basetest import BasicRsessionTest, DirSetup
from py.__.test.testing import suptest
import os

View File

@ -3,12 +3,18 @@
"""
import py
from basetest import DirSetup
from py.__.test.testing import suptest
from py.__.test.dsession.hostmanage import HostRSync, Host, HostManager, gethosts
from py.__.test.dsession.hostmanage import sethomedir, gethomedir, getpath_relto_home
from py.__.test import event
class TestHost(DirSetup):
class TmpWithSourceDest(suptest.FileCreation):
def setup_method(self, method):
super(TmpWithSourceDest, self).setup_method(method)
self.source = self.tmpdir.mkdir("source")
self.dest = self.tmpdir.mkdir("dest")
class TestHost(suptest.FileCreation):
def _gethostinfo(self, relpath=""):
exampledir = self.tmpdir.join("gethostinfo")
if relpath:
@ -119,7 +125,7 @@ class TestHost(DirSetup):
res = channel.receive()
assert res == host.gw_remotepath
class TestSyncing(DirSetup):
class TestSyncing(TmpWithSourceDest):
def _gethostinfo(self):
hostinfo = Host("localhost:%s" % self.dest)
return hostinfo
@ -181,7 +187,7 @@ class TestSyncing(DirSetup):
res2 = rsync.add_target_host(h2)
assert not res2
class TestHostManager(DirSetup):
class TestHostManager(TmpWithSourceDest):
def gethostmanager(self, dist_hosts, dist_rsync_roots=None):
l = ["dist_hosts = %r" % dist_hosts]
if dist_rsync_roots:

View File

@ -1,11 +1,11 @@
import py
from py.__.test.dsession.masterslave import MasterNode
from py.__.test.dsession.hostmanage import Host
from basetest import BasicRsessionTest
from py.__.test.dsession.hostmanage import Host
from py.__.test import event
from py.__.test.testing import suptest
class TestMasterSlaveConnection(BasicRsessionTest):
class TestMasterSlaveConnection(suptest.InlineCollection):
def getevent(self, eventtype=event.ItemTestReport, timeout=2.0):
events = []
while 1:
@ -23,15 +23,30 @@ class TestMasterSlaveConnection(BasicRsessionTest):
def setup_method(self, method):
super(TestMasterSlaveConnection, self).setup_method(method)
self.makepyfile(__init__="")
self.config = self.parseconfig(self.tmpdir)
self.queue = py.std.Queue.Queue()
self.host = Host("localhost")
self.host.initgateway()
self.node = MasterNode(self.host, self.session.config,
self.queue.put)
self.node = MasterNode(self.host, self.config, self.queue.put)
assert not self.node.channel.isclosed()
def getitem(self, source):
kw = {"test_" + self.tmpdir.basename: py.code.Source(source).strip()}
path = self.makepyfile(**kw)
fscol = self.config.getfsnode(path)
return fscol.collect_by_name("test_func")
def getitems(self, source):
kw = {"test_" + self.tmpdir.basename: py.code.Source(source).strip()}
path = self.makepyfile(**kw)
fscol = self.config.getfsnode(path)
return fscol.collect()
def teardown_method(self, method):
print "at teardown:", self.node.channel
#if not self.node.channel.isclosed():
# self.node.shutdown()
self.host.gw.exit()
def test_crash_invalid_item(self):
@ -43,7 +58,11 @@ class TestMasterSlaveConnection(BasicRsessionTest):
def test_crash_killed(self):
if not hasattr(py.std.os, 'kill'):
py.test.skip("no os.kill")
item = self.getfunc("kill15")
item = self.getitem("""
def test_func():
import os
os.kill(os.getpid(), 15)
""")
self.node.send(item)
ev = self.getevent(event.HostDown)
assert ev.host == self.host
@ -59,15 +78,15 @@ class TestMasterSlaveConnection(BasicRsessionTest):
"self.getevent(event.HostDown, timeout=0.01)")
def test_send_on_closed_channel(self):
item = self.getfunc("passed")
item = self.getitem("def test_func(): pass")
self.node.channel.close()
py.test.raises(IOError, "self.node.send(item)")
#ev = self.getevent(event.InternalException)
#assert ev.excinfo.errisinstance(IOError)
def test_send_one(self):
item = self.getfunc("passed")
self.node.send(self.getfunc("passed"))
item = self.getitem("def test_func(): pass")
self.node.send(item)
ev = self.getevent()
assert ev.passed
assert ev.colitem == item
@ -75,16 +94,22 @@ class TestMasterSlaveConnection(BasicRsessionTest):
#assert event.item is not item
def test_send_some(self):
items = self.getitems("""
def test_pass():
pass
def test_fail():
assert 0
def test_skip():
import py
py.test.skip("x")
""")
for item in items:
self.node.send(item)
for outcome in "passed failed skipped".split():
self.node.send(self.getfunc(outcome))
ev = self.getevent()
assert getattr(ev, outcome)
def test_send_list(self):
l = []
for outcome in "passed failed skipped".split():
l.append(self.getfunc(outcome))
self.node.sendlist(l)
self.node.sendlist(items)
for outcome in "passed failed skipped".split():
ev = self.getevent()
assert getattr(ev, outcome)