Consider a process composed of three algorithms: makeRuns, meld and traverse (each taking the output of the previous one as input): it is interesting to test each separately. Now consider three (or more) instances (e.g. singleton, twoPairs, SimplePair): it is interesting to test the behavior of the three algorithms on each of those instances. The result can be described as an array of (say) 3*3 tests. Is the code below the correct implementation of such a test? (I find the result of the tests to be lacking in clarity.)
import unittest, doctest
import insertionRank
##############################################################################
class TestMTDRAG(unittest.TestCase):
def test_singleton(self):
"""Basic example [1]."""
run = makeRuns([1])
self.assertEqual(run,[Node(1,1)])
graph = meld(run)
self.assertEqual(graph,Node(1,1))
lengths=traverse(graph)
self.assertEqual(lengths,[1,0])
def test_TwoPairsExample(self):
"""Two messages of distinct probabilities [1,2]."""
run = makeRuns([1,2])
self.assertEqual(run,[Node(1,1),Node(2,1)])
graph = meld(run)
self.assertEqual(graph,Node(3,1,[Node(1,1),Node(2,1)]))
lengths=traverse(graph)
self.assertEqual(lengths,[0,2,0])
def test_SimplePairExample(self):
"""Basic example of four equiprobable messages [1,1,1,1]."""
run = makeRuns([1,1,1,1])
self.assertEqual(run,[Node(1,4)])
graph = meld(run)
self.assertEqual(graph,
Node(4,1,
[Node(2,2,[Node(1,4),Node(1,4)]),
Node(2,2,[Node(1,4),Node(1,4)])
]))
lengths=traverse(graph)
self.assertEqual(lengths,[0,0,4])
##############################################################################
class Node(object):
"""
A node in a MTDRAG (Moffat-Turpin Directed Rooted Acyclic Graph).
"""
def __init__(self, weight, frequency, children=[], minDepth=None, nbPathsAtMinDepth=0, nbPathsAtMinDepthPlusOne=0):
assert frequency>0
self.weight = weight
self.frequency = frequency
self.children = children
self.minDepth= minDepth
self.nbPathsAtMinDepth = nbPathsAtMinDepth
self.nbPathsAtMinDepthPlusOne = nbPathsAtMinDepthPlusOne
def computeDepths(self,depthParent=-1,nd=1,ndp1=0):
assert self.frequency>0
if self.minDepth==None: # Node is not yet labeled
self.minDepth = depthParent+1
self.nbPathsAtMinDepth = nd
self.nbPathsAtMinDepthPlusOne = ndp1
elif self.minDepth == depthParent: # Node labeled at same depth as parent
self.nbPathsAtMinDepthPlusOne += nd
else:
assert(self.minDepth==depthParent+1) # Node labeled at depth + 1 of parent
self.nbPathsAtMinDepth += nd
self.nbPathsAtMinDepthPlusOne += ndp1
if self.nbPathsAtMinDepthPlusOne > 0:
maxDepth = depthParent+2
else:
maxDepth = depthParent+1
for child in self.children:
maxDepth = max( maxDepth,
child.computeDepths(depthParent+1,self.nbPathsAtMinDepth,self.nbPathsAtMinDepthPlusOne)
)
return maxDepth
def collectDepths(self,M):
assert self.frequency>0
if self.children == []:
M[self.minDepth] += self.nbPathsAtMinDepth
M[self.minDepth+1] += self.nbPathsAtMinDepthPlusOne
else:
for child in self.children:
child.collectDepths(M)
def __eq__(self,node):
assert self.frequency>0
if node == None:
return False
assert node.frequency>0
return self.weight==node.weight and self.frequency == node.frequency and self.children == node.children
def __str__(self):
assert self.frequency>0
if self.children == None:
return "("+str(self.weight)+", "+str(self.frequency)+")"
else:
return "("+str(self.weight)+", "+str(self.frequency)+", "+str(self.children)+")"
def __repr__(self):
assert self.frequency>0
return self.__str__()
def copy(self):
assert self.frequency>0
return Node(self.weight,self.frequency,self.children,self.minDepth,self.nbPathsAtMinDepth,self.nbPathsAtMinDepthPlusOne)
##############################################################################
def makeRuns(A):
"""Compresses a list of INTEGER weights into a list of nodes, in
sublinear time (if all weights are not distinct).
"""
A.sort() # Linear if already sorted
output = []
position = 0
while position < len(A):
weight = A[position]
frequency = insertionRank.fibonacciSearch(A[position:],weight+1)
output.append(Node(weight,frequency))
position += frequency
return output
def meld(leaves):
"""Moffat and Turpin's algorithm to simulate van Leeuwen's
algorithm on a compressed representation of a list of weights.
"""
assert(len(leaves)>0)
graphs = []
source = leaves
first = source[0]
while first.frequency>1 or len(leaves)+len(graphs)>=2:
if first.frequency>1:
graphs.append(Node(first.weight*2,first.frequency//2,[first,first]))
if first.frequency %2 == 1:
first.frequency = 1
else:
source.remove(first)
else: #first.frequency == 1 but len(leaves)+len(graphs)>=2
source.remove(first)
source = min(s for s in (leaves,graphs) if s)
second = source[0]
assert(second.frequency>0)
if second.frequency==1:
source.remove(second)
graphs.append(Node(first.weight+second.weight,1,[first,second]))
else:
extract = second.copy()
extract.frequency=1
second.frequency -= 1
graphs.append(Node(first.weight+extract.weight,1,[first,extract]))
source = min(s for s in (leaves,graphs) if s)
first = source[0]
return(first)
def traverse(graph):
"""Computes an array containing the depth of the leaves of the
graph produced by the algorithm meld().
"""
if graph==None:
return None
maxDepth = graph.computeDepths()
M = [0]*(maxDepth+2)
graph.collectDepths(M)
return M
##############################################################################
def main():
unittest.main()
if __name__ == '__main__':
doctest.testmod()
main()
##############################################################################
