summaryrefslogtreecommitdiff
path: root/test/ext
diff options
context:
space:
mode:
Diffstat (limited to 'test/ext')
-rw-r--r--test/ext/__init__.py0
-rw-r--r--test/ext/activemapper.py237
-rw-r--r--test/ext/alltests.py19
-rw-r--r--test/ext/legacy_objectstore.py113
-rw-r--r--test/ext/sqlsoup.py19
-rw-r--r--test/ext/wsgi_test.py122
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')})
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
View Lorry Controller Status