root/trunk/test_containers.py

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()