Problem 2: On my network in my workshop here, these tests take 1m30s to complete. When I have to wait that long I get distracted and start reading e-mail, building little ships in bottles, or something, and my productivity plummets.
I've been encountering problem #2 just now as I'm running the Shtoom unit tests to test my new playout patch.
Proposed solution in this patch: move tests which require something in addition to working-Shtoom-source-code into a separate dir named "livetest". After this patch it takes 1m30s to do "trial shtoom.livetest" and it takes 2s to do "trial shtoom.test". This helps me work more efficiently.
Regards,
Zooko
Index: shtoom/test/test_stun.py
===================================================================
--- shtoom/test/test_stun.py (revision 1262)
+++ shtoom/test/test_stun.py (working copy)
@@ -1,78 +0,0 @@
-# Copyright (C) 2004 Anthony Baxter
-"""Tests for STUN.
-
-You can run this with command-line:
-
- $ trial shtoom.test.test_stun
-"""
-
-from twisted.trial import unittest, util
-from twisted.internet import defer, reactor
-from twisted.internet.protocol import Factory, Protocol, DatagramProtocol
-
-class Saver:
- def save(self, arg):
- self.arg = arg
-
-class NetAddressTests(unittest.TestCase):
-
- def testNetAddressInit(self):
- from shtoom.stun import NetAddress
- ae = self.assertEquals
- ar = self.assertRaises
- n1 = NetAddress('10/8')
- n2 = NetAddress('10.0.0.0/8')
- ae((n1.net,n1.mask), (n2.net, n2.mask))
- n1 = NetAddress('10.1.2.3/32')
- n2 = NetAddress('10.1.2.3')
- ae((n1.net,n1.mask), (n2.net, n2.mask))
- n1 = NetAddress('10.1.2.3/32')
- n2 = NetAddress('10.1.2.4')
- ae((n1.net+1,n1.mask), (n2.net, n2.mask))
- ar(ValueError, NetAddress, '10.1.2.3/32/3')
- ar(ValueError, NetAddress, '10.1.2.3/33')
-
- def testNetAddressMembership(self):
- from shtoom.stun import NetAddress
- ae = self.assertEquals
- ar = self.assertRaises
- n1 = NetAddress('10/8')
- ae(n1.check('10.1.2.3'), True)
- ae(n1.check('10.0'), True)
- ae(n1.check('10.0.0.0'), True)
- ae('10.1.2.3' in n1, True)
- ae('10.0.0.0' in n1, True)
- ae(n1.check('10.255.255.255'), True)
- ae(n1.check('10.2'), True)
- ae(n1.check('11.0'), False)
- ae(n1.check('9.255.255.255'), False)
- ae(n1.check('255.255.255.255'), False)
- n2 = NetAddress('172.16/12')
- ae(n2.check('172.16.0.0'), True)
- ae(n2.check('172.31.255.255'), True)
- ae(n2.check('172.20.10.200'), True)
- ae(n2.check('172.32.0.0'), False)
- ae(n2.check('172.15.255.255'), False)
- ae(NetAddress('172.20/16') in n2, True)
- ae(NetAddress('172.20.1.2/16') in n2, True)
- ae(NetAddress('172.16/12') in n2, True)
- ae(NetAddress('172.0/8') in n2, False)
- n3 = NetAddress('224.255.0.0')
- ae(n3.check('224.255.0.0'), True)
- ae(n3.check('224.255.0.1'), False)
- ae(n3.check('224.254.255.255'), False)
- n4 = NetAddress('0/0')
- ae(n4.check('10.0.0.0'), True)
- ae(n4.check('172.20.10.200'), True)
- ae(n4.check('224.255.0.1'), True)
-
- def test_stundiscovery(self):
- from shtoom.stun import getSTUN, _NatType
- ae = self.assertEquals
- a_ = self.assert_
- d = getSTUN()
- s = Saver()
- d.addCallback(s.save)
- util.wait(d, timeout=16)
- a_(isinstance(s.arg, _NatType), "%s, %s :: %s" % (s, s.arg,
type(s.arg),))
-
Index: shtoom/test/test_natmappers.py
===================================================================
--- shtoom/test/test_natmappers.py (revision 1262)
+++ shtoom/test/test_natmappers.py (working copy)
@@ -1,149 +0,0 @@
-# Copyright (C) 2004 Anthony Baxter
-"""Tests for shtoom.nat and friends
-"""
-
-# This is more a functional test than a unit test. oh well.
-
-from twisted.trial import unittest
-from twisted.trial import util
-from twisted.internet import reactor, defer
-from twisted.internet.protocol import Protocol, Factory, DatagramProtocol
-
-from shtoom.test.test_upnp import checkUPnP
-
-import random
-
-def logerr(failure):
- print "logerr", failure
- print failure.value
- return failure
-
-class TestMapper:
- def __init__(self, mapper, port):
- self.port = port
- self.mapper = mapper
-
- def go(self):
- #print "going"
- cd = defer.Deferred()
- d = self.mapper.map(self.port)
- d.addCallback(lambda x: self.cb_mapped(x, cd)).addErrback(logerr)
- return cd
-
- def cb_mapped(self, ext, cd):
- #print "mapped"
- self.map_res = ext
- self.info_res = self.mapper.info(self.port)
- d = self.mapper.unmap(self.port)
- d.addCallback(lambda x: self.cb_unmapped(x, cd))
-
- def cb_unmapped(self, res, cd):
- #print "unmapped"
- self.unmap_res = res
- cd.callback(None)
-
-class Saver:
- def save(self, arg):
- self.arg = arg
-
-
-class MapperTest(unittest.TestCase):
-
- def tearDown(self):
- import shtoom.nat
- shtoom.nat.clearCache()
-
- def test_upnp_mapper(self):
- from shtoom.upnp import UPnPMapper
- ae = self.assertEquals
- ar = self.assertRaises
- checkUPnP()
- mapper = UPnPMapper()
- uprot = DatagramProtocol()
- uport = reactor.listenUDP(random.randint(10000,26000), uprot)
- class tfactory(Factory):
- protocol = Protocol
- tport = reactor.listenTCP(0, tfactory())
- for port in uport, tport:
- ar(ValueError, mapper.unmap, port)
- ar(ValueError, mapper.info, port)
- t = TestMapper(mapper, port)
- d = t.go()
- d.addErrback(logerr)
- util.wait(d, timeout=16)
- ae(len(t.map_res), 2)
- ae(t.map_res, t.info_res)
- ae(t.unmap_res, None)
- # Can't unmap a port that's not mapped
- ar(ValueError, mapper.unmap, port)
- d = port.stopListening()
- util.wait(d)
- # Can't map a closed port
- ar(ValueError, mapper.map, port)
- # Can't get info on a closed port
- ar(ValueError, mapper.info, port)
-
- def test_stunmapper(self):
- from shtoom.stun import getSTUN, STUNMapper
- ae = self.assertEquals
- ar = self.assertRaises
-
- d = getSTUN()
- s = Saver()
- d.addCallback(s.save)
- util.wait(d, timeout=64)
- if not s.arg.useful:
- raise unittest.SkipTest('No useful STUN')
-
- mapper = STUNMapper()
- # Check we have a hate-on for TCP ports
- class tfactory(Factory):
- protocol = Protocol
- tport = reactor.listenTCP(0, tfactory())
- ar(ValueError, mapper.map, tport)
- tport.stopListening()
- ar(ValueError, mapper.map, tport)
-
- prot = DatagramProtocol()
- port = reactor.listenUDP(0, prot)
- ar(ValueError, mapper.info, port)
- t = TestMapper(mapper, port)
- d = t.go()
- util.wait(d, timeout=512)
- ae(len(t.map_res), 2)
- ae(t.map_res, t.info_res)
- ae(t.unmap_res, None)
- d = port.stopListening()
- util.wait(d)
- ar(ValueError, mapper.map, port)
- ar(ValueError, mapper.info, port)
-
- def test_nullmapper(self):
- from shtoom.nat import getNullMapper
- mapper = getNullMapper()
- ae = self.assertEquals
- ar = self.assertRaises
- uprot = DatagramProtocol()
- uport = reactor.listenUDP(0, uprot)
- class tfactory(Factory):
- protocol = Protocol
- tport = reactor.listenTCP(0, tfactory())
- for port in uport, tport:
- ar(ValueError, mapper.unmap, port)
- ar(ValueError, mapper.info, port)
- t = TestMapper(mapper, port)
- d = t.go()
- util.wait(d)
- ae(len(t.map_res), 2)
- ae(t.map_res, t.info_res)
- ae(t.unmap_res, None)
- # Can't unmap a port that's not mapped
- ar(ValueError, mapper.unmap, port)
- d = port.stopListening()
- util.wait(d)
- # Can't map a closed port
- ar(ValueError, mapper.map, port)
- # Can't get info on a closed port
- ar(ValueError, mapper.info, port)
-
-
Index: shtoom/test/test_defcache.py
===================================================================
--- shtoom/test/test_defcache.py (revision 1262)
+++ shtoom/test/test_defcache.py (working copy)
@@ -1,216 +0,0 @@
-# Copyright (C) 2004 Anthony Baxter
-
-"""Tests for shtoom.defcache.
-"""
-
-# XXX Todo:
-# - Test errbacks
-# - Test that errors in user callbacks don't interfere with other
-# results, and neither do deferreds returned by user callbacks
-# - More tests of the decorated versions
-
-from twisted.trial import unittest
-from shtoom.defcache import DeferredCache
-from twisted.internet import defer, reactor
-from twisted.trial import util
-
-class TestBlobby:
- def __init__(self):
- self.calls = []
- self.operation = DeferredCache(self._operation, inProgressOnly=False)
-
- def _operation(self, *args, **kwargs):
- # Stub pointless operation - returns the first value passed,
- # after a small delay
- self.calls.append((args,kwargs))
- opdef = defer.Deferred()
- reactor.callLater(0, lambda : opdef.callback(args[0]))
- return opdef
-
- operation2 = DeferredCache(_operation, hashableArgs=True,
inProgressOnly=False)
- operation3 = DeferredCache(_operation, inProgressOnly=True)
-
-class Saver:
- def __init__(self):
- self.val = None
-
- def save(self, arg):
- #print "Saver", self, "called with", arg
- self.val = arg
-
-
-class DefcacheTests(unittest.TestCase):
-
- def test_defcache(self):
- ae = self.assertEquals
- ar = self.assertRaises
-
- t = TestBlobby()
- s = Saver()
- d = t.operation('foo')
- d.addCallback(s.save)
- util.wait(d)
- ae(s.val, 'foo')
-
- t = TestBlobby()
- s1 = Saver()
- s2 = Saver()
- d1 = t.operation('foo')
- d2 = t.operation('foo')
- d1.addCallback(s1.save)
- d2.addCallback(s2.save)
- util.wait(d1)
- util.wait(d2)
- ae(s1.val, 'foo')
- ae(s2.val, 'foo')
- # Check it was only called once, as expected
- ae(t.calls, [(('foo',),{})])
- # Check for the cache of completed calls
- d3 = t.operation('foo')
- s3 = Saver()
- d3.addCallback(s3.save)
- util.wait(d3)
- ae(s3.val, 'foo')
- ae(t.calls, [(('foo',),{})])
-
- # Now test kwargs
- t = TestBlobby()
- s1 = Saver()
- s2 = Saver()
- s3 = Saver()
- d1 = t.operation('foo', kw=True)
- d2 = t.operation('foo', kw=True)
- d3 = t.operation('foo', kw=False)
- d1.addCallback(s1.save)
- d2.addCallback(s2.save)
- d3.addCallback(s3.save)
- util.wait(d1)
- util.wait(d2)
- util.wait(d3)
- ae(s1.val, 'foo')
- ae(s2.val, 'foo')
- ae(s3.val, 'foo')
- # Check it was called twice only
- ae(len(t.calls), 2)
-
- t = TestBlobby()
- ar(TypeError, t.operation2, 'bar', {})
-
- s1 = Saver()
- s2 = Saver()
- d1 = t.operation2('bar', kw=True)
- d2 = t.operation('bar', kw=True)
- d1.addCallback(s1.save)
- d2.addCallback(s2.save)
- util.wait(d1)
- util.wait(d2)
- ae(s1.val, 'bar')
- ae(s2.val, 'bar')
- # Check it was called twice only
- ae(len(t.calls), 2)
-
- t = TestBlobby()
-
- s1 = Saver()
- s2 = Saver()
- d1 = t.operation('foo', kw={})
- d2 = t.operation('foo', kw={})
- d1.addCallback(s1.save)
- d2.addCallback(s2.save)
- util.wait(d1)
- util.wait(d2)
- ae(s1.val, 'foo')
- ae(s2.val, 'foo')
- # Check it was called both times
- ae(len(t.calls), 2)
-
- t = TestBlobby()
- s1 = Saver()
- s2 = Saver()
- s3 = Saver()
- s4 = Saver()
- d1 = t.operation3('foo', 1, 2, 3)
- d2 = t.operation3('foo', 1, 2, 3)
- d1.addCallback(s1.save)
- d2.addCallback(s2.save)
- util.wait(d1)
- util.wait(d2)
- ae(s1.val, 'foo')
- ae(s2.val, 'foo')
- ae(len(t.calls), 1)
- d3 = t.operation3('foo', 1, 2, 3)
- d4 = t.operation3('foo', 1, 2, 3)
- d3.addCallback(s3.save)
- d4.addCallback(s4.save)
- util.wait(d3)
- util.wait(d4)
- ae(s3.val, 'foo')
- ae(s4.val, 'foo')
- ae(len(t.calls), 2)
-
- t = TestBlobby()
- s1 = Saver()
- s2 = Saver()
- d1 = t.operation('foo')
- d2 = t.operation('foo')
- d1.addCallback(s1.save)
- d2.addCallback(s2.save)
- util.wait(d1)
- util.wait(d2)
- ae(s1.val, 'foo')
- ae(s2.val, 'foo')
- # Check it was only called once, as expected
- ae(len(t.calls), 1)
- # Clear the cache
- t.operation.clearCache()
- d3 = t.operation('foo')
- s3 = Saver()
- d3.addCallback(s3.save)
- util.wait(d3)
- ae(s3.val, 'foo')
- ae(len(t.calls), 2)
-
- # Now test kwargs
- t = TestBlobby()
- s1 = Saver()
- s2 = Saver()
- s3 = Saver()
- d1 = t.operation('foo', kw=True)
- d2 = t.operation('foo', kw=True)
- d3 = t.operation('foo', kw=False)
- d1.addCallback(s1.save)
- d2.addCallback(s2.save)
- d3.addCallback(s3.save)
- util.wait(d1)
- util.wait(d2)
- util.wait(d3)
- ae(s1.val, 'foo')
- ae(s2.val, 'foo')
- ae(s3.val, 'foo')
- # Check it was called twice only
- ae(len(t.calls), 2)
-
- def test_decorating(self):
- import sys
- if sys.version_info < (2,4):
- raise unittest.SkipTest('decorators only available on 2.4 and
later')
- ae = self.assertEquals
- ar = self.assertRaises
-
- # Has to be elsewhere to avoid SyntaxErrors :-(
- from shtoom.test.py24tests import Dectest
-
- # XXX actually test the stupid things?!
- d = Dectest()
- ae(d.foo.cache_inProgressOnly, False)
- ae(d.foo.cache_inProgressOnly, False)
- ae(d.foo2.cache_inProgressOnly, True)
- ae(d.foo3.cache_inProgressOnly, True)
- ae(d.foo3.cache_hashableArgs, False)
- ar(TypeError, d.foo2, {})
- s = Saver()
- d = d.foo(1,2,3)
- d.addCallback(s.save)
- util.wait(d)
- ae(s.val, (1,2,3))
-
Index: shtoom/test/test_upnp.py
===================================================================
--- shtoom/test/test_upnp.py (revision 1262)
+++ shtoom/test/test_upnp.py (working copy)
@@ -1,136 +0,0 @@
-# Copyright (C) 2004 Anthony Baxter
-"""Tests for shtoom.upnp
-"""
-
-# This is more a functional test than a unit test. oh well.
-
-from twisted.trial import unittest
-from twisted.trial import util
-from twisted.internet import reactor, defer
-from twisted.internet.protocol import Protocol, Factory, DatagramProtocol
-
-import random
-
-def checkUPnP():
- from shtoom.upnp import getUPnP
- d = getUPnP()
- s = Saver()
- d.addCallback(s.save)
- util.wait(d, timeout=8)
- if s.val is None:
- raise unittest.SkipTest('no UPnP available')
-
-class TestUPnP:
-
- def discovered(self, prot):
- #print "upnp discovery done,", prot
- self.prot = prot
- return self.prot.getExternalIPAddress()
-
- def gotexternal(self, res):
- #print "upnp external address,",res
- return self.prot.getPortMappings()
-
- def gotmappings(self, res):
- #print "upnp mappings:", res
- return self.prot.addPortMapping(12367,12367,'test mapping')
-
- def added(self, res):
- #print "upnp port mapping added"
- return self.prot.getPortMappings()
-
- def gotmappings2(self, res):
- #print "upnp mappings:", res
- return self.prot.deletePortMapping(12367)
-
- def deleted(self, res):
- #print "upnp port mapping removed"
- return self.prot.getPortMappings()
-
- def gotmappings3(self, res):
- #print "upnp mappings:", res
- return self.prot.getExternalIPAddress()
-
- def gotexternal2(self, res):
- #print "upnp external address,",res
- pass
-
- def go(self):
- from shtoom.upnp import getUPnP
- d = getUPnP()
- d.addCallback(self.discovered)
- d.addCallback(self.gotexternal)
- d.addCallback(self.gotmappings)
- d.addCallback(self.added)
- d.addCallback(self.gotmappings2)
- d.addCallback(self.deleted)
- d.addCallback(self.gotmappings3)
- d.addCallback(self.gotexternal2)
- return d
-
-class Saver:
- def __init__(self):
- self.val = None
- def save(self, val):
- self.val = val
-
-class TestMapper:
- def __init__(self, mapper, port):
- self.port = port
- self.mapper = mapper
- def go(self):
- #print "going"
- cd = defer.Deferred()
- d = self.mapper.map(self.port)
- d.addCallback(lambda x: self.cb_mapped(x, cd))
- return cd
- def cb_mapped(self, ext, cd):
- #print "mapped"
- self.map_res = ext
- self.info_res = self.mapper.info(self.port)
- d = self.mapper.unmap(self.port)
- d.addCallback(lambda x: self.cb_unmapped(x, cd))
- def cb_unmapped(self, res, cd):
- #print "unmapped"
- self.unmap_res = res
- cd.callback(None)
-
-class UPnPTest(unittest.TestCase):
-
-
- def test_upnp(self):
- checkUPnP()
- test = TestUPnP()
- d = test.go()
- util.wait(d, timeout=32)
-
- def test_upnp_mapper(self):
- from shtoom.upnp import UPnPMapper
- ae = self.assertEquals
- ar = self.assertRaises
- checkUPnP()
- mapper = UPnPMapper()
- uprot = DatagramProtocol()
- uport = reactor.listenUDP(random.randint(10000,20000), uprot)
- class tfactory(Factory):
- protocol = Protocol
- tport = reactor.listenTCP(0, tfactory())
- for port in uport, tport:
- ar(ValueError, mapper.unmap, port)
- ar(ValueError, mapper.info, port)
- t = TestMapper(mapper, port)
- d = t.go()
- util.wait(d, timeout=16)
- ae(len(t.map_res), 2)
- ae(t.map_res, t.info_res)
- ae(t.unmap_res, None)
- # Can't unmap a port that's not mapped
- ar(ValueError, mapper.unmap, port)
- d = port.stopListening()
- util.wait(d)
- # Can't map a closed port
- ar(ValueError, mapper.map, port)
- # Can't get info on a closed port
- ar(ValueError, mapper.info, port)
-
-
Index: shtoom/livetest/__init__.py
===================================================================
--- shtoom/livetest/__init__.py (revision 0)
+++ shtoom/livetest/__init__.py (revision 0)
@@ -0,0 +1,7 @@
+"""Live-tests for shtoom. These are tests that require something outside of
Shtoom's source code to pass, such as a network connection, a remote server or
peer, etc.
+
+You can run all tests by command-line:
+
+ $ trial shtoom.livetest
+
+"""
_______________________________________________ Shtoom mailing list [email protected] http://mail.python.org/mailman/listinfo/shtoom
