root/trunk/storage/zoo_fixture.py

"""Unit tests for Storage Managers.

Don't run this directly; call run_tests() from a script for a specific
Storage Manager, setting up and tearing down that SM as needed.
See test_store*.py in this directory.
"""

import time
import unittest
import threading
import datetime
import dejavu
from dejavu import logic, zoo


class ZooTests(unittest.TestCase):
    
    def test_0_populate(self):
        box = zoo.arena.new_sandbox()
        
        # Notice this also tests that: a Unit which is only
        # dirtied via __init__ is still saved.
        WAP = zoo.Zoo(Name = 'Wild Animal Park',
                      Founded = datetime.date(2000, 1, 1),
                      # 59 can give rounding errors with divmod, which
                      # AdapterFromADO needs to correct.
                      Opens = datetime.time(8, 15, 59),
                      LastEscape = datetime.datetime(2004, 7, 29, 5, 6, 7),
                      Admission = "4.95",
                      )
        box.memorize(WAP)
        
        SDZ = zoo.Zoo(Name = 'San Diego Zoo',
                      # This early date should play havoc with a number
                      # of implementations.
                      Founded = datetime.date(1835, 9, 13),
                      Opens = datetime.time(9, 0, 0),
                      Admission = "0",
                      )
        box.memorize(SDZ)
        
        Biodome = zoo.Zoo(Name = u'Montr\xe9al Biod\xf4me',
                          Founded = datetime.date(1992, 6, 19),
                          Opens = datetime.time(9, 0, 0),
                          Admission = "11.75",
                          )
        
        leopard = zoo.Animal(Name='Leopard', Legs=4)
        self.assertEqual(leopard.PreviousZoos, None)
        box.memorize(leopard)
        leopard.add(WAP)
        leopard.LastEscape = datetime.datetime(2004, 12, 21, 8, 15, 0)
        
        box.memorize(zoo.Animal(Name='Slug', Legs=1))
        box.memorize(zoo.Animal(Name='Tiger', Legs=4))
        box.memorize(zoo.Animal(Name='Lion', Legs=4))
        box.memorize(zoo.Animal(Name='Bear', Legs=4))
        # Notice that ostrich.PreviousZoos is [], whereas leopard is None.
        box.memorize(zoo.Animal(Name='Ostrich', Legs=2, PreviousZoos=[]))
        box.memorize(zoo.Animal(Name='Centipede', Legs=100))
        
        millipede = zoo.Animal(Name='Millipede', Legs=1000000)
        millipede.add(SDZ)
        millipede.PreviousZoos = [WAP.ID]
        box.memorize(millipede)
        
        box.flush_all()
    
    def test_1_Object_Properties(self):
        box = zoo.arena.new_sandbox()
        
        WAP = box.unit(zoo.Zoo, Name='Wild Animal Park')
        self.assertNotEqual(WAP, None)
        self.assertEqual(WAP.Founded, datetime.date(2000, 1, 1))
        self.assertEqual(WAP.Opens, datetime.time(8, 15, 59))
        # This should have been updated when leopard.LastEscape was set.
        self.assertEqual(WAP.LastEscape,
                         datetime.datetime(2004, 12, 21, 8, 15, 0))
        self.assertEqual(str(WAP.Admission), "4.95")
        
        SDZ = box.unit(zoo.Zoo, Founded=datetime.date(1835, 9, 13))
        self.assertNotEqual(SDZ, None)
        self.assertEqual(SDZ.Founded, datetime.date(1835, 9, 13))
        self.assertEqual(SDZ.Opens, datetime.time(9, 0, 0))
        self.assertEqual(SDZ.LastEscape, None)
        self.assertEqual(float(SDZ.Admission), 0)
        
        leopard = box.unit(zoo.Animal, Name='Leopard')
        self.assertEqual(leopard.Name, 'Leopard')
        self.assertEqual(leopard.Legs, 4)
        self.assertEqual(leopard.ZooID, WAP.ID)
        self.assertEqual(leopard.PreviousZoos, None)
        self.assertEqual(leopard.LastEscape,
                         datetime.datetime(2004, 12, 21, 8, 15, 0))
        
        ostrich = box.unit(zoo.Animal, Name='Ostrich')
        self.assertEqual(ostrich.Name, 'Ostrich')
        self.assertEqual(ostrich.Legs, 2)
        self.assertEqual(ostrich.ZooID, None)
        self.assertEqual(ostrich.PreviousZoos, [])
        self.assertEqual(ostrich.LastEscape, None)
        
        millipede = box.unit(zoo.Animal, Legs=1000000)
        self.assertEqual(millipede.Name, 'Millipede')
        self.assertEqual(millipede.Legs, 1000000)
        self.assertEqual(millipede.ZooID, SDZ.ID)
        self.assertEqual(millipede.PreviousZoos, [WAP.ID])
        self.assertEqual(millipede.LastEscape, None)
        box.flush_all()
    
    def test_2_Expressions(self):
        box = zoo.arena.new_sandbox()
        
        def matches(lam):
            # We flush_all to ensure a DB hit each time.
            box.flush_all()
            units = box.recall(zoo.Animal, logic.Expression(lam))
            return len([x for x in units])
        
        zoos = [x for x in box.recall(zoo.Zoo)]
        self.assertEqual(zoos[0].dirty(), False)
        self.assertEqual(len(zoos), 2)
        self.assertEqual(matches(lambda x: True), 8)
        self.assertEqual(matches(lambda x: x.Legs == 4), 4)
        self.assertEqual(matches(lambda x: x.Legs == 2), 1)
        self.assertEqual(matches(lambda x: x.Legs >= 2 and x.Legs < 20), 5)
        self.assertEqual(matches(lambda x: x.Legs > 10), 2)
        self.assertEqual(matches(lambda x: x.Name.startswith('L')), 2)
        self.assertEqual(matches(lambda x: x.Name.endswith('pede')), 2)
        self.assertEqual(matches(lambda x: x.LastEscape != None), 1)
        self.assertEqual(matches(lambda x: None == x.LastEscape), 7)
##        self.assertRaises(ValueError, matches, lambda x: None is x.LastEscape)
        
        # In operator (containedby)
        self.assertEqual(matches(lambda x: 'pede' in x.Name), 2)
        self.assertEqual(matches(lambda x: x.Name in ('Lion', 'Tiger', 'Bear')), 3)
        
        # Try In with cell references
        class thing(object): pass
        pet, pet2 = thing(), thing()
        pet.Name, pet2.Name = 'Slug', 'Ostrich'
        self.assertEqual(matches(lambda x: x.Name in (pet.Name, pet2.Name)), 2)
        
        # logic and other functions
        self.assertEqual(matches(lambda x: dejavu.ieq(x.Name, 'slug')), 1)
        self.assertEqual(matches(lambda x: dejavu.icontains(x.Name, 'PEDE')), 2)
        self.assertEqual(matches(lambda x: dejavu.icontains(('Lion', 'Banana'), x.Name)), 1)
        self.assertEqual(matches(lambda x: dejavu.icontainedby(x.Name, ('Lion', 'Bear', 'Leopard'))), 3)
        name = 'Lion'
        self.assertEqual(matches(lambda x: len(x.Name) == len(name)), 3)
        
        # This broke sometime in 2004. Rev 32 seems to have fixed it.
        self.assertEqual(matches(lambda x: 'i' in x.Name), 5)
        
        # Test now(), today(), year()
        self.assertEqual(matches(lambda x: x.LastEscape != None
                                 and x.LastEscape < dejavu.today()), 1)
        self.assertEqual(matches(lambda x: x.LastEscape == dejavu.now()), 0)
        self.assertEqual(matches(lambda x: dejavu.year(x.LastEscape) == 2004), 1)
        
        # Test AND, OR with cannot_represent.
        # Notice that we reference a method ('count') which no
        # known SM handles, so it will default back to Expr.eval().
        self.assertEqual(matches(lambda x: 'p' in x.Name
                                 and x.Name.count('e') > 1), 2)
    
    def test_3_Aggregates(self):
        box = zoo.arena.new_sandbox()
        legs = box.distinct(zoo.Animal, ['Legs'])
        legs.sort()
        self.assertEqual(legs, [1, 2, 4, 100, 1000000])
    
    def test_4_Multithreading(self):
        f = logic.Expression(lambda x: x.Legs == 4)
        def thread_recall():
            # Notice we only do reads in this thread, not writes, since
            # the order of thread execution can not be guaranteed.
            box = zoo.arena.new_sandbox()
            quadrupeds = [x for x in box.recall(zoo.Animal, f)]
            self.assertEqual(len(quadrupeds), 4)
        
        ts = []
        # PostgreSQL, for example, has a default max_connections of 100.
        for x in range(99):
            t = threading.Thread(target=thread_recall)
            t.start()
            ts.append(t)
        for t in ts:
            t.join()


def setup_SM(SM_class, opts):
    """main(SM_class, PreviousZoos) -> Set up the arena and SM for Zoo."""
    if isinstance(SM_class, basestring):
        import xray
        SM_class = xray.classes(SM_class)
    testSM = SM_class("test", zoo.arena, opts)
    zoo.arena.add_store('testSM', testSM)
    
    for cls in (zoo.Animal, zoo.Zoo, zoo.Exhibit):
        zoo.arena.create_storage(cls)

def teardown():
    zoo.arena.shutdown()
    for store in zoo.arena.stores.values():
        try:
            store.drop_database()
        except NotImplementedError:
            pass

def run_tests(SM_class, opts):
    import traceback
    try:
        try:
            setup_SM(SM_class, opts)
            unittest.main(__name__)
        except SystemExit:
            # unittest.main normally raises SystemExit when complete.
            pass
        except:
            traceback.print_exc()
            print
    finally:
        teardown()