testbase.echo = False
def suite():
- modules_to_test = ('attributes', 'historyarray', 'pool', 'engines', 'query', 'sequence', 'select', 'types', 'mapper', 'objectstore', 'manytomany', 'dependency')
-# modules_to_test = ('engines', 'mapper')
+ modules_to_test = (
+ # core utilities
+ 'historyarray',
+ 'attributes',
+ 'dependency',
+
+ # connectivity
+ 'pool',
+
+ # schema/tables
+ 'engines',
+ 'types',
+
+ # SQL syntax
+ 'select',
+ 'selectable',
+
+ # assorted round-trip tests
+ 'query',
+
+ # sequences (postgres/oracle)
+ 'sequence',
+
+ # ORM selecting
+ 'mapper',
+
+ # ORM persistence
+ 'objectstore',
+
+ # cyclical ORM persistence
+ 'cycles',
+
+ # more select/persistence, backrefs
+ 'manytomany',
+ #'onetoone',
+
+ # extensions
+ 'proxy_engine',
+ 'wsgi_test',
+
+ )
+
alltests = unittest.TestSuite()
for module in map(__import__, modules_to_test):
alltests.addTest(unittest.findTestCases(module, suiteClass=None))
return alltests
+import sys
+sys.stdout = sys.stderr
+
if __name__ == '__main__':
testbase.runTests(suite())
from tables import *
import tables
-objectstore.LOG = True
+# TODO: need assertion conditions in this suite
+
"""test cyclical mapper relationships. No assertions yet, but run it with postgres and the
foreign key checks alone will usually not work if something is wrong"""
m1 = mapper(C1, t1, properties = {
'c1s' : relation(C1, private=True),
- 'c2s' : relation(C2, t2, private=True)
+ 'c2s' : relation(mapper(C2, t2), private=True)
})
a = C1('head c1')
t1.drop()
def setUp(self):
objectstore.clear()
- objectstore.LOG = True
+ #objectstore.LOG = True
clear_mappers()
def testcycle(self):
def setUp(self):
objectstore.clear()
- objectstore.LOG = True
+ #objectstore.LOG = True
clear_mappers()
+ def tearDownAll(self):
+ t3.drop()
+ t1.drop()
+ t2.drop()
+
def testcycle(self):
class C1(object):pass
class C2(object):pass
m2 = mapper(C2, t2)
m1 = mapper(C1, t1, properties = {
'c2s' : relation(m2, primaryjoin=t1.c.c2==t2.c.c1, uselist=True),
- 'data' : relation(C1Data, t3)
+ 'data' : relation(mapper(C1Data, t3))
})
m2.add_property('c1s', relation(m1, primaryjoin=t2.c.c2==t1.c.c1, uselist=True))
objectstore.delete(d)
objectstore.delete(c)
objectstore.commit()
+
+class CycleTest2(AssertMixin):
+ def setUpAll(self):
+ testbase.db.tables.clear()
+ global person
+ global ball
+ person = Table('person', db,
+ Column('id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
+ Column('favoriteBall_id', Integer, ForeignKey('ball.id')),
+ )
+ ball = Table('ball', db,
+ Column('id', Integer, Sequence('ball_id_seq', optional=True), primary_key=True),
+ Column('person_id', Integer),
+ )
+
+ person.create()
+ ball.create()
+ ball.c.person_id.append_item(ForeignKey('person.id'))
+
+ def tearDownAll(self):
+ ball.drop()
+ person.drop()
+
+ def setUp(self):
+ objectstore.clear()
+ #objectstore.LOG = True
+ clear_mappers()
+
+ def testcycle(self):
+ """this test has a peculiar aspect in that it doesnt create as many dependent
+ relationships as the other tests, and revealed a small glitch in the circular dependency sorting."""
+ class Person(object):
+ pass
+
+ class Ball(object):
+ pass
+
+ Ball.mapper = mapper(Ball, ball)
+ Person.mapper = mapper(Person, person, properties= dict(
+ balls = relation(Ball.mapper, primaryjoin=ball.c.person_id==person.c.id, foreignkey=ball.c.person_id),
+ favorateBall = relation(Ball.mapper, primaryjoin=person.c.favoriteBall_id==ball.c.id, foreignkey=person.c.favoriteBall_id),
+ )
+ )
+
+ print str(Person.mapper.props['balls'].primaryjoin)
+
+ b = Ball()
+ p = Person()
+ p.balls.append(b)
+ objectstore.commit()
+
+
if __name__ == "__main__":
testbase.main()
import sqlalchemy.mapping.topological as topological
import unittest, sys, os
+
+# TODO: need assertion conditions in this suite
+
+
class DependencySorter(topological.QueueDependencySorter):pass
class thingy(object):
def __init__(self, name=None):
self.name = name
+class PlaceThingy(object):
+ '''represents a thingy attached to a Place'''
+ def __init__(self, name=None):
+ self.name = name
+
class Transition(object):
'''represents a transition'''
def __init__(self, name=None):
Column('name', String(30), nullable=False),
)
+ global place_thingy
+ place_thingy = Table('place_thingy', db,
+ Column('thingy_id', Integer, Sequence('thid_seq', optional=True), primary_key=True),
+ Column('place_id', Integer, ForeignKey('place.place_id'), nullable=False),
+ Column('name', String(30), nullable=False)
+ )
+
# association table #1
global place_input
place_input = Table('place_input', db,
transition.create()
place_input.create()
place_output.create()
+ place_thingy.create()
def tearDownAll(self):
place_input.drop()
"""tests that a mapper can have two eager relations to the same table, via
two different association tables. aliases are required."""
- Place.mapper = mapper(Place, place)
+ Place.mapper = mapper(Place, place, properties = {
+ 'thingies':relation(mapper(PlaceThingy, place_thingy), lazy=False)
+ })
+
Transition.mapper = mapper(Transition, transition, properties = dict(
inputs = relation(Place.mapper, place_output, lazy=False, selectalias='op_alias'),
outputs = relation(Place.mapper, place_input, lazy=False, selectalias='ip_alias'),
db.echo = False
tables.drop()
db.echo = testbase.echo
- def setUp(self):
+ def tearDown(self):
objectstore.clear()
clear_mappers()
-
+ def setUp(self):
+ pass
class MapperTest(MapperSuperTest):
def testget(self):
])
objectstore.clear()
m3 = m2.options(undefer('user_id'))
- print m3.hashkey
def go():
l = m3.select()
print l[3].user_id
)
]
)
-
+ return
objects[2].keywords.append(k)
dkid = objects[5].keywords[1].keyword_id
del objects[5].keywords[1]
keywordmapper = mapper(Keyword, keywords)
+ # note that we are breaking a rule here, making a second mapper(Keyword, keywords)
+ # the reorganization of mapper construction affected this, but was fixed again
m = mapper(Item, items, properties = dict(
keywords = relation(mapper(IKAssociation, itemkeywords, properties = dict(
keyword = relation(mapper(Keyword, keywords), lazy = False, uselist = False)
from testbase import PersistTest
import testbase
+import os
#
# Define an engine, table and mapper at the module level, to show that the
# table and mapper can be used with different real engines in multiple threads
#
-module_engine = ProxyEngine()
+
+module_engine = ProxyEngine()
users = Table('users', module_engine,
Column('user_id', Integer, primary_key=True),
Column('user_name', String(16)),
class User(object):
pass
-assign_mapper(User, users)
-class ProxyEngineTest(PersistTest):
+class ProxyEngineTest1(PersistTest):
def setUp(self):
+ clear_mappers()
objectstore.clear()
def test_engine_connect(self):
# connect to a real engine
module_engine.connect(testbase.db_uri)
users.create()
+ assign_mapper(User, users)
+ try:
+ objectstore.begin()
- objectstore.begin()
+ user = User()
+ user.user_name='fred'
+ user.password='*'
+ objectstore.commit()
- user = User()
- user.user_name='fred'
- user.password='*'
- objectstore.commit()
-
- # select
- sqluser = User.select_by(user_name='fred')[0]
- assert sqluser.user_name == 'fred'
-
- # modify
- sqluser.user_name = 'fred jones'
+ # select
+ sqluser = User.select_by(user_name='fred')[0]
+ assert sqluser.user_name == 'fred'
- # commit - saves everything that changed
- objectstore.commit()
+ # modify
+ sqluser.user_name = 'fred jones'
+
+ # commit - saves everything that changed
+ objectstore.commit()
+
+ allusers = [ user.user_name for user in User.select() ]
+ assert allusers == [ 'fred jones' ]
+ finally:
+ users.drop()
- allusers = [ user.user_name for user in User.select() ]
- assert allusers == [ 'fred jones' ]
- users.drop()
+class ThreadProxyTest(PersistTest):
+ def setUp(self):
+ assign_mapper(User, users)
+ def tearDown(self):
+ clear_mappers()
+ def tearDownAll(self):
+ pass
+ os.remove('threadtesta.db')
+ os.remove('threadtestb.db')
def test_multi_thread(self):
+
from threading import Thread
from Queue import Queue
qb = Queue()
def run(db_uri, uname, queue):
def test():
+
try:
module_engine.connect(db_uri)
users.create()
-
- objectstore.begin()
-
- all = User.select()[:]
- assert all == []
-
- u = User()
- u.user_name = uname
- u.password = 'whatever'
- objectstore.commit()
-
- names = [ us.user_name for us in User.select() ]
- assert names == [ uname ]
- users.drop()
-
+ try:
+ objectstore.begin()
+
+ all = User.select()[:]
+ assert all == []
+
+ u = User()
+ u.user_name = uname
+ u.password = 'whatever'
+ objectstore.commit()
+
+ names = [ us.user_name for us in User.select() ]
+ assert names == [ uname ]
+ finally:
+ users.drop()
+ module_engine.dispose()
except Exception, e:
import traceback
traceback.print_exc()
if res != False:
raise res
+class ProxyEngineTest2(PersistTest):
+
+ def setUp(self):
+ clear_mappers()
+ objectstore.clear()
+
def test_table_singleton_a(self):
"""set up for table singleton check
"""
cats.create()
cats.drop()
- ProxyEngineTest.cats_table_a = cats
+ ProxyEngineTest2.cats_table_a = cats
assert isinstance(cats, Table)
def test_table_singleton_b(self):
cats = Table('cats', engine,
Column('cat_id', Integer, primary_key=True),
Column('cat_name', String))
- assert id(cats) != id(ProxyEngineTest.cats_table_a)
+ assert id(cats) != id(ProxyEngineTest2.cats_table_a)
# the real test -- if we're still using the old engine reference,
# this will fail because the old reference's local storage will
orders.drop()
addresses.drop()
users.drop()
+ db.commit()
def delete():
itemkeywords.delete().execute()
for key in item.keys():
ckey = self.convert_statement(key)
item[ckey] = item[key]
- if ckey != key:
+ if ckey != key:
del item[key]
item['_converted'] = True
try:
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
