reorganized unit tests into subdirectories

6 files changed, 510 insertions, 0 deletions

diff --git a/test/ext/__init__.py b/test/ext/__init__.py
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/test/ext/__init__.py
diff --git a/test/ext/activemapper.py b/test/ext/activemapper.py
new file mode 100644
index 000000000..6a8b0904e
--- /dev/null
+++ b/test/ext/activemapper.py
@@ -0,0 +1,237 @@
+from sqlalchemy.ext.activemapper           import ActiveMapper, column, one_to_many, one_to_one, objectstore
+from sqlalchemy             import and_, or_, clear_mappers
+from sqlalchemy             import ForeignKey, String, Integer, DateTime
+from datetime               import datetime
+
+import unittest
+import sqlalchemy.ext.activemapper as activemapper
+
+import testbase
+
+class testcase(testbase.PersistTest):
+    def setUpAll(self):
+        global Person, Preferences, Address
+        
+        class Person(ActiveMapper):
+            class mapping:
+                id          = column(Integer, primary_key=True)
+                full_name   = column(String)
+                first_name  = column(String)
+                middle_name = column(String)
+                last_name   = column(String)
+                birth_date  = column(DateTime)
+                ssn         = column(String)
+                gender      = column(String)
+                home_phone  = column(String)
+                cell_phone  = column(String)
+                work_phone  = column(String)
+                prefs_id    = column(Integer, foreign_key=ForeignKey('preferences.id'))
+                addresses   = one_to_many('Address', colname='person_id', backref='person')
+                preferences = one_to_one('Preferences', colname='pref_id', backref='person')
+
+            def __str__(self):
+                s =  '%s\n' % self.full_name
+                s += '  * birthdate: %s\n' % (self.birth_date or 'not provided')
+                s += '  * fave color: %s\n' % (self.preferences.favorite_color or 'Unknown')
+                s += '  * personality: %s\n' % (self.preferences.personality_type or 'Unknown')
+
+                for address in self.addresses:
+                    s += '  * address: %s\n' % address.address_1
+                    s += '             %s, %s %s\n' % (address.city, address.state, address.postal_code)
+
+                return s
+
+        class Preferences(ActiveMapper):
+            class mapping:
+                __table__        = 'preferences'
+                id               = column(Integer, primary_key=True)
+                favorite_color   = column(String)
+                personality_type = column(String)
+
+        class Address(ActiveMapper):
+            class mapping:
+                id          = column(Integer, primary_key=True)
+                type        = column(String)
+                address_1   = column(String)
+                city        = column(String)
+                state       = column(String)
+                postal_code = column(String)
+                person_id   = column(Integer, foreign_key=ForeignKey('person.id'))
+
+        activemapper.metadata.connect(testbase.db)
+        activemapper.create_tables()
+
+    def tearDownAll(self):
+        clear_mappers()
+        activemapper.drop_tables()
+        
+    def tearDown(self):
+        for t in activemapper.metadata.table_iterator(reverse=True):
+            t.delete().execute()
+        #people = Person.select()
+        #for person in people: person.delete()
+        
+        #addresses = Address.select()
+        #for address in addresses: address.delete()
+        
+        #preferences = Preferences.select()
+        #for preference in preferences: preference.delete()
+        
+        #objectstore.flush()
+        #objectstore.clear()
+    
+    def create_person_one(self):
+        # create a person
+        p1 = Person(
+                full_name='Jonathan LaCour',
+                birth_date=datetime(1979, 10, 12),
+                preferences=Preferences(
+                                favorite_color='Green',
+                                personality_type='ENTP'
+                            ),
+                addresses=[
+                    Address(
+                        address_1='123 Some Great Road.',
+                        city='Atlanta',
+                        state='GA',
+                        postal_code='30338'
+                    ),
+                    Address(
+                        address_1='435 Franklin Road.',
+                        city='Atlanta',
+                        state='GA',
+                        postal_code='30342'
+                    )
+                ]
+             )
+        return p1
+    
+    
+    def create_person_two(self):
+        p2 = Person(
+                full_name='Lacey LaCour',
+                addresses=[
+                    Address(
+                        address_1='123 Some Great Road.',
+                        city='Atlanta',
+                        state='GA',
+                        postal_code='30338'
+                    ),
+                    Address(
+                        address_1='200 Main Street',
+                        city='Roswell',
+                        state='GA',
+                        postal_code='30075'
+                    )
+                ]
+             )
+        # I don't like that I have to do this... and putting
+        # a "self.preferences = Preferences()" into the __init__
+        # of Person also doens't seem to fix this
+        p2.preferences = Preferences()
+        
+        return p2
+    
+    
+    def test_create(self):
+        p1 = self.create_person_one()
+        objectstore.flush()
+        objectstore.clear()
+        
+        results = Person.select()
+        
+        self.assertEquals(len(results), 1)
+        
+        person = results[0]
+        self.assertEquals(person.id, p1.id)
+        self.assertEquals(len(person.addresses), 2)
+        self.assertEquals(person.addresses[0].postal_code, '30338')
+    
+    
+    def test_delete(self):
+        p1 = self.create_person_one()
+        
+        objectstore.flush()
+        objectstore.clear()
+        
+        results = Person.select()
+        self.assertEquals(len(results), 1)
+        
+        results[0].delete()
+        objectstore.flush()
+        objectstore.clear()
+        
+        results = Person.select()
+        self.assertEquals(len(results), 0)
+    
+    
+    def test_multiple(self):
+        p1 = self.create_person_one()
+        p2 = self.create_person_two()
+        
+        objectstore.flush()
+        objectstore.clear()
+        
+        # select and make sure we get back two results
+        people = Person.select()
+        self.assertEquals(len(people), 2)
+                
+        # make sure that our backwards relationships work
+        self.assertEquals(people[0].addresses[0].person.id, p1.id)
+        self.assertEquals(people[1].addresses[0].person.id, p2.id)
+        
+        # try a more complex select
+        results = Person.select(
+            or_(
+                and_(
+                    Address.c.person_id == Person.c.id,
+                    Address.c.postal_code.like('30075')
+                ),
+                and_(
+                    Person.c.prefs_id == Preferences.c.id,
+                    Preferences.c.favorite_color == 'Green'
+                )
+            )
+        )
+        self.assertEquals(len(results), 2)
+        
+    
+    def test_oneway_backref(self):
+        # FIXME: I don't know why, but it seems that my backwards relationship
+        #        on preferences still ends up being a list even though I pass
+        #        in uselist=False...
+        # FIXED: the backref is a new PropertyLoader which needs its own "uselist".
+        # uses a function which I dont think existed when you first wrote ActiveMapper.
+        p1 = self.create_person_one()
+        self.assertEquals(p1.preferences.person, p1)
+        p1.delete()
+        
+        objectstore.flush()
+        objectstore.clear()
+    
+    
+    def test_select_by(self):
+        # FIXME: either I don't understand select_by, or it doesn't work.
+        # FIXED (as good as we can for now): yup....everyone thinks it works that way....it only
+        # generates joins for keyword arguments, not ColumnClause args.  would need a new layer of
+        # "MapperClause" objects to use properties in expressions. (MB)
+        
+        p1 = self.create_person_one()
+        p2 = self.create_person_two()
+        
+        objectstore.flush()
+        objectstore.clear()
+        
+        results = Person.select(
+            Address.c.postal_code.like('30075') &
+            Person.join_to('addresses')
+        )
+        self.assertEquals(len(results), 1)
+
+
+    
+if __name__ == '__main__':
+    # go ahead and setup the database connection, and create the tables
+    
+    # launch the unit tests
+    unittest.main()
\ No newline at end of file
diff --git a/test/ext/alltests.py b/test/ext/alltests.py
new file mode 100644
index 000000000..67513f032
--- /dev/null
+++ b/test/ext/alltests.py
@@ -0,0 +1,19 @@
+import testbase
+import unittest
+
+def suite():
+    modules_to_test = (
+	'ext.activemapper',
+	'ext.sqlsoup'
+        )
+    alltests = unittest.TestSuite()
+    for name in modules_to_test:
+        mod = __import__(name)
+        for token in name.split('.')[1:]:
+            mod = getattr(mod, token)
+        alltests.addTest(unittest.findTestCases(mod, suiteClass=None))
+    return alltests
+
+
+if __name__ == '__main__':
+    testbase.runTests(suite())
diff --git a/test/ext/legacy_objectstore.py b/test/ext/legacy_objectstore.py
new file mode 100644
index 000000000..3aa99a1ae
--- /dev/null
+++ b/test/ext/legacy_objectstore.py
@@ -0,0 +1,113 @@
+from testbase import PersistTest, AssertMixin
+import unittest, sys, os
+from sqlalchemy import *
+import StringIO
+import testbase
+
+from tables import *
+import tables
+
+install_mods('legacy_session')
+
+
+class LegacySessionTest(AssertMixin):
+    def setUpAll(self):
+        db.echo = False
+        users.create()
+        db.echo = testbase.echo
+    def tearDownAll(self):
+        db.echo = False
+        users.drop()
+        db.echo = testbase.echo
+    def setUp(self):
+        objectstore.get_session().clear()
+        clear_mappers()
+        tables.user_data()
+        #db.echo = "debug"
+    def tearDown(self):
+        tables.delete_user_data()
+        
+    def test_nested_begin_commit(self):
+        """tests that nesting objectstore transactions with multiple commits
+        affects only the outermost transaction"""
+        class User(object):pass
+        m = mapper(User, users)
+        def name_of(id):
+            return users.select(users.c.user_id == id).execute().fetchone().user_name
+        name1 = "Oliver Twist"
+        name2 = 'Mr. Bumble'
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        s = objectstore.get_session()
+        trans = s.begin()
+        trans2 = s.begin()
+        m.get(7).user_name = name1
+        trans3 = s.begin()
+        m.get(8).user_name = name2
+        trans3.commit()
+        s.commit() # should do nothing
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        trans2.commit()
+        s.commit()  # should do nothing
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        trans.commit()
+        self.assert_(name_of(7) == name1, msg="user_name should be %s" % name1)
+        self.assert_(name_of(8) == name2, msg="user_name should be %s" % name2)
+
+    def test_nested_rollback(self):
+        """tests that nesting objectstore transactions with a rollback inside
+        affects only the outermost transaction"""
+        class User(object):pass
+        m = mapper(User, users)
+        def name_of(id):
+            return users.select(users.c.user_id == id).execute().fetchone().user_name
+        name1 = "Oliver Twist"
+        name2 = 'Mr. Bumble'
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        s = objectstore.get_session()
+        trans = s.begin()
+        trans2 = s.begin()
+        m.get(7).user_name = name1
+        trans3 = s.begin()
+        m.get(8).user_name = name2
+        trans3.rollback()
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        trans2.commit()
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        trans.commit()
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+
+    def test_true_nested(self):
+        """tests creating a new Session inside a database transaction, in 
+        conjunction with an engine-level nested transaction, which uses
+        a second connection in order to achieve a nested transaction that commits, inside
+        of another engine session that rolls back."""
+#        testbase.db.echo='debug'
+        class User(object):
+            pass
+        testbase.db.begin()
+        try:
+            m = mapper(User, users)
+            name1 = "Oliver Twist"
+            name2 = 'Mr. Bumble'
+            m.get(7).user_name = name1
+            s = objectstore.Session(nest_on=testbase.db)
+            m.using(s).get(8).user_name = name2
+            s.commit()
+            objectstore.commit()
+            testbase.db.rollback()
+        except:
+            testbase.db.rollback()
+            raise
+        objectstore.clear()
+        self.assert_(m.get(8).user_name == name2)
+        self.assert_(m.get(7).user_name != name1)
+
+if __name__ == "__main__":
+    testbase.main()        
diff --git a/test/ext/sqlsoup.py b/test/ext/sqlsoup.py
new file mode 100644
index 000000000..4d4dbb69a
--- /dev/null
+++ b/test/ext/sqlsoup.py
@@ -0,0 +1,19 @@
+import testbase
+
+import sqlalchemy.ext.sqlsoup as sqlsoup
+
+class SqlSoupTest(testbase.AssertMixin):
+    def tearDown(self):
+        pass
+    def tearDownAll(self):
+        pass
+    def setUpAll(self):
+        pass
+    def setUp(self):
+        pass
+    def testall(self):
+        import doctest
+        doctest.testmod(m=sqlsoup,verbose=True)
+        
+if __name__ == "__main__":
+    testbase.main()        
diff --git a/test/ext/wsgi_test.py b/test/ext/wsgi_test.py
new file mode 100644
index 000000000..1330f88b6
--- /dev/null
+++ b/test/ext/wsgi_test.py
@@ -0,0 +1,122 @@
+"""Interactive wsgi test
+
+Small WSGI application that uses a table and mapper defined at the module
+level, with per-application uris enabled by the ProxyEngine.
+
+Requires the wsgiutils package from:
+
+http://www.owlfish.com/software/wsgiutils/
+
+Run the script with python wsgi_test.py, then visit http://localhost:8080/a
+and http://localhost:8080/b with a browser. You should see two distinct lists
+of colors.
+"""
+
+from sqlalchemy import *
+from sqlalchemy.ext.proxy import ProxyEngine
+from wsgiutils import wsgiServer
+
+engine = ProxyEngine()
+
+colors = Table('colors', engine,
+               Column('id', Integer, primary_key=True),
+               Column('name', String(32)),
+               Column('hex', String(6)))
+
+class Color(object):
+    pass
+
+assign_mapper(Color, colors)
+
+data = { 'a': (('fff','white'), ('aaa','gray'), ('000','black'),
+               ('f00', 'red'), ('0f0', 'green')),
+         'b': (('00f','blue'), ('ff0', 'yellow'), ('0ff','purple')) }
+
+db_uri = { 'a': 'sqlite://filename=wsgi_db_a.db',
+           'b': 'sqlite://filename=wsgi_db_b.db' }
+
+def app(dataset):
+    print '... connecting to database %s: %s' % (dataset, db_uri[dataset])
+    engine.connect(db_uri[dataset], echo=True, echo_pool=True)
+    colors.create()
+
+    print '... populating data into %s' % db_uri[dataset]
+    for hex, name in data[dataset]:
+        c = Color()
+        c.hex = hex
+        c.name = name
+    objectstore.commit()
+    objectstore.clear()
+    
+    def call(environ, start_response):
+        engine.connect(db_uri[dataset], echo=True, echo_pool=True)
+
+        # NOTE: must clear objectstore on each request, or you'll see
+        # objects from another thread here
+        objectstore.clear()
+        objectstore.begin()
+        
+        c = Color.select()
+
+        start_response('200 OK', [('content-type','text/html')])
+        yield '<html><head><title>Test dataset %s</title></head>' % dataset
+        yield '<body>'
+        yield '<p>uri: %s</p>' % db_uri[dataset]
+        yield '<p>engine: <xmp>%s</xmp></p>' % engine.engine
+        yield '<p>Colors!</p>'
+        for color in c:
+            yield '<div style="background: #%s">%s</div>' % (color.hex,
+                                                             color.name)
+        yield '</body></html>'
+    return call
+
+def cleanup():
+    for uri in db_uri.values():
+        print "Cleaning db %s" % uri
+        engine.connect(uri)
+        colors.drop()
+
+def run_server(apps, host='localhost', port=8080):
+    print "Serving test app at http://%s:%s/" % (host, port)
+    print "Visit http://%(host)s:%(port)s/a and " \
+        "http://%(host)s:%(port)s/b to test apps" % {'host': host,
+                                                     'port': port}
+    
+    server = wsgiServer.WSGIServer((host, port), apps, serveFiles=False)
+    try:
+        server.serve_forever()
+    except:
+        cleanup()
+        raise
+    
+if __name__ == '__main__':
+    run_server({'/a':app('a'), '/b':app('b')})
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+