RomanNum3ral
/
open_toontown_panda3d
mirror of https://github.com/open-toontown/panda3d.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move test code from across `direct` to proper test modules 

Closes #1480
This commit is contained in:
WMOkiishi 2023-08-02 08:24:44 +02:00 committed by rdb
parent d1f4212acd
commit f7f3179a89
42 changed files with 1336 additions and 1344 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
.github/workflows/ci.yml vendored
View File

@ -207,7 +207,7 @@ jobs:
PYTHONPATH: ${{ matrix.config }}
run: |
PYTHON_EXECUTABLE=$(grep 'Python_EXECUTABLE:' CMakeCache.txt | sed 's/.*=//')
$PYTHON_EXECUTABLE -m pip install pytest pytest-cov
$PYTHON_EXECUTABLE -m pip install -r ../requirements-test.txt
export COVERAGE_FILE=.coverage.$RANDOM LLVM_PROFILE_FILE=$PWD/pid-%p.profraw
$PYTHON_EXECUTABLE -m pytest ../tests --cov=.
# END B
@ -239,7 +239,7 @@ jobs:
PYTHONPATH: ${{ matrix.config }}
run: |
PYTHON_EXECUTABLE=$(grep 'Python_EXECUTABLE:' CMakeCache.txt | sed 's/.*=//')
$PYTHON_EXECUTABLE -m pip install pytest pytest-cov
$PYTHON_EXECUTABLE -m pip install -r ../requirements-test.txt
export COVERAGE_FILE=.coverage.$RANDOM LLVM_PROFILE_FILE=$PWD/pid-%p.profraw
$PYTHON_EXECUTABLE -m pytest ../tests --cov=.
# END B
@ -271,7 +271,7 @@ jobs:
PYTHONPATH: ${{ matrix.config }}
run: |
PYTHON_EXECUTABLE=$(grep 'Python_EXECUTABLE:' CMakeCache.txt | sed 's/.*=//')
$PYTHON_EXECUTABLE -m pip install pytest pytest-cov
$PYTHON_EXECUTABLE -m pip install -r ../requirements-test.txt
export COVERAGE_FILE=.coverage.$RANDOM LLVM_PROFILE_FILE=$PWD/pid-%p.profraw
$PYTHON_EXECUTABLE -m pytest ../tests --cov=.
# END B
@ -303,7 +303,7 @@ jobs:
PYTHONPATH: ${{ matrix.config }}
run: |
PYTHON_EXECUTABLE=$(grep 'Python_EXECUTABLE:' CMakeCache.txt | sed 's/.*=//')
$PYTHON_EXECUTABLE -m pip install pytest pytest-cov
$PYTHON_EXECUTABLE -m pip install -r ../requirements-test.txt
export COVERAGE_FILE=.coverage.$RANDOM LLVM_PROFILE_FILE=$PWD/pid-%p.profraw
$PYTHON_EXECUTABLE -m pytest ../tests --cov=.
# END B
@ -335,7 +335,7 @@ jobs:
PYTHONPATH: ${{ matrix.config }}
run: |
PYTHON_EXECUTABLE=$(grep 'Python_EXECUTABLE:' CMakeCache.txt | sed 's/.*=//')
$PYTHON_EXECUTABLE -m pip install pytest pytest-cov
$PYTHON_EXECUTABLE -m pip install -r ../requirements-test.txt
export COVERAGE_FILE=.coverage.$RANDOM LLVM_PROFILE_FILE=$PWD/pid-%p.profraw
$PYTHON_EXECUTABLE -m pytest ../tests --cov=.
# END B
@ -398,7 +398,7 @@ jobs:
- name: Test Python 3.11
shell: bash
run: |
python -m pip install pytest
python -m pip install -r requirements-test.txt
PYTHONPATH=built LD_LIBRARY_PATH=built/lib DYLD_LIBRARY_PATH=built/lib python -m pytest
- name: Set up Python 3.10
@ -412,7 +412,7 @@ jobs:
- name: Test Python 3.10
shell: bash
run: |
python -m pip install pytest
python -m pip install -r requirements-test.txt
PYTHONPATH=built LD_LIBRARY_PATH=built/lib DYLD_LIBRARY_PATH=built/lib python -m pytest
- name: Set up Python 3.9
@ -426,7 +426,7 @@ jobs:
- name: Test Python 3.9
shell: bash
run: |
python -m pip install pytest
python -m pip install -r requirements-test.txt
PYTHONPATH=built LD_LIBRARY_PATH=built/lib DYLD_LIBRARY_PATH=built/lib python -m pytest
- name: Set up Python 3.8
@ -440,7 +440,7 @@ jobs:
- name: Test Python 3.8
shell: bash
run: |
python -m pip install pytest
python -m pip install -r requirements-test.txt
PYTHONPATH=built LD_LIBRARY_PATH=built/lib DYLD_LIBRARY_PATH=built/lib python -m pytest
- name: Set up Python 3.7
@ -454,7 +454,7 @@ jobs:
- name: Test Python 3.7
shell: bash
run: |
python -m pip install pytest
python -m pip install -r requirements-test.txt
PYTHONPATH=built LD_LIBRARY_PATH=built/lib DYLD_LIBRARY_PATH=built/lib python -m pytest
- name: Make installer

38
direct/src/distributed/CRDataCache.py
View File

@ -76,41 +76,3 @@ if __debug__:
def flush(self):
CachedDOData.flush(self)
self._flushed = True
dc = CRDataCache()
dc._startMemLeakCheck()
cd = CachedDOData()
cd.foo = 34
dc.setCachedData(1, 'testCachedData', cd)
del cd
cd = CachedDOData()
cd.bar = 45
dc.setCachedData(1, 'testCachedData2', cd)
del cd
assert dc.hasCachedData(1)
assert dc.hasCachedData(1)
assert not dc.hasCachedData(2)
# data is dict of dataName->data
data = dc.popCachedData(1)
assert len(data) == 2
assert 'testCachedData' in data
assert 'testCachedData2' in data
assert data['testCachedData'].foo == 34
assert data['testCachedData2'].bar == 45
for cd in data.values():
cd.flush()
del data
dc._checkMemLeaks()
cd = CachedDOData()
cd.bar = 1234
dc.setCachedData(43, 'testCachedData2', cd)
del cd
assert dc.hasCachedData(43)
dc.flush()
dc._checkMemLeaks()
dc._stopMemLeakCheck()
dc.destroy()
del dc

155
direct/src/fsm/StatePush.py
View File

@ -46,11 +46,6 @@ class PushesStateChanges:
for subscriber in self._subscribers:
subscriber._recvStatePush(self)
if __debug__:
psc = PushesStateChanges(0)
assert psc.getState() == 0
psc.destroy()
del psc
class ReceivesStateChanges:
# base class for objects that subscribe to state changes from PushesStateChanges objects
@ -83,10 +78,6 @@ class ReceivesStateChanges:
def _recvStatePush(self, source):
pass
if __debug__:
rsc = ReceivesStateChanges(None)
rsc.destroy()
del rsc
class StateVar(PushesStateChanges):
# coder-friendly object that allows values to be set on it and pushes those values
@ -97,13 +88,6 @@ class StateVar(PushesStateChanges):
def get(self):
return PushesStateChanges.getState(self)
if __debug__:
sv = StateVar(0)
assert sv.get() == 0
sv.set(1)
assert sv.get() == 1
sv.destroy()
del sv
class StateChangeNode(PushesStateChanges, ReceivesStateChanges):
# base class that can be used to create a state-change notification chain
@ -120,31 +104,6 @@ class StateChangeNode(PushesStateChanges, ReceivesStateChanges):
# got a state push, apply new state to self
self._handlePotentialStateChange(source._value)
if __debug__:
sv = StateVar(0)
assert sv.get() == 0
scn = StateChangeNode(sv)
assert scn.getState() == 0
sv.set(1)
assert sv.get() == 1
assert scn.getState() == 1
scn2 = StateChangeNode(scn)
assert scn2.getState() == 1
sv.set(2)
assert scn2.getState() == 2
scn3 = StateChangeNode(scn)
assert scn3.getState() == 2
sv.set(3)
assert scn2.getState() == 3
assert scn3.getState() == 3
scn3.destroy()
scn2.destroy()
scn.destroy()
sv.destroy()
del scn3
del scn2
del scn
del sv
class ReceivesMultipleStateChanges:
# base class for objects that subscribe to state changes from multiple PushesStateChanges
@ -179,15 +138,6 @@ class ReceivesMultipleStateChanges:
def _recvMultiStatePush(self, key, source):
pass
if __debug__:
rsc = ReceivesMultipleStateChanges()
sv = StateVar(0)
sv2 = StateVar('b')
rsc._subscribeTo(sv, 'a')
rsc._subscribeTo(sv2, 2)
rsc._unsubscribe('a')
rsc.destroy()
del rsc
class FunctionCall(ReceivesMultipleStateChanges, PushesStateChanges):
# calls func with provided args whenever arguments' state changes
@ -249,47 +199,6 @@ class FunctionCall(ReceivesMultipleStateChanges, PushesStateChanges):
self._func(*self._bakedArgs, **self._bakedKargs)
PushesStateChanges._handleStateChange(self)
if __debug__:
l = []
def handler1(value, l=l):
l.append(value)
assert not l
sv = StateVar(0)
fc = FunctionCall(handler1, sv)
assert not l
fc.pushCurrentState()
assert l == [0,]
sv.set(1)
assert l == [0,1,]
sv.set(2)
assert l == [0,1,2,]
fc.destroy()
sv.destroy()
del fc
del sv
del handler1
del l
l = []
def handler2(value, kDummy=None, kValue=None, l=l):
l.append((value, kValue))
assert not l
sv = StateVar(0)
ksv = StateVar('a')
fc = FunctionCall(handler2, sv, kValue=ksv)
assert not l
fc.pushCurrentState()
assert l == [(0,'a',),]
sv.set(1)
assert l == [(0,'a'),(1,'a'),]
ksv.set('b')
assert l == [(0,'a'),(1,'a'),(1,'b'),]
fc.destroy()
sv.destroy()
del fc
del sv
del handler2
del l
class EnterExit(StateChangeNode):
# call enterFunc when our state becomes true, exitFunc when it becomes false
@ -314,33 +223,6 @@ class EnterExit(StateChangeNode):
self._exitFunc()
StateChangeNode._handleStateChange(self)
if __debug__:
l = []
def enter(l=l):
l.append(1)
def exit(l=l):
l.append(0)
sv = StateVar(0)
ee = EnterExit(sv, enter, exit)
sv.set(0)
assert not l
sv.set(1)
assert l == [1,]
sv.set(2)
assert l == [1,]
sv.set(0)
assert l == [1,0,]
sv.set(True)
assert l == [1,0,1,]
sv.set(False)
assert l == [1,0,1,0,]
ee.destroy()
sv.destroy()
del ee
del sv
del enter
del exit
del l
class Pulse(PushesStateChanges):
# changes state to True then immediately to False whenever sendPulse is called
@ -351,25 +233,6 @@ class Pulse(PushesStateChanges):
self._handlePotentialStateChange(True)
self._handlePotentialStateChange(False)
if __debug__:
l = []
def handler(value, l=l):
l.append(value)
p = Pulse()
fc = FunctionCall(handler, p)
assert not l
fc.pushCurrentState()
assert l == [False, ]
p.sendPulse()
assert l == [False, True, False, ]
p.sendPulse()
assert l == [False, True, False, True, False, ]
fc.destroy()
p.destroy()
del fc
del p
del l
del handler
class EventPulse(Pulse, DirectObject):
# sends a True-False "pulse" whenever a specific messenger message is sent
@ -381,6 +244,7 @@ class EventPulse(Pulse, DirectObject):
self.ignoreAll()
Pulse.destroy(self)
class EventArgument(PushesStateChanges, DirectObject):
# tracks a particular argument to a particular messenger event
def __init__(self, event, index=0):
@ -396,6 +260,7 @@ class EventArgument(PushesStateChanges, DirectObject):
def _handleEvent(self, *args):
self._handlePotentialStateChange(args[self._index])
class AttrSetter(StateChangeNode):
def __init__(self, source, object, attrName):
self._object = object
@ -406,19 +271,3 @@ class AttrSetter(StateChangeNode):
def _handleStateChange(self):
setattr(self._object, self._attrName, self._value)
StateChangeNode._handleStateChange(self)
if __debug__:
from direct.showbase.PythonUtil import ScratchPad
o = ScratchPad()
svar = StateVar(0)
aset = AttrSetter(svar, o, 'testAttr')
assert hasattr(o, 'testAttr')
assert o.testAttr == 0
svar.set('red')
assert o.testAttr == 'red'
aset.destroy()
svar.destroy()
o.destroy()
del aset
del svar
del o

0
direct/src/physics/__init__.py
View File

144
direct/src/showbase/CountedResource.py
View File

@ -81,147 +81,3 @@ class CountedResource(object):
def __del__(self):
self.decrementCounter()
if __debug__ and __name__ == '__main__':
class MouseResource(CountedResource):
"""
A simple class to demonstrate the acquisition of a resource.
"""
@classmethod
def acquire(cls):
# The call to the super-class's acquire() is
# not necessary at the moment, but may be in
# the future, so do it now for good measure.
super(MouseResource, cls).acquire()
# Now acquire the resource this class is
# managing.
print('-- Acquire Mouse')
@classmethod
def release(cls):
# First, release the resource this class is
# managing.
print('-- Release Mouse')
# The call to the super-class's release() is
# not necessary at the moment, but may be in
# the future, so do it now for good measure.
super(MouseResource, cls).release()
def __init__(self):
super(MouseResource, self).__init__()
def __del__(self):
super(MouseResource, self).__del__()
class CursorResource(CountedResource):
"""
A class to demonstrate how to implement a dependent
resource. Notice how this class also inherits from
CountedResource. Instead of subclassing MouseCounter,
we will just acquire it in our __init__() and release
it in our __del__().
"""
@classmethod
def acquire(cls):
super(CursorResource, cls).acquire()
print('-- Acquire Cursor')
@classmethod
def release(cls):
print('-- Release Cursor')
super(CursorResource, cls).release()
def __init__(self):
# The required resource references should
# be stored on 'self' since we want to
# release it when the object is deleted.
self.__mouseResource = MouseResource()
# Call the super-classes __init__()
# after all required resources are
# referenced.
super(CursorResource, self).__init__()
def __del__(self):
# Free up the most dependent resource
# first, the one this class is managing.
super(CursorResource, self).__del__()
# Now unlink any required resources.
del self.__mouseResource
class InvalidResource(MouseResource):
@classmethod
def acquire(cls):
super(InvalidResource, cls).acquire()
print('-- Acquire Invalid')
@classmethod
def release(cls):
print('-- Release Invalid')
super(InvalidResource, cls).release()
print('\nAllocate Mouse')
m = MouseResource()
print('Free up Mouse')
del m
print('\nAllocate Cursor')
c = CursorResource()
print('Free up Cursor')
del c
print('\nAllocate Mouse then Cursor')
m = MouseResource()
c = CursorResource()
print('Free up Cursor')
del c
print('Free up Mouse')
del m
print('\nAllocate Mouse then Cursor')
m = MouseResource()
c = CursorResource()
print('Free up Mouse')
del m
print('Free up Cursor')
del c
print('\nAllocate Cursor then Mouse')
c = CursorResource()
m = MouseResource()
print('Free up Mouse')
del m
print('Free up Cursor')
del c
print('\nAllocate Cursor then Mouse')
c = CursorResource()
m = MouseResource()
print('Free up Cursor')
del c
# example of an invalid subclass
try:
print('\nAllocate Invalid')
i = InvalidResource()
print('Free up Invalid')
except AssertionError as e:
print(e)
print('')
print('Free up Mouse')
del m
def demoFunc():
print('\nAllocate Cursor within function')
c = CursorResource()
print('Cursor will be freed on function exit')
demoFunc()

28
direct/src/showbase/DistancePhasedNode.py
View File

@ -327,31 +327,3 @@ class BufferedDistancePhasedNode(DistancePhasedNode):
for x,sphere in enumerate(self._colSpheres[phase+1:]):
sphere.node().modifySolid(0).setRadius(self.bufferParamList[x+phase+1][1][0])
sphere.node().markInternalBoundsStale()
if __debug__ and 0:
cSphere = CollisionSphere(0, 0, 0, 0.1)
cNode = CollisionNode('camCol')
cNode.addSolid(cSphere)
cNodePath = NodePath(cNode)
cNodePath.reparentTo(base.cam)
# cNodePath.show()
# cNodePath.setPos(25,0,0)
base.cTrav = CollisionTraverser()
eventHandler = CollisionHandlerEvent()
eventHandler.addInPattern('enter%in')
eventHandler.addOutPattern('exit%in')
# messenger.toggleVerbose()
base.cTrav.addCollider(cNodePath, eventHandler)
p = BufferedDistancePhasedNode('p', {'At': (10, 20), 'Near': (100, 200), 'Far': (1000, 1020)},
autoCleanup=False,
fromCollideNode=cNodePath,
)
p.reparentTo(render)
p._DistancePhasedNode__oneTimeCollide()
base.eventMgr.doEvents()

47
direct/src/showbase/PythonUtil.py
View File

@ -2458,18 +2458,6 @@ def formatTimeCompact(seconds):
return result
if __debug__ and __name__ == '__main__':
ftc = formatTimeCompact
assert ftc(0) == '0s'
assert ftc(1) == '1s'
assert ftc(60) == '1m0s'
assert ftc(64) == '1m4s'
assert ftc(60*60) == '1h0m0s'
assert ftc(24*60*60) == '1d0h0m0s'
assert ftc(24*60*60 + 2*60*60 + 34*60 + 12) == '1d2h34m12s'
del ftc
def formatTimeExact(seconds):
# like formatTimeCompact but leaves off '0 seconds', '0 minutes' etc. for
# times that are e.g. 1 hour, 3 days etc.
@ -2496,19 +2484,6 @@ def formatTimeExact(seconds):
return result
if __debug__ and __name__ == '__main__':
fte = formatTimeExact
assert fte(0) == '0s'
assert fte(1) == '1s'
assert fte(2) == '2s'
assert fte(61) == '1m1s'
assert fte(60) == '1m'
assert fte(60*60) == '1h'
assert fte(24*60*60) == '1d'
assert fte((24*60*60) + (2 * 60)) == '1d0h2m'
del fte
class AlphabetCounter:
# object that produces 'A', 'B', 'C', ... 'AA', 'AB', etc.
def __init__(self):
@ -2540,28 +2515,6 @@ class AlphabetCounter:
__next__ = next
if __debug__ and __name__ == '__main__':
def testAlphabetCounter():
tempList = []
ac = AlphabetCounter()
for i in range(26*3):
tempList.append(ac.next())
assert tempList == [ 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z',
'AA','AB','AC','AD','AE','AF','AG','AH','AI','AJ','AK','AL','AM','AN','AO','AP','AQ','AR','AS','AT','AU','AV','AW','AX','AY','AZ',
'BA','BB','BC','BD','BE','BF','BG','BH','BI','BJ','BK','BL','BM','BN','BO','BP','BQ','BR','BS','BT','BU','BV','BW','BX','BY','BZ',]
ac = AlphabetCounter()
num = 26 # A-Z
num += (26*26) # AA-ZZ
num += 26 # AAZ
num += 1 # ABA
num += 2 # ABC
for i in range(num):
x = ac.next()
assert x == 'ABC'
testAlphabetCounter()
del testAlphabetCounter
class Default:
# represents 'use the default value'
# useful for keyword arguments to virtual methods

105
direct/src/stdpy/threading.py
View File

@ -435,108 +435,3 @@ def setprofile(func):
def stack_size(size = None):
raise ThreadError
if __debug__:
def _test():
from collections import deque
_sleep = core.Thread.sleep
_VERBOSE = False
class _Verbose(object):
def __init__(self, verbose=None):
if verbose is None:
verbose = _VERBOSE
self.__verbose = verbose
def _note(self, format, *args):
if self.__verbose:
format = format % args
format = "%s: %s\n" % (
currentThread().getName(), format)
_sys.stderr.write(format)
class BoundedQueue(_Verbose):
def __init__(self, limit):
_Verbose.__init__(self)
self.mon = Lock(name = "BoundedQueue.mon")
self.rc = Condition(self.mon)
self.wc = Condition(self.mon)
self.limit = limit
self.queue = deque()
def put(self, item):
self.mon.acquire()
while len(self.queue) >= self.limit:
self._note("put(%s): queue full", item)
self.wc.wait()
self.queue.append(item)
self._note("put(%s): appended, length now %d",
item, len(self.queue))
self.rc.notify()
self.mon.release()
def get(self):
self.mon.acquire()
while not self.queue:
self._note("get(): queue empty")
self.rc.wait()
item = self.queue.popleft()
self._note("get(): got %s, %d left", item, len(self.queue))
self.wc.notify()
self.mon.release()
return item
class ProducerThread(Thread):
def __init__(self, queue, quota):
Thread.__init__(self, name="Producer")
self.queue = queue
self.quota = quota
def run(self):
from random import random
counter = 0
while counter < self.quota:
counter = counter + 1
self.queue.put("%s.%d" % (self.getName(), counter))
_sleep(random() * 0.00001)
class ConsumerThread(Thread):
def __init__(self, queue, count):
Thread.__init__(self, name="Consumer")
self.queue = queue
self.count = count
def run(self):
while self.count > 0:
item = self.queue.get()
print(item)
self.count = self.count - 1
NP = 3
QL = 4
NI = 5
Q = BoundedQueue(QL)
P = []
for i in range(NP):
t = ProducerThread(Q, NI)
t.setName("Producer-%d" % (i+1))
P.append(t)
C = ConsumerThread(Q, NI*NP)
for t in P:
t.start()
_sleep(0.000001)
C.start()
for t in P:
t.join()
C.join()
if __name__ == '__main__':
_test()

89
direct/src/stdpy/threading2.py
View File

@ -762,92 +762,3 @@ def main_thread():
## from thread import _local as local
## except ImportError:
## from _threading_local import local
# Self-test code
if __debug__:
def _test():
from collections import deque
class BoundedQueue(_Verbose):
def __init__(self, limit):
_Verbose.__init__(self)
self.mon = RLock()
self.rc = Condition(self.mon)
self.wc = Condition(self.mon)
self.limit = limit
self.queue = deque()
def put(self, item):
self.mon.acquire()
while len(self.queue) >= self.limit:
self._note("put(%s): queue full", item)
self.wc.wait()
self.queue.append(item)
self._note("put(%s): appended, length now %d",
item, len(self.queue))
self.rc.notify()
self.mon.release()
def get(self):
self.mon.acquire()
while not self.queue:
self._note("get(): queue empty")
self.rc.wait()
item = self.queue.popleft()
self._note("get(): got %s, %d left", item, len(self.queue))
self.wc.notify()
self.mon.release()
return item
class ProducerThread(Thread):
def __init__(self, queue, quota):
Thread.__init__(self, name="Producer")
self.queue = queue
self.quota = quota
def run(self):
from random import random
counter = 0
while counter < self.quota:
counter = counter + 1
self.queue.put("%s.%d" % (self.getName(), counter))
_sleep(random() * 0.00001)
class ConsumerThread(Thread):
def __init__(self, queue, count):
Thread.__init__(self, name="Consumer")
self.queue = queue
self.count = count
def run(self):
while self.count > 0:
item = self.queue.get()
print(item)
self.count = self.count - 1
NP = 3
QL = 4
NI = 5
Q = BoundedQueue(QL)
P = []
for i in range(NP):
t = ProducerThread(Q, NI)
t.setName("Producer-%d" % (i+1))
P.append(t)
C = ConsumerThread(Q, NI*NP)
for t in P:
t.start()
_sleep(0.000001)
C.start()
for t in P:
t.join()
C.join()
if __name__ == '__main__':
_test()

526
direct/src/task/Task.py
View File

@ -801,532 +801,6 @@ class TaskManager:
# time.sleep(delta)
# delta = minFinTime - self.globalClock.getRealTime()
if __debug__:
# to catch memory leaks during the tests at the bottom of the file
def _startTrackingMemLeaks(self):
pass
def _stopTrackingMemLeaks(self):
pass
def _checkMemLeaks(self):
pass
def _runTests(self):
if __debug__:
tm = TaskManager()
tm.setClock(ClockObject())
tm.setupTaskChain("default", tickClock = True)
# check for memory leaks after every test
tm._startTrackingMemLeaks()
tm._checkMemLeaks()
# run-once task
l = []
def _testDone(task, l=l):
l.append(None)
return task.done
tm.add(_testDone, 'testDone')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 1
_testDone = None
tm._checkMemLeaks()
# remove by name
def _testRemoveByName(task):
return task.done
tm.add(_testRemoveByName, 'testRemoveByName')
assert tm.remove('testRemoveByName') == 1
assert tm.remove('testRemoveByName') == 0
_testRemoveByName = None
tm._checkMemLeaks()
# duplicate named tasks
def _testDupNamedTasks(task):
return task.done
tm.add(_testDupNamedTasks, 'testDupNamedTasks')
tm.add(_testDupNamedTasks, 'testDupNamedTasks')
assert tm.remove('testRemoveByName') == 0
_testDupNamedTasks = None
tm._checkMemLeaks()
# continued task
l = []
def _testCont(task, l = l):
l.append(None)
return task.cont
tm.add(_testCont, 'testCont')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 2
tm.remove('testCont')
_testCont = None
tm._checkMemLeaks()
# continue until done task
l = []
def _testContDone(task, l = l):
l.append(None)
if len(l) >= 2:
return task.done
else:
return task.cont
tm.add(_testContDone, 'testContDone')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 2
tm.step()
assert len(l) == 2
assert not tm.hasTaskNamed('testContDone')
_testContDone = None
tm._checkMemLeaks()
# hasTaskNamed
def _testHasTaskNamed(task):
return task.done
tm.add(_testHasTaskNamed, 'testHasTaskNamed')
assert tm.hasTaskNamed('testHasTaskNamed')
tm.step()
assert not tm.hasTaskNamed('testHasTaskNamed')
_testHasTaskNamed = None
tm._checkMemLeaks()
# task sort
l = []
def _testPri1(task, l = l):
l.append(1)
return task.cont
def _testPri2(task, l = l):
l.append(2)
return task.cont
tm.add(_testPri1, 'testPri1', sort = 1)
tm.add(_testPri2, 'testPri2', sort = 2)
tm.step()
assert len(l) == 2
assert l == [1, 2,]
tm.step()
assert len(l) == 4
assert l == [1, 2, 1, 2,]
tm.remove('testPri1')
tm.remove('testPri2')
_testPri1 = None
_testPri2 = None
tm._checkMemLeaks()
# task extraArgs
l = []
def _testExtraArgs(arg1, arg2, l=l):
l.extend([arg1, arg2,])
return done
tm.add(_testExtraArgs, 'testExtraArgs', extraArgs=[4,5])
tm.step()
assert len(l) == 2
assert l == [4, 5,]
_testExtraArgs = None
tm._checkMemLeaks()
# task appendTask
l = []
def _testAppendTask(arg1, arg2, task, l=l):
l.extend([arg1, arg2,])
return task.done
tm.add(_testAppendTask, '_testAppendTask', extraArgs=[4,5], appendTask=True)
tm.step()
assert len(l) == 2
assert l == [4, 5,]
_testAppendTask = None
tm._checkMemLeaks()
# task uponDeath
l = []
def _uponDeathFunc(task, l=l):
l.append(task.name)
def _testUponDeath(task):
return done
tm.add(_testUponDeath, 'testUponDeath', uponDeath=_uponDeathFunc)
tm.step()
assert len(l) == 1
assert l == ['testUponDeath']
_testUponDeath = None
_uponDeathFunc = None
tm._checkMemLeaks()
# task owner
class _TaskOwner:
def _addTask(self, task):
self.addedTaskName = task.name
def _clearTask(self, task):
self.clearedTaskName = task.name
to = _TaskOwner()
l = []
def _testOwner(task):
return done
tm.add(_testOwner, 'testOwner', owner=to)
tm.step()
assert getattr(to, 'addedTaskName', None) == 'testOwner'
assert getattr(to, 'clearedTaskName', None) == 'testOwner'
_testOwner = None
del to
_TaskOwner = None
tm._checkMemLeaks()
doLaterTests = [0,]
# doLater
l = []
def _testDoLater1(task, l=l):
l.append(1)
def _testDoLater2(task, l=l):
l.append(2)
def _monitorDoLater(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .03:
assert l == [1, 2,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLater1, 'testDoLater1')
tm.doMethodLater(.02, _testDoLater2, 'testDoLater2')
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLater, 'monitorDoLater', sort=10)
_testDoLater1 = None
_testDoLater2 = None
_monitorDoLater = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater sort
l = []
def _testDoLaterPri1(task, l=l):
l.append(1)
def _testDoLaterPri2(task, l=l):
l.append(2)
def _monitorDoLaterPri(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .02:
assert l == [1, 2,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterPri1, 'testDoLaterPri1', sort=1)
tm.doMethodLater(.01, _testDoLaterPri2, 'testDoLaterPri2', sort=2)
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterPri, 'monitorDoLaterPri', sort=10)
_testDoLaterPri1 = None
_testDoLaterPri2 = None
_monitorDoLaterPri = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater extraArgs
l = []
def _testDoLaterExtraArgs(arg1, l=l):
l.append(arg1)
def _monitorDoLaterExtraArgs(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .02:
assert l == [3,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterExtraArgs, 'testDoLaterExtraArgs', extraArgs=[3,])
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterExtraArgs, 'monitorDoLaterExtraArgs', sort=10)
_testDoLaterExtraArgs = None
_monitorDoLaterExtraArgs = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater appendTask
l = []
def _testDoLaterAppendTask(arg1, task, l=l):
assert task.name == 'testDoLaterAppendTask'
l.append(arg1)
def _monitorDoLaterAppendTask(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .02:
assert l == [4,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterAppendTask, 'testDoLaterAppendTask',
extraArgs=[4,], appendTask=True)
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterAppendTask, 'monitorDoLaterAppendTask', sort=10)
_testDoLaterAppendTask = None
_monitorDoLaterAppendTask = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater uponDeath
l = []
def _testUponDeathFunc(task, l=l):
assert task.name == 'testDoLaterUponDeath'
l.append(10)
def _testDoLaterUponDeath(arg1, l=l):
return done
def _monitorDoLaterUponDeath(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .02:
assert l == [10,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterUponDeath, 'testDoLaterUponDeath',
uponDeath=_testUponDeathFunc)
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterUponDeath, 'monitorDoLaterUponDeath', sort=10)
_testUponDeathFunc = None
_testDoLaterUponDeath = None
_monitorDoLaterUponDeath = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater owner
class _DoLaterOwner:
def _addTask(self, task):
self.addedTaskName = task.name
def _clearTask(self, task):
self.clearedTaskName = task.name
doLaterOwner = _DoLaterOwner()
l = []
def _testDoLaterOwner(l=l):
pass
def _monitorDoLaterOwner(task, tm=tm, l=l, doLaterOwner=doLaterOwner,
doLaterTests=doLaterTests):
if task.time > .02:
assert getattr(doLaterOwner, 'addedTaskName', None) == 'testDoLaterOwner'
assert getattr(doLaterOwner, 'clearedTaskName', None) == 'testDoLaterOwner'
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterOwner, 'testDoLaterOwner',
owner=doLaterOwner)
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterOwner, 'monitorDoLaterOwner', sort=10)
_testDoLaterOwner = None
_monitorDoLaterOwner = None
del doLaterOwner
_DoLaterOwner = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# run the doLater tests
while doLaterTests[0] > 0:
tm.step()
del doLaterTests
tm._checkMemLeaks()
# getTasks
def _testGetTasks(task):
return task.cont
# No doLaterProcessor in the new world.
assert len(tm.getTasks()) == 0
tm.add(_testGetTasks, 'testGetTasks1')
assert len(tm.getTasks()) == 1
assert (tm.getTasks()[0].name == 'testGetTasks1' or
tm.getTasks()[1].name == 'testGetTasks1')
tm.add(_testGetTasks, 'testGetTasks2')
tm.add(_testGetTasks, 'testGetTasks3')
assert len(tm.getTasks()) == 3
tm.remove('testGetTasks2')
assert len(tm.getTasks()) == 2
tm.remove('testGetTasks1')
tm.remove('testGetTasks3')
assert len(tm.getTasks()) == 0
_testGetTasks = None
tm._checkMemLeaks()
# getDoLaters
def _testGetDoLaters():
pass
assert len(tm.getDoLaters()) == 0
tm.doMethodLater(.1, _testGetDoLaters, 'testDoLater1')
assert len(tm.getDoLaters()) == 1
assert tm.getDoLaters()[0].name == 'testDoLater1'
tm.doMethodLater(.1, _testGetDoLaters, 'testDoLater2')
tm.doMethodLater(.1, _testGetDoLaters, 'testDoLater3')
assert len(tm.getDoLaters()) == 3
tm.remove('testDoLater2')
assert len(tm.getDoLaters()) == 2
tm.remove('testDoLater1')
tm.remove('testDoLater3')
assert len(tm.getDoLaters()) == 0
_testGetDoLaters = None
tm._checkMemLeaks()
# duplicate named doLaters removed via taskMgr.remove
def _testDupNameDoLaters():
pass
# the doLaterProcessor is always running
tm.doMethodLater(.1, _testDupNameDoLaters, 'testDupNameDoLater')
tm.doMethodLater(.1, _testDupNameDoLaters, 'testDupNameDoLater')
assert len(tm.getDoLaters()) == 2
tm.remove('testDupNameDoLater')
assert len(tm.getDoLaters()) == 0
_testDupNameDoLaters = None
tm._checkMemLeaks()
# duplicate named doLaters removed via remove()
def _testDupNameDoLatersRemove():
pass
# the doLaterProcessor is always running
dl1 = tm.doMethodLater(.1, _testDupNameDoLatersRemove, 'testDupNameDoLaterRemove')
dl2 = tm.doMethodLater(.1, _testDupNameDoLatersRemove, 'testDupNameDoLaterRemove')
assert len(tm.getDoLaters()) == 2
dl2.remove()
assert len(tm.getDoLaters()) == 1
dl1.remove()
assert len(tm.getDoLaters()) == 0
_testDupNameDoLatersRemove = None
# nameDict etc. isn't cleared out right away with task.remove()
tm._checkMemLeaks()
# getTasksNamed
def _testGetTasksNamed(task):
return task.cont
assert len(tm.getTasksNamed('testGetTasksNamed')) == 0
tm.add(_testGetTasksNamed, 'testGetTasksNamed')
assert len(tm.getTasksNamed('testGetTasksNamed')) == 1
assert tm.getTasksNamed('testGetTasksNamed')[0].name == 'testGetTasksNamed'
tm.add(_testGetTasksNamed, 'testGetTasksNamed')
tm.add(_testGetTasksNamed, 'testGetTasksNamed')
assert len(tm.getTasksNamed('testGetTasksNamed')) == 3
tm.remove('testGetTasksNamed')
assert len(tm.getTasksNamed('testGetTasksNamed')) == 0
_testGetTasksNamed = None
tm._checkMemLeaks()
# removeTasksMatching
def _testRemoveTasksMatching(task):
return task.cont
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching')
assert len(tm.getTasksNamed('testRemoveTasksMatching')) == 1
tm.removeTasksMatching('testRemoveTasksMatching')
assert len(tm.getTasksNamed('testRemoveTasksMatching')) == 0
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching1')
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching2')
assert len(tm.getTasksNamed('testRemoveTasksMatching1')) == 1
assert len(tm.getTasksNamed('testRemoveTasksMatching2')) == 1
tm.removeTasksMatching('testRemoveTasksMatching*')
assert len(tm.getTasksNamed('testRemoveTasksMatching1')) == 0
assert len(tm.getTasksNamed('testRemoveTasksMatching2')) == 0
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching1a')
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching2a')
assert len(tm.getTasksNamed('testRemoveTasksMatching1a')) == 1
assert len(tm.getTasksNamed('testRemoveTasksMatching2a')) == 1
tm.removeTasksMatching('testRemoveTasksMatching?a')
assert len(tm.getTasksNamed('testRemoveTasksMatching1a')) == 0
assert len(tm.getTasksNamed('testRemoveTasksMatching2a')) == 0
_testRemoveTasksMatching = None
tm._checkMemLeaks()
# create Task object and add to mgr
l = []
def _testTaskObj(task, l=l):
l.append(None)
return task.cont
t = Task(_testTaskObj)
tm.add(t, 'testTaskObj')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 2
tm.remove('testTaskObj')
tm.step()
assert len(l) == 2
_testTaskObj = None
tm._checkMemLeaks()
# remove Task via task.remove()
l = []
def _testTaskObjRemove(task, l=l):
l.append(None)
return task.cont
t = Task(_testTaskObjRemove)
tm.add(t, 'testTaskObjRemove')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 2
t.remove()
tm.step()
assert len(l) == 2
del t
_testTaskObjRemove = None
tm._checkMemLeaks()
# this test fails, and it's not clear what the correct behavior should be.
# sort passed to Task.__init__ is always overridden by taskMgr.add()
# even if no sort is specified, and calling Task.setSort() has no
# effect on the taskMgr's behavior.
# set/get Task sort
#l = []
#def _testTaskObjSort(arg, task, l=l):
# l.append(arg)
# return task.cont
#t1 = Task(_testTaskObjSort, sort=1)
#t2 = Task(_testTaskObjSort, sort=2)
#tm.add(t1, 'testTaskObjSort1', extraArgs=['a',], appendTask=True)
#tm.add(t2, 'testTaskObjSort2', extraArgs=['b',], appendTask=True)
#tm.step()
#assert len(l) == 2
#assert l == ['a', 'b']
#assert t1.getSort() == 1
#assert t2.getSort() == 2
#t1.setSort(3)
#assert t1.getSort() == 3
#tm.step()
#assert len(l) == 4
#assert l == ['a', 'b', 'b', 'a',]
#t1.remove()
#t2.remove()
#tm.step()
#assert len(l) == 4
#del t1
#del t2
#_testTaskObjSort = None
#tm._checkMemLeaks()
del l
tm.destroy()
del tm
if __debug__:
def checkLeak():

18
direct/src/tkpanels/ParticlePanel.py
View File

@ -2909,21 +2909,3 @@ class ParticlePanel(AppShell):
min = 0.01,
value = force.getRadius())
self.createForceActiveWidget(frame, pageName, forceName, force)
######################################################################
# Create demo in root window for testing.
if __name__ == '__main__':
try:
base
except NameError:
from direct.showbase.ShowBase import ShowBase
base = ShowBase()
root = Pmw.initialise()
pp = ParticlePanel()
base.pp=pp
#ve = VectorEntry(Toplevel(), relief = GROOVE)
#ve.pack()
base.run()

8
direct/src/tkpanels/Placer.py
View File

@ -786,11 +786,3 @@ class Placer(AppShell):
def place(nodePath):
return Placer(nodePath = nodePath)
######################################################################
# Create demo in root window for testing.
if __name__ == '__main__':
root = Pmw.initialise()
widget = Placer()

4
direct/src/tkwidgets/AppShell.py
View File

@ -550,7 +550,3 @@ class TestAppShell(AppShell):
def createInterface(self):
self.createButtons()
self.createMain()
if __name__ == '__main__':
test = TestAppShell(balloon_state='none')

15
direct/src/tkwidgets/Dial.py
View File

@ -422,18 +422,3 @@ class DialWidget(Pmw.MegaWidget):
""" User redefinable callback executed on button release """
if self['postCallback']:
self['postCallback'](*self['callbackData'])
if __name__ == '__main__':
tl = tk.Toplevel()
d = Dial(tl)
d2 = Dial(tl, dial_numSegments = 12, max = 360,
dial_fRollover = 0, value = 180)
d3 = Dial(tl, dial_numSegments = 12, max = 90, min = -90,
dial_fRollover = 0)
d4 = Dial(tl, dial_numSegments = 16, max = 256,
dial_fRollover = 0)
d.pack(expand = 1, fill = tk.X)
d2.pack(expand = 1, fill = tk.X)
d3.pack(expand = 1, fill = tk.X)
d4.pack(expand = 1, fill = tk.X)

42
direct/src/tkwidgets/EntryScale.py
View File

@ -517,45 +517,3 @@ def rgbPanel(nodePath, callback = None):
messenger.send('RGBPanel_setColor', [nodePath, r, g, b, a])
esg['postCallback'] = onRelease
return esg
## SAMPLE CODE
if __name__ == '__main__':
# Initialise Tkinter and Pmw.
root = tk.Toplevel()
root.title('Pmw EntryScale demonstration')
# Dummy command
def printVal(val):
print(val)
# Create and pack a EntryScale megawidget.
mega1 = EntryScale(root, command = printVal)
mega1.pack(side = 'left', expand = 1, fill = 'x')
# These are things you can set/configure
# Starting value for entryScale
#mega1['value'] = 123.456
#mega1['text'] = 'Drive delta X'
#mega1['min'] = 0.0
#mega1['max'] = 1000.0
#mega1['resolution'] = 1.0
# To change the color of the label:
#mega1.label['foreground'] = 'Red'
# Max change/update, default is 100
# To have really fine control, for example
# mega1['maxVelocity'] = 0.1
# Number of digits to the right of the decimal point, default = 2
# mega1['numDigits'] = 5
# To create a entryScale group to set an RGBA value:
group1 = EntryScaleGroup(root, dim = 4,
title = 'Simple RGBA Panel',
labels = ('R', 'G', 'B', 'A'),
Valuator_min = 0.0,
Valuator_max = 255.0,
Valuator_resolution = 1.0,
command = printVal)
# Uncomment this if you aren't running in IDLE
#root.mainloop()

39
direct/src/tkwidgets/Floater.py
View File

@ -329,42 +329,3 @@ class FloaterGroup(Pmw.MegaToplevel):
def reset(self):
self.set(self['value'])
## SAMPLE CODE
if __name__ == '__main__':
# Initialise Tkinter and Pmw.
root = tk.Toplevel()
root.title('Pmw Floater demonstration')
# Dummy command
def printVal(val):
print(val)
# Create and pack a Floater megawidget.
mega1 = Floater(root, command = printVal)
mega1.pack(side = 'left', expand = 1, fill = 'x')
# These are things you can set/configure
# Starting value for floater
#mega1['value'] = 123.456
#mega1['text'] = 'Drive delta X'
# To change the color of the label:
#mega1.label['foreground'] = 'Red'
# Max change/update, default is 100
# To have really fine control, for example
#mega1['maxVelocity'] = 0.1
# Number of digits to the right of the decimal point, default = 2
#mega1['numDigits'] = 5
# To create a floater group to set an RGBA value:
group1 = FloaterGroup(root, dim = 4,
title = 'Simple RGBA Panel',
labels = ('R', 'G', 'B', 'A'),
Valuator_min = 0.0,
Valuator_max = 255.0,
Valuator_resolution = 1.0,
command = printVal)
# Uncomment this if you aren't running in IDLE
#root.mainloop()

14
direct/src/tkwidgets/VectorWidgets.py
View File

@ -329,17 +329,3 @@ class ColorEntry(VectorEntry):
initialcolor = tuple(self.get()[:3]))[0]
if color:
self.set((color[0], color[1], color[2], self.getAt(3)))
if __name__ == '__main__':
root = tk.Toplevel()
root.title('Vector Widget demo')
ve = VectorEntry(root)
ve.pack()
v3e = Vector3Entry(root)
v3e.pack()
v4e = Vector4Entry(root)
v4e.pack()
ce = ColorEntry(root)
ce.pack()

4
requirements-test.txt Normal file
View File

@ -0,0 +1,4 @@
pytest
pytest-cov
Pmw-py3==2.1

23
tests/display/conftest.py → tests/conftest.py
View File

@ -1,4 +1,27 @@
import sys
import tkinter as tk
import pytest
from panda3d import core
from direct.showbase.ShowBase import ShowBase
@pytest.fixture
def base():
base = ShowBase(windowType='none')
yield base
base.destroy()
@pytest.fixture
def tk_toplevel():
if sys.platform == 'darwin' and not core.ConfigVariableBool('want-tk', False):
pytest.skip('"want-tk" must be true to use tkinter with Panda3D on macOS')
try:
root = tk.Toplevel()
except tk.TclError as e:
pytest.skip(str(e))
yield root
root.destroy()
@pytest.fixture(scope='session')

42
tests/distributed/test_CRDataCache.py Normal file
View File

@ -0,0 +1,42 @@
from direct.distributed.CachedDOData import CachedDOData
from direct.distributed.CRDataCache import CRDataCache
def test_CRDataCache():
dc = CRDataCache()
dc._startMemLeakCheck()
cd = CachedDOData()
cd.foo = 34
dc.setCachedData(1, 'testCachedData', cd)
del cd
cd = CachedDOData()
cd.bar = 45
dc.setCachedData(1, 'testCachedData2', cd)
del cd
assert dc.hasCachedData(1)
assert dc.hasCachedData(1)
assert not dc.hasCachedData(2)
# data is dict of dataName->data
data = dc.popCachedData(1)
assert len(data) == 2
assert 'testCachedData' in data
assert 'testCachedData2' in data
assert data['testCachedData'].foo == 34
assert data['testCachedData2'].bar == 45
for cd in data.values():
cd.flush()
del data
dc._checkMemLeaks()
cd = CachedDOData()
cd.bar = 1234
dc.setCachedData(43, 'testCachedData2', cd)
del cd
assert dc.hasCachedData(43)
dc.flush()
dc._checkMemLeaks()
dc._stopMemLeakCheck()
dc.destroy()
del dc

144
tests/fsm/test_StatePush.py Normal file
View File

@ -0,0 +1,144 @@
from direct.fsm import StatePush
def test_PushesStateChanges():
psc = StatePush.PushesStateChanges(0)
assert psc.getState() == 0
psc.destroy()
def test_ReceivesStateChanges():
rsc = StatePush.ReceivesStateChanges(None)
rsc.destroy()
def test_StateVar():
sv = StatePush.StateVar(0)
assert sv.get() == 0
sv.set(1)
assert sv.get() == 1
sv.destroy()
def test_StateChangeNode():
sv = StatePush.StateVar(0)
assert sv.get() == 0
scn = StatePush.StateChangeNode(sv)
assert scn.getState() == 0
sv.set(1)
assert sv.get() == 1
assert scn.getState() == 1
scn2 = StatePush.StateChangeNode(scn)
assert scn2.getState() == 1
sv.set(2)
assert scn2.getState() == 2
scn3 = StatePush.StateChangeNode(scn)
assert scn3.getState() == 2
sv.set(3)
assert scn2.getState() == 3
assert scn3.getState() == 3
scn3.destroy()
scn2.destroy()
scn.destroy()
sv.destroy()
def test_ReceivesMultipleStateChanges():
rsc = StatePush.ReceivesMultipleStateChanges()
sv = StatePush.StateVar(0)
sv2 = StatePush.StateVar('b')
rsc._subscribeTo(sv, 'a')
rsc._subscribeTo(sv2, 2)
rsc._unsubscribe('a')
rsc.destroy()
def test_FunctionCall_1():
l = []
def handler1(value, l=l):
l.append(value)
assert not l
sv = StatePush.StateVar(0)
fc = StatePush.FunctionCall(handler1, sv)
assert not l
fc.pushCurrentState()
assert l == [0,]
sv.set(1)
assert l == [0,1,]
sv.set(2)
assert l == [0,1,2,]
fc.destroy()
sv.destroy()
def test_FunctionCall_2():
l = []
def handler2(value, kDummy=None, kValue=None, l=l):
l.append((value, kValue))
assert not l
sv = StatePush.StateVar(0)
ksv = StatePush.StateVar('a')
fc = StatePush.FunctionCall(handler2, sv, kValue=ksv)
assert not l
fc.pushCurrentState()
assert l == [(0,'a',),]
sv.set(1)
assert l == [(0,'a'),(1,'a'),]
ksv.set('b')
assert l == [(0,'a'),(1,'a'),(1,'b'),]
fc.destroy()
sv.destroy()
def test_EnterExit():
l = []
def enter(l=l):
l.append(1)
def exit(l=l):
l.append(0)
sv = StatePush.StateVar(0)
ee = StatePush.EnterExit(sv, enter, exit)
sv.set(0)
assert not l
sv.set(1)
assert l == [1,]
sv.set(2)
assert l == [1,]
sv.set(0)
assert l == [1,0,]
sv.set(True)
assert l == [1,0,1,]
sv.set(False)
assert l == [1,0,1,0,]
ee.destroy()
sv.destroy()
def test_Pulse():
l = []
def handler(value, l=l):
l.append(value)
p = StatePush.Pulse()
fc = StatePush.FunctionCall(handler, p)
assert not l
fc.pushCurrentState()
assert l == [False, ]
p.sendPulse()
assert l == [False, True, False, ]
p.sendPulse()
assert l == [False, True, False, True, False, ]
fc.destroy()
p.destroy()
def test_AttrSetter():
from types import SimpleNamespace
o = SimpleNamespace()
svar = StatePush.StateVar(0)
aset = StatePush.AttrSetter(svar, o, 'testAttr')
assert hasattr(o, 'testAttr')
assert o.testAttr == 0
svar.set('red')
assert o.testAttr == 'red'
aset.destroy()
svar.destroy()

22
direct/src/gui/DirectGuiTest.py → tests/gui/test_DirectGui.py
View File

@ -1,19 +1,11 @@
"""Undocumented Module"""
__all__ = []
from random import random
from panda3d.core import Point3
from direct.task.TaskManagerGlobal import taskMgr
from direct.gui import DirectGuiGlobals
from direct.gui.DirectGui import DirectButton, DirectDialog, DirectEntry, DirectFrame, YesNoDialog
if __name__ == "__main__":
from panda3d.core import Point3
from direct.showbase.ShowBase import ShowBase
from direct.task.TaskManagerGlobal import taskMgr
from . import DirectGuiGlobals
from .DirectGui import DirectButton, DirectDialog, DirectEntry, DirectFrame, YesNoDialog
#from whrandom import *
from random import random
base = ShowBase()
def test_DirectGui(base):
# EXAMPLE CODE
# Load a model
smiley = base.loader.loadModel('models/misc/smiley')
@ -142,5 +134,3 @@ if __name__ == "__main__":
print('WIDTH: %s' % de1.getWidth())
print('HEIGHT: %s' % de1.getHeight())
print('CENTER: %s' % (de1.getCenter(),))
base.run()

20
direct/src/particles/ParticleTest.py → tests/particles/test_particle.py
View File

@ -1,15 +1,11 @@
from panda3d.core import Vec3
from panda3d.physics import LinearVectorForce
from direct.particles import ParticleEffect
from direct.particles import Particles
from direct.particles import ForceGroup
if __name__ == "__main__":
from panda3d.core import Vec3
from panda3d.physics import LinearVectorForce
from direct.directbase.TestStart import base
from direct.tkpanels import ParticlePanel
from . import ParticleEffect
from . import Particles
from . import ForceGroup
def test_particle(base):
# Showbase
base.enableParticles()
@ -29,6 +25,4 @@ if __name__ == "__main__":
pe.addParticles(p)
# Particle Panel
pp = ParticlePanel.ParticlePanel(pe)
base.run()
# pp = ParticlePanel.ParticlePanel(pe)

8
direct/src/particles/ParticleFloorTest.py → tests/particles/test_particle_floor.py Executable file → Normal file
View File

@ -1,4 +1,3 @@
from panda3d.core import NodePath, Vec3
from panda3d.physics import LinearVectorForce
from direct.particles import ParticleEffect
@ -49,10 +48,9 @@ class ParticleFloorTest(NodePath):
self.f.enable()
if __name__ == "__main__":
from direct.directbase.TestStart import base
def test_ParticleFloorTest(base):
base.enableParticles()
pt = ParticleFloorTest()
pt.reparentTo(base.render)
pt.start()
base.camera.setY(-10.0)
base.run()
# base.camera.setY(-10.0)

15
direct/src/physics/FallTest.py → tests/physics/test_fall.py
View File

@ -26,9 +26,9 @@ class FallTest(NodePath):
avatarNodePath = base.loader.loadModel("models/misc/smiley")
assert not avatarNodePath.isEmpty()
camLL = base.render.find("**/camLL")
camLL.reparentTo(avatarNodePath)
camLL.setPosHpr(0, -10, 0, 0, 0, 0)
# camLL = base.render.find("**/camLL")
# camLL.reparentTo(avatarNodePath)
# camLL.setPosHpr(0, -10, 0, 0, 0, 0)
avatarNodePath.reparentTo(actorNodePath)
#avatarNodePath.setPos(Vec3(0))
#avatarNodePath.setHpr(Vec3(0))
@ -86,10 +86,11 @@ class FallTest(NodePath):
#self.actorNode.getPhysicsObject().resetPosition(self.avatarNodePath.getPos())
#self.actorNode.updateTransform()
if __name__ == "__main__":
from direct.directbase.ThreeUpStart import base
def test_FallTest(base):
base.enableParticles()
base.addAngularIntegrator()
test = FallTest()
test.reparentTo(base.render)
test.setup()
base.camera.setY(-10.0)
base.run()
# base.camera.setY(-10.0)

15
direct/src/physics/RotationTest.py → tests/physics/test_rotation.py
View File

@ -27,9 +27,9 @@ class RotationTest(NodePath):
avatarNodePath = base.loader.loadModel("models/misc/smiley")
assert not avatarNodePath.isEmpty()
camLL = base.render.find("**/camLL")
camLL.reparentTo(avatarNodePath)
camLL.setPosHpr(0, -10, 0, 0, 0, 0)
# camLL = base.render.find("**/camLL")
# camLL.reparentTo(avatarNodePath)
# camLL.setPosHpr(0, -10, 0, 0, 0, 0)
avatarNodePath.reparentTo(actorNodePath)
#avatarNodePath.setPos(Vec3(0))
#avatarNodePath.setHpr(Vec3(0))
@ -97,10 +97,11 @@ class RotationTest(NodePath):
#self.actorNode.getPhysicsObject().resetPosition(self.avatarNodePath.getPos())
#self.actorNode.updateTransform()
if __name__ == "__main__":
from direct.directbase.ThreeUpStart import base
def test_RotationTest(base):
base.enableParticles()
base.addAngularIntegrator()
test = RotationTest()
test.reparentTo(base.render)
test.setup()
base.camera.setY(-10.0)
base.run()
# base.camera.setY(-10.0)

145
tests/showbase/test_CountedResource.py Normal file
View File

@ -0,0 +1,145 @@
from direct.showbase.CountedResource import CountedResource
def test_CountedResource():
class MouseResource(CountedResource):
"""
A simple class to demonstrate the acquisition of a resource.
"""
@classmethod
def acquire(cls):
# The call to the super-class's acquire() is
# not necessary at the moment, but may be in
# the future, so do it now for good measure.
super(MouseResource, cls).acquire()
# Now acquire the resource this class is
# managing.
print('-- Acquire Mouse')
@classmethod
def release(cls):
# First, release the resource this class is
# managing.
print('-- Release Mouse')
# The call to the super-class's release() is
# not necessary at the moment, but may be in
# the future, so do it now for good measure.
super(MouseResource, cls).release()
def __init__(self):
super(MouseResource, self).__init__()
def __del__(self):
super(MouseResource, self).__del__()
class CursorResource(CountedResource):
"""
A class to demonstrate how to implement a dependent
resource. Notice how this class also inherits from
CountedResource. Instead of subclassing MouseCounter,
we will just acquire it in our __init__() and release
it in our __del__().
"""
@classmethod
def acquire(cls):
super(CursorResource, cls).acquire()
print('-- Acquire Cursor')
@classmethod
def release(cls):
print('-- Release Cursor')
super(CursorResource, cls).release()
def __init__(self):
# The required resource references should
# be stored on 'self' since we want to
# release it when the object is deleted.
self.__mouseResource = MouseResource()
# Call the super-classes __init__()
# after all required resources are
# referenced.
super(CursorResource, self).__init__()
def __del__(self):
# Free up the most dependent resource
# first, the one this class is managing.
super(CursorResource, self).__del__()
# Now unlink any required resources.
del self.__mouseResource
class InvalidResource(MouseResource):
@classmethod
def acquire(cls):
super(InvalidResource, cls).acquire()
print('-- Acquire Invalid')
@classmethod
def release(cls):
print('-- Release Invalid')
super(InvalidResource, cls).release()
print('\nAllocate Mouse')
m = MouseResource()
print('Free up Mouse')
del m
print('\nAllocate Cursor')
c = CursorResource()
print('Free up Cursor')
del c
print('\nAllocate Mouse then Cursor')
m = MouseResource()
c = CursorResource()
print('Free up Cursor')
del c
print('Free up Mouse')
del m
print('\nAllocate Mouse then Cursor')
m = MouseResource()
c = CursorResource()
print('Free up Mouse')
del m
print('Free up Cursor')
del c
print('\nAllocate Cursor then Mouse')
c = CursorResource()
m = MouseResource()
print('Free up Mouse')
del m
print('Free up Cursor')
del c
print('\nAllocate Cursor then Mouse')
c = CursorResource()
m = MouseResource()
print('Free up Cursor')
del c
# example of an invalid subclass
try:
print('\nAllocate Invalid')
i = InvalidResource()
print('Free up Invalid')
except AssertionError as e:
print(e)
print('')
print('Free up Mouse')
del m
def demoFunc():
print('\nAllocate Cursor within function')
c = CursorResource()
print('Cursor will be freed on function exit')
demoFunc()

30
tests/showbase/test_DistancePhasedNode.py Normal file
View File

@ -0,0 +1,30 @@
from panda3d import core
from direct.showbase.DistancePhasedNode import BufferedDistancePhasedNode
def test_BufferedDistancePhasedNode(base):
cSphere = core.CollisionSphere(0, 0, 0, 0.1)
cNode = core.CollisionNode('camCol')
cNode.addSolid(cSphere)
cNodePath = core.NodePath(cNode)
# cNodePath.reparentTo(base.cam)
# cNodePath.show()
# cNodePath.setPos(25,0,0)
base.cTrav = core.CollisionTraverser()
eventHandler = core.CollisionHandlerEvent()
eventHandler.addInPattern('enter%in')
eventHandler.addOutPattern('exit%in')
# messenger.toggleVerbose()
base.cTrav.addCollider(cNodePath, eventHandler)
p = BufferedDistancePhasedNode('p', {'At': (10, 20), 'Near': (100, 200), 'Far': (1000, 1020)},
autoCleanup=False,
fromCollideNode=cNodePath,
)
p.reparentTo(base.render)
p._DistancePhasedNode__oneTimeCollide()
base.eventMgr.doEvents()

43
tests/showbase/test_PythonUtil.py
View File

@ -56,6 +56,49 @@ def test_flywheel():
assert obj2count[4] == 4 * 3
def test_formatTimeCompact():
ftc = PythonUtil.formatTimeCompact
assert ftc(0) == '0s'
assert ftc(1) == '1s'
assert ftc(60) == '1m0s'
assert ftc(64) == '1m4s'
assert ftc(60*60) == '1h0m0s'
assert ftc(24*60*60) == '1d0h0m0s'
assert ftc(24*60*60 + 2*60*60 + 34*60 + 12) == '1d2h34m12s'
def test_formatTimeExact():
fte = PythonUtil.formatTimeExact
assert fte(0) == '0s'
assert fte(1) == '1s'
assert fte(2) == '2s'
assert fte(61) == '1m1s'
assert fte(60) == '1m'
assert fte(60*60) == '1h'
assert fte(24*60*60) == '1d'
assert fte((24*60*60) + (2 * 60)) == '1d0h2m'
del fte
def test_AlphabetCounter():
tempList = []
ac = PythonUtil.AlphabetCounter()
for i in range(26*3):
tempList.append(ac.next())
assert tempList == [ 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z',
'AA','AB','AC','AD','AE','AF','AG','AH','AI','AJ','AK','AL','AM','AN','AO','AP','AQ','AR','AS','AT','AU','AV','AW','AX','AY','AZ',
'BA','BB','BC','BD','BE','BF','BG','BH','BI','BJ','BK','BL','BM','BN','BO','BP','BQ','BR','BS','BT','BU','BV','BW','BX','BY','BZ',]
ac = PythonUtil.AlphabetCounter()
num = 26 # A-Z
num += (26*26) # AA-ZZ
num += 26 # AAZ
num += 1 # ABA
num += 2 # ABC
for i in range(num):
x = ac.next()
assert x == 'ABC'
def test_unescape_html_string():
assert PythonUtil.unescapeHtmlString('asdf') == 'asdf'
assert PythonUtil.unescapeHtmlString('as+df') == 'as df'

103
tests/stdpy/test_threading.py
View File

@ -1,3 +1,5 @@
import sys
from panda3d import core
from direct.stdpy import threading
import pytest
@ -5,3 +7,104 @@ import pytest
def test_threading_error():
with pytest.raises(threading.ThreadError):
threading.stack_size()
def test_threading():
from collections import deque
_sleep = core.Thread.sleep
_VERBOSE = False
class _Verbose(object):
def __init__(self, verbose=None):
if verbose is None:
verbose = _VERBOSE
self.__verbose = verbose
def _note(self, format, *args):
if self.__verbose:
format = format % args
format = "%s: %s\n" % (
threading.currentThread().getName(), format)
sys.stderr.write(format)
class BoundedQueue(_Verbose):
def __init__(self, limit):
_Verbose.__init__(self)
self.mon = threading.Lock(name = "BoundedQueue.mon")
self.rc = threading.Condition(self.mon)
self.wc = threading.Condition(self.mon)
self.limit = limit
self.queue = deque()
def put(self, item):
self.mon.acquire()
while len(self.queue) >= self.limit:
self._note("put(%s): queue full", item)
self.wc.wait()
self.queue.append(item)
self._note("put(%s): appended, length now %d",
item, len(self.queue))
self.rc.notify()
self.mon.release()
def get(self):
self.mon.acquire()
while not self.queue:
self._note("get(): queue empty")
self.rc.wait()
item = self.queue.popleft()
self._note("get(): got %s, %d left", item, len(self.queue))
self.wc.notify()
self.mon.release()
return item
class ProducerThread(threading.Thread):
def __init__(self, queue, quota):
threading.Thread.__init__(self, name="Producer")
self.queue = queue
self.quota = quota
def run(self):
from random import random
counter = 0
while counter < self.quota:
counter = counter + 1
self.queue.put("%s.%d" % (self.getName(), counter))
_sleep(random() * 0.00001)
class ConsumerThread(threading.Thread):
def __init__(self, queue, count):
threading.Thread.__init__(self, name="Consumer")
self.queue = queue
self.count = count
def run(self):
while self.count > 0:
item = self.queue.get()
print(item)
self.count = self.count - 1
NP = 3
QL = 4
NI = 5
Q = BoundedQueue(QL)
P = []
for i in range(NP):
t = ProducerThread(Q, NI)
t.setName("Producer-%d" % (i+1))
P.append(t)
C = ConsumerThread(Q, NI*NP)
for t in P:
t.start()
_sleep(0.000001)
C.start()
for t in P:
t.join()
C.join()

84
tests/stdpy/test_threading2.py Normal file
View File

@ -0,0 +1,84 @@
from collections import deque
from panda3d import core
from direct.stdpy import threading2
def test_threading2():
class BoundedQueue(threading2._Verbose):
def __init__(self, limit):
threading2._Verbose.__init__(self)
self.mon = threading2.RLock()
self.rc = threading2.Condition(self.mon)
self.wc = threading2.Condition(self.mon)
self.limit = limit
self.queue = deque()
def put(self, item):
self.mon.acquire()
while len(self.queue) >= self.limit:
self._note("put(%s): queue full", item)
self.wc.wait()
self.queue.append(item)
self._note("put(%s): appended, length now %d",
item, len(self.queue))
self.rc.notify()
self.mon.release()
def get(self):
self.mon.acquire()
while not self.queue:
self._note("get(): queue empty")
self.rc.wait()
item = self.queue.popleft()
self._note("get(): got %s, %d left", item, len(self.queue))
self.wc.notify()
self.mon.release()
return item
class ProducerThread(threading2.Thread):
def __init__(self, queue, quota):
threading2.Thread.__init__(self, name="Producer")
self.queue = queue
self.quota = quota
def run(self):
from random import random
counter = 0
while counter < self.quota:
counter = counter + 1
self.queue.put("%s.%d" % (self.getName(), counter))
core.Thread.sleep(random() * 0.00001)
class ConsumerThread(threading2.Thread):
def __init__(self, queue, count):
threading2.Thread.__init__(self, name="Consumer")
self.queue = queue
self.count = count
def run(self):
while self.count > 0:
item = self.queue.get()
print(item)
self.count = self.count - 1
NP = 3
QL = 4
NI = 5
Q = BoundedQueue(QL)
P = []
for i in range(NP):
t = ProducerThread(Q, NI)
t.setName("Producer-%d" % (i+1))
P.append(t)
C = ConsumerThread(Q, NI*NP)
for t in P:
t.start()
core.Thread.sleep(0.000001)
C.start()
for t in P:
t.join()
C.join()

518
tests/task/test_Task.py Normal file
View File

@ -0,0 +1,518 @@
from panda3d import core
from direct.task import Task
def test_TaskManager():
tm = Task.TaskManager()
tm.mgr = core.AsyncTaskManager("Test manager")
tm.setClock(core.ClockObject())
tm.setupTaskChain("default", tickClock = True)
tm._startTrackingMemLeaks = lambda: None
tm._stopTrackingMemLeaks = lambda: None
tm._checkMemLeaks = lambda: None
# check for memory leaks after every test
tm._startTrackingMemLeaks()
tm._checkMemLeaks()
# run-once task
l = []
def _testDone(task, l=l):
l.append(None)
return task.done
tm.add(_testDone, 'testDone')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 1
_testDone = None
tm._checkMemLeaks()
# remove by name
def _testRemoveByName(task):
return task.done
tm.add(_testRemoveByName, 'testRemoveByName')
assert tm.remove('testRemoveByName') == 1
assert tm.remove('testRemoveByName') == 0
_testRemoveByName = None
tm._checkMemLeaks()
# duplicate named tasks
def _testDupNamedTasks(task):
return task.done
tm.add(_testDupNamedTasks, 'testDupNamedTasks')
tm.add(_testDupNamedTasks, 'testDupNamedTasks')
assert tm.remove('testRemoveByName') == 0
_testDupNamedTasks = None
tm._checkMemLeaks()
# continued task
l = []
def _testCont(task, l = l):
l.append(None)
return task.cont
tm.add(_testCont, 'testCont')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 2
tm.remove('testCont')
_testCont = None
tm._checkMemLeaks()
# continue until done task
l = []
def _testContDone(task, l = l):
l.append(None)
if len(l) >= 2:
return task.done
else:
return task.cont
tm.add(_testContDone, 'testContDone')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 2
tm.step()
assert len(l) == 2
assert not tm.hasTaskNamed('testContDone')
_testContDone = None
tm._checkMemLeaks()
# hasTaskNamed
def _testHasTaskNamed(task):
return task.done
tm.add(_testHasTaskNamed, 'testHasTaskNamed')
assert tm.hasTaskNamed('testHasTaskNamed')
tm.step()
assert not tm.hasTaskNamed('testHasTaskNamed')
_testHasTaskNamed = None
tm._checkMemLeaks()
# task sort
l = []
def _testPri1(task, l = l):
l.append(1)
return task.cont
def _testPri2(task, l = l):
l.append(2)
return task.cont
tm.add(_testPri1, 'testPri1', sort = 1)
tm.add(_testPri2, 'testPri2', sort = 2)
tm.step()
assert len(l) == 2
assert l == [1, 2,]
tm.step()
assert len(l) == 4
assert l == [1, 2, 1, 2,]
tm.remove('testPri1')
tm.remove('testPri2')
_testPri1 = None
_testPri2 = None
tm._checkMemLeaks()
# task extraArgs
l = []
def _testExtraArgs(arg1, arg2, l=l):
l.extend([arg1, arg2,])
return Task.done
tm.add(_testExtraArgs, 'testExtraArgs', extraArgs=[4,5])
tm.step()
assert len(l) == 2
assert l == [4, 5,]
_testExtraArgs = None
tm._checkMemLeaks()
# task appendTask
l = []
def _testAppendTask(arg1, arg2, task, l=l):
l.extend([arg1, arg2,])
return task.done
tm.add(_testAppendTask, '_testAppendTask', extraArgs=[4,5], appendTask=True)
tm.step()
assert len(l) == 2
assert l == [4, 5,]
_testAppendTask = None
tm._checkMemLeaks()
# task uponDeath
l = []
def _uponDeathFunc(task, l=l):
l.append(task.name)
def _testUponDeath(task):
return Task.done
tm.add(_testUponDeath, 'testUponDeath', uponDeath=_uponDeathFunc)
tm.step()
assert len(l) == 1
assert l == ['testUponDeath']
_testUponDeath = None
_uponDeathFunc = None
tm._checkMemLeaks()
# task owner
class _TaskOwner:
def _addTask(self, task):
self.addedTaskName = task.name
def _clearTask(self, task):
self.clearedTaskName = task.name
to = _TaskOwner()
l = []
def _testOwner(task):
return Task.done
tm.add(_testOwner, 'testOwner', owner=to)
tm.step()
assert getattr(to, 'addedTaskName', None) == 'testOwner'
assert getattr(to, 'clearedTaskName', None) == 'testOwner'
_testOwner = None
del to
_TaskOwner = None
tm._checkMemLeaks()
doLaterTests = [0,]
# doLater
l = []
def _testDoLater1(task, l=l):
l.append(1)
def _testDoLater2(task, l=l):
l.append(2)
def _monitorDoLater(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .03:
assert l == [1, 2,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLater1, 'testDoLater1')
tm.doMethodLater(.02, _testDoLater2, 'testDoLater2')
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLater, 'monitorDoLater', sort=10)
_testDoLater1 = None
_testDoLater2 = None
_monitorDoLater = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater sort
l = []
def _testDoLaterPri1(task, l=l):
l.append(1)
def _testDoLaterPri2(task, l=l):
l.append(2)
def _monitorDoLaterPri(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .02:
assert l == [1, 2,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterPri1, 'testDoLaterPri1', sort=1)
tm.doMethodLater(.01, _testDoLaterPri2, 'testDoLaterPri2', sort=2)
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterPri, 'monitorDoLaterPri', sort=10)
_testDoLaterPri1 = None
_testDoLaterPri2 = None
_monitorDoLaterPri = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater extraArgs
l = []
def _testDoLaterExtraArgs(arg1, l=l):
l.append(arg1)
def _monitorDoLaterExtraArgs(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .02:
assert l == [3,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterExtraArgs, 'testDoLaterExtraArgs', extraArgs=[3,])
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterExtraArgs, 'monitorDoLaterExtraArgs', sort=10)
_testDoLaterExtraArgs = None
_monitorDoLaterExtraArgs = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater appendTask
l = []
def _testDoLaterAppendTask(arg1, task, l=l):
assert task.name == 'testDoLaterAppendTask'
l.append(arg1)
def _monitorDoLaterAppendTask(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .02:
assert l == [4,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterAppendTask, 'testDoLaterAppendTask',
extraArgs=[4,], appendTask=True)
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterAppendTask, 'monitorDoLaterAppendTask', sort=10)
_testDoLaterAppendTask = None
_monitorDoLaterAppendTask = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater uponDeath
l = []
def _testUponDeathFunc(task, l=l):
assert task.name == 'testDoLaterUponDeath'
l.append(10)
def _testDoLaterUponDeath(arg1, l=l):
return Task.done
def _monitorDoLaterUponDeath(task, tm=tm, l=l, doLaterTests=doLaterTests):
if task.time > .02:
assert l == [10,]
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterUponDeath, 'testDoLaterUponDeath',
uponDeath=_testUponDeathFunc)
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterUponDeath, 'monitorDoLaterUponDeath', sort=10)
_testUponDeathFunc = None
_testDoLaterUponDeath = None
_monitorDoLaterUponDeath = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# doLater owner
class _DoLaterOwner:
def _addTask(self, task):
self.addedTaskName = task.name
def _clearTask(self, task):
self.clearedTaskName = task.name
doLaterOwner = _DoLaterOwner()
l = []
def _testDoLaterOwner(l=l):
pass
def _monitorDoLaterOwner(task, tm=tm, l=l, doLaterOwner=doLaterOwner,
doLaterTests=doLaterTests):
if task.time > .02:
assert getattr(doLaterOwner, 'addedTaskName', None) == 'testDoLaterOwner'
assert getattr(doLaterOwner, 'clearedTaskName', None) == 'testDoLaterOwner'
doLaterTests[0] -= 1
return task.done
return task.cont
tm.doMethodLater(.01, _testDoLaterOwner, 'testDoLaterOwner',
owner=doLaterOwner)
doLaterTests[0] += 1
# make sure we run this task after the doLaters if they all occur on the same frame
tm.add(_monitorDoLaterOwner, 'monitorDoLaterOwner', sort=10)
_testDoLaterOwner = None
_monitorDoLaterOwner = None
del doLaterOwner
_DoLaterOwner = None
# don't check until all the doLaters are finished
#tm._checkMemLeaks()
# run the doLater tests
while doLaterTests[0] > 0:
tm.step()
del doLaterTests
tm._checkMemLeaks()
# getTasks
def _testGetTasks(task):
return task.cont
# No doLaterProcessor in the new world.
assert len(tm.getTasks()) == 0
tm.add(_testGetTasks, 'testGetTasks1')
assert len(tm.getTasks()) == 1
assert (tm.getTasks()[0].name == 'testGetTasks1' or
tm.getTasks()[1].name == 'testGetTasks1')
tm.add(_testGetTasks, 'testGetTasks2')
tm.add(_testGetTasks, 'testGetTasks3')
assert len(tm.getTasks()) == 3
tm.remove('testGetTasks2')
assert len(tm.getTasks()) == 2
tm.remove('testGetTasks1')
tm.remove('testGetTasks3')
assert len(tm.getTasks()) == 0
_testGetTasks = None
tm._checkMemLeaks()
# getDoLaters
def _testGetDoLaters():
pass
assert len(tm.getDoLaters()) == 0
tm.doMethodLater(.1, _testGetDoLaters, 'testDoLater1')
assert len(tm.getDoLaters()) == 1
assert tm.getDoLaters()[0].name == 'testDoLater1'
tm.doMethodLater(.1, _testGetDoLaters, 'testDoLater2')
tm.doMethodLater(.1, _testGetDoLaters, 'testDoLater3')
assert len(tm.getDoLaters()) == 3
tm.remove('testDoLater2')
assert len(tm.getDoLaters()) == 2
tm.remove('testDoLater1')
tm.remove('testDoLater3')
assert len(tm.getDoLaters()) == 0
_testGetDoLaters = None
tm._checkMemLeaks()
# duplicate named doLaters removed via taskMgr.remove
def _testDupNameDoLaters():
pass
# the doLaterProcessor is always running
tm.doMethodLater(.1, _testDupNameDoLaters, 'testDupNameDoLater')
tm.doMethodLater(.1, _testDupNameDoLaters, 'testDupNameDoLater')
assert len(tm.getDoLaters()) == 2
tm.remove('testDupNameDoLater')
assert len(tm.getDoLaters()) == 0
_testDupNameDoLaters = None
tm._checkMemLeaks()
# duplicate named doLaters removed via remove()
def _testDupNameDoLatersRemove():
pass
# the doLaterProcessor is always running
dl1 = tm.doMethodLater(.1, _testDupNameDoLatersRemove, 'testDupNameDoLaterRemove')
dl2 = tm.doMethodLater(.1, _testDupNameDoLatersRemove, 'testDupNameDoLaterRemove')
assert len(tm.getDoLaters()) == 2
dl2.remove()
assert len(tm.getDoLaters()) == 1
dl1.remove()
assert len(tm.getDoLaters()) == 0
_testDupNameDoLatersRemove = None
# nameDict etc. isn't cleared out right away with task.remove()
tm._checkMemLeaks()
# getTasksNamed
def _testGetTasksNamed(task):
return task.cont
assert len(tm.getTasksNamed('testGetTasksNamed')) == 0
tm.add(_testGetTasksNamed, 'testGetTasksNamed')
assert len(tm.getTasksNamed('testGetTasksNamed')) == 1
assert tm.getTasksNamed('testGetTasksNamed')[0].name == 'testGetTasksNamed'
tm.add(_testGetTasksNamed, 'testGetTasksNamed')
tm.add(_testGetTasksNamed, 'testGetTasksNamed')
assert len(tm.getTasksNamed('testGetTasksNamed')) == 3
tm.remove('testGetTasksNamed')
assert len(tm.getTasksNamed('testGetTasksNamed')) == 0
_testGetTasksNamed = None
tm._checkMemLeaks()
# removeTasksMatching
def _testRemoveTasksMatching(task):
return task.cont
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching')
assert len(tm.getTasksNamed('testRemoveTasksMatching')) == 1
tm.removeTasksMatching('testRemoveTasksMatching')
assert len(tm.getTasksNamed('testRemoveTasksMatching')) == 0
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching1')
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching2')
assert len(tm.getTasksNamed('testRemoveTasksMatching1')) == 1
assert len(tm.getTasksNamed('testRemoveTasksMatching2')) == 1
tm.removeTasksMatching('testRemoveTasksMatching*')
assert len(tm.getTasksNamed('testRemoveTasksMatching1')) == 0
assert len(tm.getTasksNamed('testRemoveTasksMatching2')) == 0
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching1a')
tm.add(_testRemoveTasksMatching, 'testRemoveTasksMatching2a')
assert len(tm.getTasksNamed('testRemoveTasksMatching1a')) == 1
assert len(tm.getTasksNamed('testRemoveTasksMatching2a')) == 1
tm.removeTasksMatching('testRemoveTasksMatching?a')
assert len(tm.getTasksNamed('testRemoveTasksMatching1a')) == 0
assert len(tm.getTasksNamed('testRemoveTasksMatching2a')) == 0
_testRemoveTasksMatching = None
tm._checkMemLeaks()
# create Task object and add to mgr
l = []
def _testTaskObj(task, l=l):
l.append(None)
return task.cont
t = Task.Task(_testTaskObj)
tm.add(t, 'testTaskObj')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 2
tm.remove('testTaskObj')
tm.step()
assert len(l) == 2
_testTaskObj = None
tm._checkMemLeaks()
# remove Task via task.remove()
l = []
def _testTaskObjRemove(task, l=l):
l.append(None)
return task.cont
t = Task.Task(_testTaskObjRemove)
tm.add(t, 'testTaskObjRemove')
tm.step()
assert len(l) == 1
tm.step()
assert len(l) == 2
t.remove()
tm.step()
assert len(l) == 2
del t
_testTaskObjRemove = None
tm._checkMemLeaks()
# set/get Task sort
l = []
def _testTaskObjSort(arg, task, l=l):
l.append(arg)
return task.cont
t1 = Task.Task(_testTaskObjSort)
t2 = Task.Task(_testTaskObjSort)
tm.add(t1, 'testTaskObjSort1', extraArgs=['a',], appendTask=True, sort=1)
tm.add(t2, 'testTaskObjSort2', extraArgs=['b',], appendTask=True, sort=2)
tm.step()
assert len(l) == 2
assert l == ['a', 'b']
assert t1.getSort() == 1
assert t2.getSort() == 2
t1.setSort(3)
assert t1.getSort() == 3
tm.step()
assert len(l) == 4
assert l == ['a', 'b', 'b', 'a',]
t1.remove()
t2.remove()
tm.step()
assert len(l) == 4
del t1
del t2
_testTaskObjSort = None
tm._checkMemLeaks()
del l
tm.destroy()
del tm

4
tests/task/test_task_arg.py
View File

@ -2,7 +2,7 @@ from direct.showbase.ShowBase import ShowBase
from direct.task import Task
from panda3d.core import Vec2
def test_task_arg():
def test_task_arg(base):
def test(ship, flood, task):
ship.y += flood
return task.done
@ -10,7 +10,6 @@ def test_task_arg():
ship = Vec2(2.2, 2)
flood = 1
base = ShowBase(windowType='none')
task = base.addTask(test, 'test_task', extraArgs=[ship, flood], appendTask=True)
base.taskMgr.step()
assert ship.y == 3
@ -25,4 +24,3 @@ def test_task_arg():
task = base.taskMgr.add(task)
base.taskMgr.step()
assert ship.y == 6

10
direct/src/task/TaskTester.py → tests/task/test_tasks.py Executable file → Normal file
View File

@ -1,11 +1,7 @@
"""Undocumented Module"""
__all__ = []
from direct.task.TaskManagerGlobal import taskMgr
from direct.task import Task
import random
taskMgr = Task.TaskManager()
numTasks = 10000
maxDelay = 20
counter = 0
@ -24,8 +20,10 @@ def taskCallback(task):
spawnNewTask()
return Task.done
if __name__ == '__main__':
def test_tasks():
taskMgr.removeTasksMatching("taskTester*")
for i in range(numTasks):
spawnNewTask()
taskMgr.destroy()

7
tests/test_imports.py
View File

@ -93,7 +93,6 @@ def test_imports_direct():
import direct.gui.DirectGui
import direct.gui.DirectGuiBase
import direct.gui.DirectGuiGlobals
import direct.gui.DirectGuiTest
import direct.gui.DirectLabel
import direct.gui.DirectOptionMenu
import direct.gui.DirectRadioButton
@ -112,7 +111,6 @@ def test_imports_direct():
import direct.interval.Interval
import direct.interval.IntervalGlobal
import direct.interval.IntervalManager
import direct.interval.IntervalTest
import direct.interval.LerpBlendHelpers
import direct.interval.LerpInterval
import direct.interval.MetaInterval
@ -185,7 +183,6 @@ def test_imports_direct():
import direct.task.Task
import direct.task.TaskManagerGlobal
import direct.task.TaskProfiler
import direct.task.TaskTester
import direct.task.Timer
@ -196,13 +193,9 @@ def test_imports_direct_physics():
import direct.particles.ForceGroup
import direct.particles.GlobalForceGroup
import direct.particles.ParticleEffect
import direct.particles.ParticleFloorTest
import direct.particles.ParticleManagerGlobal
import direct.particles.ParticleTest
import direct.particles.Particles
import direct.particles.SpriteParticleRendererExt
import direct.physics.FallTest
import direct.physics.RotationTest
import direct.showbase.PhysicsManagerGlobal

11
tests/tkpanels/test_ParticlePanel.py Normal file
View File

@ -0,0 +1,11 @@
import pytest
Pmw = pytest.importorskip('Pmw')
from direct.tkpanels.ParticlePanel import ParticlePanel
def test_ParticlePanel(base, tk_toplevel):
root = Pmw.initialise()
pp = ParticlePanel()
base.pp=pp
#ve = VectorEntry(Toplevel(), relief = GROOVE)
#ve.pack()

12
tests/tkpanels/test_Placer.py Normal file
View File

@ -0,0 +1,12 @@
import pytest
Pmw = pytest.importorskip('Pmw')
from direct.showbase.ShowBase import ShowBase
from direct.tkpanels.Placer import Placer
def test_Placer(window):
base = ShowBase()
base.start_direct()
root = Pmw.initialise()
widget = Placer()
base.destroy()

7
tests/tkwidgets/test_AppShell.py Normal file
View File

@ -0,0 +1,7 @@
import pytest
pytest.importorskip('Pmw')
from direct.tkwidgets import AppShell
def test_TestAppShell(tk_toplevel):
test = AppShell.TestAppShell(balloon_state='none')

19
tests/tkwidgets/test_Dial.py Normal file
View File

@ -0,0 +1,19 @@
import tkinter as tk
import pytest
pytest.importorskip('Pmw')
from direct.tkwidgets.Dial import Dial
def test_Dial(tk_toplevel):
tl = tk_toplevel
d = Dial(tl)
d2 = Dial(tl, dial_numSegments = 12, max = 360,
dial_fRollover = 0, value = 180)
d3 = Dial(tl, dial_numSegments = 12, max = 90, min = -90,
dial_fRollover = 0)
d4 = Dial(tl, dial_numSegments = 16, max = 256,
dial_fRollover = 0)
d.pack(expand = 1, fill = tk.X)
d2.pack(expand = 1, fill = tk.X)
d3.pack(expand = 1, fill = tk.X)
d4.pack(expand = 1, fill = tk.X)

44
tests/tkwidgets/test_EntryScale.py Normal file
View File

@ -0,0 +1,44 @@
import pytest
pytest.importorskip('Pmw')
from direct.tkwidgets.EntryScale import EntryScale, EntryScaleGroup
def test_EntryScale(tk_toplevel):
# Initialise Tkinter and Pmw.
root = tk_toplevel
root.title('Pmw EntryScale demonstration')
# Dummy command
def printVal(val):
print(val)
# Create and pack a EntryScale megawidget.
mega1 = EntryScale(root, command = printVal)
mega1.pack(side = 'left', expand = 1, fill = 'x')
# These are things you can set/configure
# Starting value for entryScale
#mega1['value'] = 123.456
#mega1['text'] = 'Drive delta X'
#mega1['min'] = 0.0
#mega1['max'] = 1000.0
#mega1['resolution'] = 1.0
# To change the color of the label:
#mega1.label['foreground'] = 'Red'
# Max change/update, default is 100
# To have really fine control, for example
# mega1['maxVelocity'] = 0.1
# Number of digits to the right of the decimal point, default = 2
# mega1['numDigits'] = 5
# To create a entryScale group to set an RGBA value:
group1 = EntryScaleGroup(root, dim = 4,
title = 'Simple RGBA Panel',
labels = ('R', 'G', 'B', 'A'),
Valuator_min = 0.0,
Valuator_max = 255.0,
Valuator_resolution = 1.0,
command = printVal)
# Uncomment this if you aren't running in IDLE
#root.mainloop()

41
tests/tkwidgets/test_Floater.py Normal file
View File

@ -0,0 +1,41 @@
import pytest
pytest.importorskip('Pmw')
from direct.tkwidgets.Floater import Floater, FloaterGroup
def test_Floater(tk_toplevel):
# Initialise Tkinter and Pmw.
root = tk_toplevel
root.title('Pmw Floater demonstration')
# Dummy command
def printVal(val):
print(val)
# Create and pack a Floater megawidget.
mega1 = Floater(root, command = printVal)
mega1.pack(side = 'left', expand = 1, fill = 'x')
# These are things you can set/configure
# Starting value for floater
#mega1['value'] = 123.456
#mega1['text'] = 'Drive delta X'
# To change the color of the label:
#mega1.label['foreground'] = 'Red'
# Max change/update, default is 100
# To have really fine control, for example
#mega1['maxVelocity'] = 0.1
# Number of digits to the right of the decimal point, default = 2
#mega1['numDigits'] = 5
# To create a floater group to set an RGBA value:
group1 = FloaterGroup(root, dim = 4,
title = 'Simple RGBA Panel',
labels = ('R', 'G', 'B', 'A'),
Valuator_min = 0.0,
Valuator_max = 255.0,
Valuator_resolution = 1.0,
command = printVal)
# Uncomment this if you aren't running in IDLE
#root.mainloop()

17
tests/tkwidgets/test_VectorWidgets.py Normal file
View File

@ -0,0 +1,17 @@
import pytest
pytest.importorskip('Pmw')
from direct.tkwidgets import VectorWidgets
def test_VectorEntry(tk_toplevel):
root = tk_toplevel
root.title('Vector Widget demo')
ve = VectorWidgets.VectorEntry(root)
ve.pack()
v3e = VectorWidgets.Vector3Entry(root)
v3e.pack()
v4e = VectorWidgets.Vector4Entry(root)
v4e.pack()
ce = VectorWidgets.ColorEntry(root)
ce.pack()