import unittest import operator from dejavu import containers class WarehouseTests(unittest.TestCase): def test_builtin_types(self): # ints avail, rem = containers.warehouse([1,2,3]) self.assertEqual([avail.next() for x in xrange(5)], [1,2,3,0,0]) avail, rem = containers.warehouse([1,2,3]) self.assertEqual([avail.next() for x in xrange(2)], [1,2]) self.assertEqual([x for x in rem], [3]) # strings avail, rem = containers.warehouse(['fish', 'bananas', 'old pyjamas']) self.assertEqual([avail.next() for x in xrange(5)], ['fish', 'bananas', 'old pyjamas', '', '']) avail, rem = containers.warehouse(['fish', 'bananas', 'old pyjamas']) self.assertEqual([avail.next() for x in xrange(2)], ['fish', 'bananas']) self.assertEqual([x for x in rem], ['old pyjamas']) # Empty seq avail, rem = containers.warehouse([]) self.assertRaises(ValueError, avail.next) def test_custom_classes(self): class Thing: def __init__(self, value=None): self.value = value things = Thing(1), Thing(2), Thing(3), Thing(4) avail, rem = containers.warehouse(things) self.assertEqual([avail.next().value for x in xrange(5)], [1, 2, 3, 4, None]) avail, rem = containers.warehouse(things) self.assertEqual([avail.next().value for x in xrange(2)], [1, 2]) self.assertEqual([x for x in rem], [things[2], things[3]]) # Empty seq avail, rem = containers.warehouse([], Thing) self.assertEqual([avail.next().value for x in xrange(2)], [None, None]) self.assertEqual([x for x in rem], []) class GraphTests(unittest.TestCase): def test_creation(self): g = containers.Graph() self.assertEqual(g, {}) g = containers.Graph({'a': []}) self.assertEqual(g, {'a': []}) g = containers.Graph({'a': []}, True) self.assertEqual(g, {'a': []}) def test_connect(self): g = containers.Graph() g.connect('A', 'B') self.assertEqual(g, {'A': ['B'], 'B': ['A'], }) g.connect('C', ('A', 'B')) self.assertEqual(g, {'A': ['B', 'C'], 'B': ['A', 'C'], 'C': ['A', 'B'], }) def test_chain(self): # Form the undirected graph: # A--B--C--D # | |\ / # | | \ / # E--F--G g = containers.Graph() g.chain('A', 'B', 'C', 'D', 'G', 'F', 'E', 'A') g.chain('B', 'F', 'G', 'B') self.assertEqual(g, {'A': ['B', 'E'], 'B': ['A', 'C', 'F', 'G'], 'C': ['B', 'D'], 'D': ['C', 'G'], 'E': ['F', 'A'], 'F': ['G', 'E', 'B'], 'G': ['D', 'F', 'B'], }) # Form the directed graph: # A-->B->C->D # | | | # +>E-+->F--+->G g = containers.Graph(directed=True) g.chain('A', 'B', 'C', 'D', 'G') g.chain('A', 'E', 'F', 'G') g.chain('B', 'F') self.assertEqual(g, {'A': ['B', 'E'], 'B': ['C', 'F'], 'C': ['D'], 'D': ['G'], 'E': ['F'], 'F': ['G'], }) def test_shortest_path(self): # Form the graph: # A--B--C--D # | |\ / # | | \ / # E--F--G g = containers.Graph() g.connect('A', ('B', 'E')) g.connect('B', ('C', 'F', 'G')) g.connect('D', ('C', 'G')) g.connect('E', 'F') g.connect('F', 'G') self.assertEqual(g.shortest_path('A', 'D'), ['A', 'B', 'C', 'D']) self.assertEqual(g.shortest_path('B', 'A'), ['B', 'A']) self.assertEqual(g.shortest_path('E', 'C'), ['E', 'A', 'B', 'C']) self.assertEqual(g.shortest_path('A', 'G'), ['A', 'B', 'G']) # Do the same test again to see if caching works. self.assertEqual(g.shortest_path('A', 'D'), ['A', 'B', 'C', 'D']) self.assertEqual(g.shortest_path('B', 'A'), ['B', 'A']) self.assertEqual(g.shortest_path('E', 'C'), ['E', 'A', 'B', 'C']) self.assertEqual(g.shortest_path('A', 'G'), ['A', 'B', 'G']) # Test invalid paths. self.assertRaises(KeyError, g.shortest_path, 'R', 'D') self.assertEqual(g.shortest_path('D', 'R'), None) def test_shortest_path_directed(self): # Form the graph: # A-->B->C->D # | | | # +>E-+->F--+->G g = containers.Graph(directed=True) g.connect('A', ('B', 'E')) g.connect('B', ('C', 'F')) g.connect('C', 'D') g.connect('D', 'G') g.connect('E', 'F') g.connect('F', 'G') self.assertEqual(g, {'A': ['B', 'E'], 'B': ['C', 'F'], 'C': ['D'], 'D': ['G'], 'E': ['F'], 'F': ['G'], }) self.assertEqual(g.shortest_path('A', 'D'), ['A', 'B', 'C', 'D']) self.assertEqual(g.shortest_path('B', 'A'), None) self.assertEqual(g.shortest_path('E', 'C'), None) self.assertEqual(g.shortest_path('A', 'G'), ['A', 'B', 'F', 'G']) class Artwork(object): def __init__(self, Name='', Artist='', Rate=''): self.Name = Name self.Artist = Artist self.Rate = Rate class IndexTests(unittest.TestCase): def test_Index(self): artworks = containers.Index('Name', 'Artist', 'Rate') for index in artworks: self.assertEqual(index in ['Name', 'Artist', 'Rate'], True) # Store a single object. guernica = Artwork('Guernica', 'Pablo Picasso', 185.00) artworks.store(guernica) self.assertEqual(artworks.bucket('Name', 'Guernica'), [guernica]) # Add another that should collide on an indexed attribute. frenzy = Artwork('Frenzy', 'Pablo Picasso', 203.45) artworks.store(frenzy) self.assertEqual(artworks.bucket('Artist', 'Pablo Picasso'), [guernica, frenzy]) # Add a bunch. magrittes = [Artwork('Empire of Light', 'Rene Magritte', 512.13), Artwork('The Rape', 'Rene Magritte', 121.99), Artwork('The Son of Man', 'Rene Magritte', 1024.33), Artwork('The Difficult Crossing', 'Rene Magritte', 0.32), ] for work in magrittes: artworks.store(work) # Manually find all artworks rated under 200 cheapos = [] for rate in artworks.handles('Rate'): if rate < 200: cheapos.extend(artworks.bucket('Rate', rate)) expected = [magrittes[1], guernica, magrittes[3]] self.assertEqual(len(cheapos), len(expected)) for work in expected: self.assertEqual(work in cheapos, True) # Retrieve() all artworks rated over 200. assets = artworks.retrieve('Rate', operator.gt, 200) expected = [magrittes[2], frenzy, magrittes[0]] self.assertEqual(len(assets), len(expected)) for work in expected: self.assertEqual(work in assets, True) # Retrieve() all artworks. Use two different attributes # and compare the results. byrate = artworks.retrieve('Rate') byartist = artworks.retrieve('Artist') self.assertEqual(len(byrate), len(byartist)) for work in byrate: self.assertEqual(work in byartist, True) # Delete a single artwork. artworks.remove(guernica) self.assertEqual(artworks.bucket('Artist', 'Pablo Picasso'), [frenzy]) self.assertRaises(KeyError, artworks.bucket, 'Name', 'Guernica') # Change an indexed value of an artwork. guernica.Rate = 300.0 artworks.reindex(guernica, 'Rate') self.assertEqual(artworks.retrieve('Rate', operator.eq, 300), [guernica]) # Reindex with the known last value. guernica.Rate = 135.0 artworks.reindex(guernica, 'Rate', 300.0) self.assertEqual(artworks.retrieve('Rate', operator.eq, 135), [guernica]) self.assertEqual(artworks.retrieve('Rate', operator.eq, 300), []) self.assertRaises(KeyError, artworks.bucket, 'Rate', 300.0) class PrismTests(unittest.TestCase): def test_Prism(self): artworks = containers.Prism('Name', 'Artist', 'Rate') for index in artworks.facets.keys(): self.assertEqual(index in ['Name', 'Artist', 'Rate'], True) # Store a single row. artworks.add(Name='Guernica', Artist='Pablo Picasso', Rate=185.00) self.assertEqual(artworks.Artist(Name='Guernica'), 'Pablo Picasso') # Add another that should collide on an indexed attribute. artworks.add(Name='Frenzy', Artist='Pablo Picasso', Rate=203.45) self.assertEqual(artworks.Name(Artist='Pablo Picasso'), 'Guernica') self.assertEqual(artworks.Rate(Artist='Pablo Picasso'), 185.00) # Add a row with missing facets. artworks.add(Name='Empire of Light', Artist='Rene Magritte') self.assertEqual(artworks.Rate(Artist='Rene Magritte'), None) # Fill in the missing value. i = artworks.row_number(Name='Empire of Light') artworks.facets['Rate'][i] = 512.13 self.assertEqual(artworks.Rate(Artist='Rene Magritte'), 512.13) if __name__ == "__main__": unittest.main()