import testbase
from sqlalchemy import *
from sqlalchemy.databases import mysql
+from testbase import Table, Column
import sys, StringIO
db = testbase.db
+++ /dev/null
-from testbase import PersistTest
-import testbase
-from sqlalchemy import *
-from sqlalchemy.ext.proxy import AutoConnectEngine
-
-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 = AutoConnectEngine( testbase.db_uri )
-users = Table('users', module_engine,
- Column('user_id', Integer, primary_key=True),
- Column('user_name', String(16)),
- Column('password', String(20))
- )
-
-class User(object):
- pass
-
-
-class AutoConnectEngineTest1(PersistTest):
-
- def setUp(self):
- clear_mappers()
- objectstore.clear()
-
- def test_engine_connect(self):
- users.create()
- assign_mapper(User, users)
- try:
- trans = objectstore.begin()
-
- user = User()
- user.user_name='fred'
- user.password='*'
- trans.commit()
-
- # select
- sqluser = User.select_by(user_name='fred')[0]
- assert sqluser.user_name == 'fred'
-
- # 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()
-
-
-
-
-if __name__ == "__main__":
- testbase.main()
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
import tables
db = testbase.db
from sqlalchemy import *
-
+from testbase import Table, Column
class ExecuteTest(testbase.PersistTest):
def setUpAll(self):
users = Table('users', metadata,
Column('user_id', INT, primary_key = True),
Column('user_name', VARCHAR(20)),
- mysql_engine='InnoDB'
)
metadata.create_all()
import testbase
from sqlalchemy import *
+from testbase import Table, Column
class MetaDataTest(testbase.PersistTest):
def test_metadata_connect(self):
metadata.drop_all()
if __name__ == '__main__':
- testbase.main()
\ No newline at end of file
+ testbase.main()
from sqlalchemy import *
from sqlalchemy.exceptions import NoSuchTableError
import sqlalchemy.databases.mysql as mysql
-
+from testbase import Table, Column
import unittest, re, StringIO
class ReflectionTest(PersistTest):
use_string_defaults = use_function_defaults or testbase.db.engine.__module__.endswith('sqlite')
+ if (testbase.db.engine.name == 'mysql' and
+ testbase.db.dialect.get_version_info(testbase.db) < (4, 1, 1)):
+ return
+
if use_function_defaults:
defval = func.current_date()
deftype = Date
Column('test_passivedefault4', deftype3, PassiveDefault(defval3)),
Column('test9', Binary(100)),
Column('test_numeric', Numeric(None, None)),
- mysql_engine='InnoDB'
+ test_needs_fk=True,
)
-
+
addresses = Table('engine_email_addresses', meta,
Column('address_id', Integer, primary_key = True),
Column('remote_user_id', Integer, ForeignKey(users.c.user_id)),
Column('email_address', String(20)),
- mysql_engine='InnoDB'
+ test_needs_fk=True,
)
meta.drop_all()
def test_composite_fk(self):
"""test reflection of composite foreign keys"""
+
+ if (testbase.db.engine.name == 'mysql' and
+ testbase.db.dialect.get_version_info(testbase.db) < (4, 1, 1)):
+ return
meta = BoundMetaData(testbase.db)
table = Table(
'multi', meta,
Column('multi_hoho', Integer, primary_key=True),
Column('name', String(50), nullable=False),
Column('val', String(100)),
- mysql_engine='InnoDB'
+ test_needs_fk=True,
)
table2 = Table('multi2', meta,
Column('id', Integer, primary_key=True),
Column('lala', Integer),
Column('data', String(50)),
ForeignKeyConstraint(['foo', 'bar', 'lala'], ['multi.multi_id', 'multi.multi_rev', 'multi.multi_hoho']),
- mysql_engine='InnoDB'
+ test_needs_fk=True,
)
assert table.c.multi_hoho
meta.create_all()
db = testbase.db
from sqlalchemy import *
from sqlalchemy.orm import *
-
+from testbase import Table, Column
class TransactionTest(testbase.PersistTest):
def setUpAll(self):
users = Table('query_users', metadata,
Column('user_id', INT, primary_key = True),
Column('user_name', VARCHAR(20)),
- mysql_engine='InnoDB'
+ test_needs_acid=True,
)
users.create(testbase.db)
users = Table('deadlock_users', metadata,
Column('user_id', INT, primary_key = True),
Column('user_name', VARCHAR(20)),
+ test_needs_acid=True,
)
users.create(conn1)
conn1.execute("select * from deadlock_users")
users = Table('query_users', metadata,
Column('user_id', INT, primary_key = True),
Column('user_name', VARCHAR(20)),
- mysql_engine='InnoDB'
+ test_needs_acid=True,
)
users.create(tlengine)
def tearDown(self):
counters = Table('forupdate_counters', metadata,
Column('counter_id', INT, primary_key = True),
Column('counter_value', INT),
- mysql_engine='InnoDB'
+ test_needs_acid=True,
)
counters.create(testbase.db)
def tearDown(self):
from sqlalchemy.ext.assignmapper import assign_mapper
from sqlalchemy.ext.sessioncontext import SessionContext
+from testbase import Table, Column
class OverrideAttributesTest(PersistTest):
def setUpAll(self):
assert SomeObject.get_by(id=s.id).options[0].id == sso.id
if __name__ == '__main__':
- testbase.main()
\ No newline at end of file
+ testbase.main()
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.associationproxy import *
+from testbase import Table, Column
db = testbase.db
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.orderinglist import *
+from testbase import Table, Column
db = testbase.db
metadata = None
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
from sqlalchemy.ext.selectresults import SelectResultsExt, SelectResults
+++ /dev/null
-"""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')})
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
from sqlalchemy import *
from sqlalchemy.orm import *
-
+from testbase import Table, Column
class AssociationTest(testbase.PersistTest):
def setUpAll(self):
from sqlalchemy.ext.sessioncontext import SessionContext
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
class O2MCascadeTest(testbase.AssertMixin):
def tearDown(self):
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
class CompileTest(testbase.AssertMixin):
from sqlalchemy.orm import *
import StringIO
import testbase
+from testbase import Table, Column
from tables import *
import tables
import unittest, sys, os
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
import datetime
class EagerTest(AssertMixin):
import unittest, sys, os
from sqlalchemy import *
from sqlalchemy.orm import *
-import datetime
from sqlalchemy.ext.sessioncontext import SessionContext
+from testbase import Table, Column
+import datetime
class EagerTest(AssertMixin):
def setUpAll(self):
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.selectresults import SelectResults
+from testbase import Table, Column
import random
class EagerTest(AssertMixin):
from testbase import PersistTest, AssertMixin
import unittest
+import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
-import testbase
from sqlalchemy.ext.sessioncontext import SessionContext
+from testbase import Table, Column
from tables import *
import tables
from sqlalchemy import *
+from testbase import Table, Column
class Base(object):
def __init__(self, **kwargs):
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy import exceptions
+from testbase import Table, Column
class Foo(object):
pass
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
from sqlalchemy.orm.sync import ONETOMANY, MANYTOONE
import testbase
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
class O2MTest(testbase.ORMTest):
from sqlalchemy import *
from sqlalchemy.orm import *
import testbase
+from testbase import Table, Column
class ConcreteTest1(testbase.ORMTest):
def define_tables(self, metadata):
if __name__ == '__main__':
- testbase.main()
\ No newline at end of file
+ testbase.main()
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
class BaseObject(object):
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
class InheritTest(testbase.ORMTest):
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
+
class PolymorphicCircularTest(testbase.ORMTest):
keep_mappers = True
+"""tests basic polymorphic mapper loading/saving, minimal relations"""
+
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
-
+from testbase import Table, Column
import sets
-# tests basic polymorphic mapper loading/saving, minimal relations
class Person(object):
def __init__(self, **kwargs):
from sqlalchemy import *
from sqlalchemy.orm import *
-
+from testbase import Table, Column
import testbase
+
class AttrSettable(object):
def __init__(self, **kwargs):
[setattr(self, k, v) for k, v in kwargs.iteritems()]
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
from datetime import datetime
from sqlalchemy import *
from sqlalchemy.orm import *
-
+from testbase import Table, Column
import testbase
+
class SingleInheritanceTest(testbase.AssertMixin):
def setUpAll(self):
metadata = BoundMetaData(testbase.db)
assert session.query(JuniorEngineer).select() == [e2]
if __name__ == '__main__':
- testbase.main()
\ No newline at end of file
+ testbase.main()
import unittest, sys, os
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
import datetime
class LazyTest(AssertMixin):
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
import string
class Place(object):
Mapper = mapperlib.Mapper
import gc
import testbase
+from testbase import Table, Column
import tables
class A(object):pass
import testbase
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
from tables import *
import tables
if __name__ == "__main__":
testbase.main()
-
\ No newline at end of file
+
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.sessioncontext import SessionContext
+from testbase import Table, Column
class Jack(object):
def __repr__(self):
from sqlalchemy import *
from sqlalchemy.orm import *
import testbase
+from testbase import Table, Column
from fixtures import *
class Base(object):
import testbase
import unittest, sys, datetime
-
-db = testbase.db
-#db.
-
from sqlalchemy import *
from sqlalchemy.orm import *
+from testbase import Table, Column
+db = testbase.db
class RelationTest(testbase.PersistTest):
"""this is essentially an extension of the "dependency.py" topological sort test.
from sqlalchemy.orm import *
import testbase
+from testbase import Table, Column
metadata = MetaData()
users = Table('users', metadata,
Column('user_id', Integer, Sequence('user_id_seq', optional=True), primary_key = True),
Column('user_name', String(40)),
- mysql_engine='innodb'
)
class SessionContextTest(AssertMixin):
from sqlalchemy import *
from sqlalchemy.orm import *
import testbase
+from testbase import Table, Column
import pickleable
from sqlalchemy.orm.mapper import global_extensions
from sqlalchemy.orm import util as ormutil
mytable = Table('mytable', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
- mysql_engine='InnoDB'
+ test_need_fk=True,
)
myothertable = Table('myothertable', metadata,
Column('parent_id', Integer),
Column('data', String(30)),
ForeignKeyConstraint(['parent_id'],['mytable.id'], ondelete="CASCADE"),
- mysql_engine='InnoDB'
+ test_need_fk=True,
)
metadata.create_all()
from sqlalchemy import *
+from testbase import Table, Column
from timeit import Timer
import sys
print "\nCreate backward associations"
create_it = tt.create_back_assoc
- tt.run(10)
\ No newline at end of file
+ tt.run(10)
from testbase import PersistTest, AssertMixin
import unittest, sys, os
from sqlalchemy import *
+from testbase import Table, Column
import StringIO
import testbase
import gc
import unittest, sys, os
from sqlalchemy import *
import sqlalchemy.orm.attributes as attributes
+from testbase import Table, Column
import StringIO
import testbase
import gc
except:
pass
from sqlalchemy import *
+from testbase import Table, Column
import time
metadata = create_engine('sqlite://', echo=True)
import unittest, sys, os
from sqlalchemy import *
import sqlalchemy.attributes as attributes
+from testbase import Table, Column
import StringIO
import testbase
import gc
from sqlalchemy import *
import sqlalchemy.pool as pool
+from testbase import Table, Column
import psycopg2 as psycopg
import thread,time
psycopg = pool.manage(psycopg,pool_size=2,max_overflow=1, timeout=5, echo=True)
thread.start_new_thread(run, ())
while True:
- time.sleep(5)
\ No newline at end of file
+ time.sleep(5)
from sqlalchemy import *
import thread, time, random
from sqlalchemy.orm import mapperlib
+from testbase import Table, Column
meta = BoundMetaData('sqlite:///foo.db')
import sqlalchemy.pool as pool
import thread
from sqlalchemy import exceptions
+from testbase import Table, Column
import logging
logging.basicConfig()
import sys
import testbase
from sqlalchemy import *
+from testbase import Table, Column
class CaseTest(testbase.PersistTest):
import testbase
from sqlalchemy import *
+from testbase import Table, Column
import sys
class ConstraintTest(testbase.AssertMixin):
import testbase
from sqlalchemy import *
from sqlalchemy.orm import mapper, create_session
+from testbase import Table, Column
import sqlalchemy
db = testbase.db
import testbase
-
from sqlalchemy import *
+from testbase import Table, Column
# TODO: either create a mock dialect with named paramstyle and a short identifier length,
# or find a way to just use sqlite dialect and make those changes
from sqlalchemy import *
from sqlalchemy.engine import ResultProxy, RowProxy
from sqlalchemy import exceptions
+from testbase import Table, Column
+
class QueryTest(PersistTest):
from testbase import PersistTest
import testbase
from sqlalchemy import *
+from testbase import Table, Column
+
class QuoteTest(PersistTest):
def setUpAll(self):
from sqlalchemy import *
+from testbase import Table, Column
import testbase
+
class FoundRowsTest(testbase.AssertMixin):
"""tests rowcount functionality"""
def setUpAll(self):
import testbase
from sqlalchemy import *
from sqlalchemy.databases import sqlite, postgres, mysql, oracle, firebird, mssql
+from testbase import Table, Column
import unittest, re, operator
-"""tests that various From objects properly export their columns, as well as useable primary keys\r
-and foreign keys. Full relational algebra depends on every selectable unit behaving\r
-nicely with others.."""\r
-\r
-import testbase\r
-import unittest, sys, datetime\r
-\r
-\r
-db = testbase.db\r
-\r
-from sqlalchemy import *\r
-\r
-metadata = BoundMetaData(db)\r
-table = Table('table1', metadata, \r
- Column('col1', Integer, primary_key=True),\r
- Column('col2', String(20)),\r
- Column('col3', Integer),\r
- Column('colx', Integer),\r
- \r
-)\r
-\r
-table2 = Table('table2', metadata,\r
- Column('col1', Integer, primary_key=True),\r
- Column('col2', Integer, ForeignKey('table1.col1')),\r
- Column('col3', String(20)),\r
- Column('coly', Integer),\r
-)\r
-\r
-class SelectableTest(testbase.AssertMixin):\r
- def testjoinagainstself(self):\r
- jj = select([table.c.col1.label('bar_col1')])\r
- jjj = join(table, jj, table.c.col1==jj.c.bar_col1)\r
- assert jjj.corresponding_column(jjj.c.table1_col1) is jjj.c.table1_col1\r
-\r
- def testjoinagainstjoin(self):\r
- j = outerjoin(table, table2, table.c.col1==table2.c.col2)\r
- jj = select([ table.c.col1.label('bar_col1')],from_obj=[j]).alias('foo')\r
- jjj = join(table, jj, table.c.col1==jj.c.bar_col1)\r
- assert jjj.corresponding_column(jjj.c.table1_col1) is jjj.c.table1_col1\r
- \r
- \r
- def testtablealias(self):\r
- a = table.alias('a')\r
- \r
- j = join(a, table2)\r
- \r
- criterion = a.c.col1 == table2.c.col2\r
- print\r
- print str(j)\r
- self.assert_(criterion.compare(j.onclause))\r
-\r
- def testunion(self):\r
- # tests that we can correspond a column in a Select statement with a certain Table, against\r
- # a column in a Union where one of its underlying Selects matches to that same Table\r
- u = select([table.c.col1, table.c.col2, table.c.col3, table.c.colx, null().label('coly')]).union(\r
- select([table2.c.col1, table2.c.col2, table2.c.col3, null().label('colx'), table2.c.coly])\r
- )\r
- s1 = table.select(use_labels=True)\r
- s2 = table2.select(use_labels=True)\r
- print ["%d %s" % (id(c),c.key) for c in u.c]\r
- c = u.corresponding_column(s1.c.table1_col2)\r
- print "%d %s" % (id(c), c.key)\r
- assert u.corresponding_column(s1.c.table1_col2) is u.c.col2\r
- assert u.corresponding_column(s2.c.table2_col2) is u.c.col2\r
-\r
- def testaliasunion(self):\r
- # same as testunion, except its an alias of the union\r
- u = select([table.c.col1, table.c.col2, table.c.col3, table.c.colx, null().label('coly')]).union(\r
- select([table2.c.col1, table2.c.col2, table2.c.col3, null().label('colx'), table2.c.coly])\r
- ).alias('analias')\r
- s1 = table.select(use_labels=True)\r
- s2 = table2.select(use_labels=True)\r
- assert u.corresponding_column(s1.c.table1_col2) is u.c.col2\r
- assert u.corresponding_column(s2.c.table2_col2) is u.c.col2\r
- assert u.corresponding_column(s2.c.table2_coly) is u.c.coly\r
- assert s2.corresponding_column(u.c.coly) is s2.c.table2_coly\r
-\r
- def testselectunion(self):\r
- # like testaliasunion, but off a Select off the union.\r
- u = select([table.c.col1, table.c.col2, table.c.col3, table.c.colx, null().label('coly')]).union(\r
- select([table2.c.col1, table2.c.col2, table2.c.col3, null().label('colx'), table2.c.coly])\r
- ).alias('analias')\r
- s = select([u])\r
- s1 = table.select(use_labels=True)\r
- s2 = table2.select(use_labels=True)\r
- assert s.corresponding_column(s1.c.table1_col2) is s.c.col2\r
- assert s.corresponding_column(s2.c.table2_col2) is s.c.col2\r
-\r
- def testunionagainstjoin(self):\r
- # same as testunion, except its an alias of the union\r
- u = select([table.c.col1, table.c.col2, table.c.col3, table.c.colx, null().label('coly')]).union(\r
- select([table2.c.col1, table2.c.col2, table2.c.col3, null().label('colx'), table2.c.coly])\r
- ).alias('analias')\r
- j1 = table.join(table2)\r
- assert u.corresponding_column(j1.c.table1_colx) is u.c.colx\r
- assert j1.corresponding_column(u.c.colx) is j1.c.table1_colx\r
- \r
- def testjoin(self):\r
- a = join(table, table2)\r
- print str(a.select(use_labels=True))\r
- b = table2.alias('b')\r
- j = join(a, b)\r
- print str(j)\r
- criterion = a.c.table1_col1 == b.c.col2\r
- self.assert_(criterion.compare(j.onclause))\r
-\r
- def testselectalias(self):\r
- a = table.select().alias('a')\r
- print str(a.select())\r
- j = join(a, table2)\r
- \r
- criterion = a.c.col1 == table2.c.col2\r
- print\r
- print str(j)\r
- self.assert_(criterion.compare(j.onclause))\r
-\r
- def testselectlabels(self):\r
- a = table.select(use_labels=True)\r
- print str(a.select())\r
- j = join(a, table2)\r
- \r
- criterion = a.c.table1_col1 == table2.c.col2\r
- print\r
- print str(j)\r
- self.assert_(criterion.compare(j.onclause))\r
-\r
- def testcolumnlabels(self):\r
- a = select([table.c.col1.label('acol1'), table.c.col2.label('acol2'), table.c.col3.label('acol3')])\r
- print str(a)\r
- print [c for c in a.columns]\r
- print str(a.select())\r
- j = join(a, table2)\r
- criterion = a.c.acol1 == table2.c.col2\r
- print str(j)\r
- self.assert_(criterion.compare(j.onclause))\r
- \r
- def testselectaliaslabels(self):\r
- a = table2.select(use_labels=True).alias('a')\r
- print str(a.select())\r
- j = join(a, table)\r
- \r
- criterion = table.c.col1 == a.c.table2_col2\r
- print str(criterion)\r
- print str(j.onclause)\r
- self.assert_(criterion.compare(j.onclause))\r
- \r
-if __name__ == "__main__":\r
- testbase.main()\r
-
\ No newline at end of file
+"""tests that various From objects properly export their columns, as well as
+useable primary keys and foreign keys. Full relational algebra depends on
+every selectable unit behaving nicely with others.."""
+
+import testbase
+import unittest, sys, datetime
+from sqlalchemy import *
+from testbase import Table, Column
+
+
+db = testbase.db
+metadata = BoundMetaData(db)
+table = Table('table1', metadata,
+ Column('col1', Integer, primary_key=True),
+ Column('col2', String(20)),
+ Column('col3', Integer),
+ Column('colx', Integer),
+
+)
+
+table2 = Table('table2', metadata,
+ Column('col1', Integer, primary_key=True),
+ Column('col2', Integer, ForeignKey('table1.col1')),
+ Column('col3', String(20)),
+ Column('coly', Integer),
+)
+
+class SelectableTest(testbase.AssertMixin):
+ def testjoinagainstself(self):
+ jj = select([table.c.col1.label('bar_col1')])
+ jjj = join(table, jj, table.c.col1==jj.c.bar_col1)
+ assert jjj.corresponding_column(jjj.c.table1_col1) is jjj.c.table1_col1
+
+ def testjoinagainstjoin(self):
+ j = outerjoin(table, table2, table.c.col1==table2.c.col2)
+ jj = select([ table.c.col1.label('bar_col1')],from_obj=[j]).alias('foo')
+ jjj = join(table, jj, table.c.col1==jj.c.bar_col1)
+ assert jjj.corresponding_column(jjj.c.table1_col1) is jjj.c.table1_col1
+
+
+ def testtablealias(self):
+ a = table.alias('a')
+
+ j = join(a, table2)
+
+ criterion = a.c.col1 == table2.c.col2
+ print
+ print str(j)
+ self.assert_(criterion.compare(j.onclause))
+
+ def testunion(self):
+ # tests that we can correspond a column in a Select statement with a certain Table, against
+ # a column in a Union where one of its underlying Selects matches to that same Table
+ u = select([table.c.col1, table.c.col2, table.c.col3, table.c.colx, null().label('coly')]).union(
+ select([table2.c.col1, table2.c.col2, table2.c.col3, null().label('colx'), table2.c.coly])
+ )
+ s1 = table.select(use_labels=True)
+ s2 = table2.select(use_labels=True)
+ print ["%d %s" % (id(c),c.key) for c in u.c]
+ c = u.corresponding_column(s1.c.table1_col2)
+ print "%d %s" % (id(c), c.key)
+ assert u.corresponding_column(s1.c.table1_col2) is u.c.col2
+ assert u.corresponding_column(s2.c.table2_col2) is u.c.col2
+
+ def testaliasunion(self):
+ # same as testunion, except its an alias of the union
+ u = select([table.c.col1, table.c.col2, table.c.col3, table.c.colx, null().label('coly')]).union(
+ select([table2.c.col1, table2.c.col2, table2.c.col3, null().label('colx'), table2.c.coly])
+ ).alias('analias')
+ s1 = table.select(use_labels=True)
+ s2 = table2.select(use_labels=True)
+ assert u.corresponding_column(s1.c.table1_col2) is u.c.col2
+ assert u.corresponding_column(s2.c.table2_col2) is u.c.col2
+ assert u.corresponding_column(s2.c.table2_coly) is u.c.coly
+ assert s2.corresponding_column(u.c.coly) is s2.c.table2_coly
+
+ def testselectunion(self):
+ # like testaliasunion, but off a Select off the union.
+ u = select([table.c.col1, table.c.col2, table.c.col3, table.c.colx, null().label('coly')]).union(
+ select([table2.c.col1, table2.c.col2, table2.c.col3, null().label('colx'), table2.c.coly])
+ ).alias('analias')
+ s = select([u])
+ s1 = table.select(use_labels=True)
+ s2 = table2.select(use_labels=True)
+ assert s.corresponding_column(s1.c.table1_col2) is s.c.col2
+ assert s.corresponding_column(s2.c.table2_col2) is s.c.col2
+
+ def testunionagainstjoin(self):
+ # same as testunion, except its an alias of the union
+ u = select([table.c.col1, table.c.col2, table.c.col3, table.c.colx, null().label('coly')]).union(
+ select([table2.c.col1, table2.c.col2, table2.c.col3, null().label('colx'), table2.c.coly])
+ ).alias('analias')
+ j1 = table.join(table2)
+ assert u.corresponding_column(j1.c.table1_colx) is u.c.colx
+ assert j1.corresponding_column(u.c.colx) is j1.c.table1_colx
+
+ def testjoin(self):
+ a = join(table, table2)
+ print str(a.select(use_labels=True))
+ b = table2.alias('b')
+ j = join(a, b)
+ print str(j)
+ criterion = a.c.table1_col1 == b.c.col2
+ self.assert_(criterion.compare(j.onclause))
+
+ def testselectalias(self):
+ a = table.select().alias('a')
+ print str(a.select())
+ j = join(a, table2)
+
+ criterion = a.c.col1 == table2.c.col2
+ print
+ print str(j)
+ self.assert_(criterion.compare(j.onclause))
+
+ def testselectlabels(self):
+ a = table.select(use_labels=True)
+ print str(a.select())
+ j = join(a, table2)
+
+ criterion = a.c.table1_col1 == table2.c.col2
+ print
+ print str(j)
+ self.assert_(criterion.compare(j.onclause))
+
+ def testcolumnlabels(self):
+ a = select([table.c.col1.label('acol1'), table.c.col2.label('acol2'), table.c.col3.label('acol3')])
+ print str(a)
+ print [c for c in a.columns]
+ print str(a.select())
+ j = join(a, table2)
+ criterion = a.c.acol1 == table2.c.col2
+ print str(j)
+ self.assert_(criterion.compare(j.onclause))
+
+ def testselectaliaslabels(self):
+ a = table2.select(use_labels=True).alias('a')
+ print str(a.select())
+ j = join(a, table)
+
+ criterion = table.c.col1 == a.c.table2_col2
+ print str(criterion)
+ print str(j.onclause)
+ self.assert_(criterion.compare(j.onclause))
+
+if __name__ == "__main__":
+ testbase.main()
+
from sqlalchemy import *
import string,datetime, re, sys, os
import sqlalchemy.engine.url as url
-
import sqlalchemy.types
from sqlalchemy.databases import mssql, oracle
+from testbase import Table, Column
+
db = testbase.db
# coding: utf-8
-import testbase
+"""verrrrry basic unicode column name testing"""
+import testbase
from sqlalchemy import *
from sqlalchemy.orm import mapper, relation, create_session, eagerload
+from testbase import Table, Column
-"""verrrrry basic unicode column name testing"""
class UnicodeSchemaTest(testbase.PersistTest):
def setUpAll(self):
from sqlalchemy import *
import os
import testbase
-
+from testbase import Table, Column
ECHO = testbase.echo
db = testbase.db
users = Table('users', metadata,
Column('user_id', Integer, Sequence('user_id_seq', optional=True), primary_key = True),
Column('user_name', String(40)),
- mysql_engine='innodb'
+ test_needs_acid=True,
+ test_needs_fk=True,
)
addresses = Table('email_addresses', metadata,
Column('address_id', Integer, Sequence('address_id_seq', optional=True), primary_key = True),
Column('user_id', Integer, ForeignKey(users.c.user_id)),
Column('email_address', String(40)),
-
)
orders = Table('orders', metadata,
Column('user_id', Integer, ForeignKey(users.c.user_id)),
Column('description', String(50)),
Column('isopen', Integer),
-
)
orderitems = Table('items', metadata,
Column('item_id', INT, Sequence('items_id_seq', optional=True), primary_key = True),
Column('order_id', INT, ForeignKey("orders")),
Column('item_name', VARCHAR(50)),
-
)
keywords = Table('keywords', metadata,
Column('keyword_id', Integer, Sequence('keyword_id_seq', optional=True), primary_key = True),
Column('name', VARCHAR(50)),
-
)
userkeywords = Table('userkeywords', metadata,
import os, unittest, StringIO, re, ConfigParser, optparse
sys.path.insert(0, os.path.join(os.getcwd(), 'lib'))
import sqlalchemy
-from sqlalchemy import sql, engine, pool, BoundMetaData
+from sqlalchemy import sql, schema, engine, pool, BoundMetaData
from sqlalchemy.orm import clear_mappers
db = None
metadata = None
db_uri = None
echo = True
+table_options = {}
# redefine sys.stdout so all those print statements go to the echo func
local_stdout = sys.stdout
parser.add_option("--reversetop", action="store_true", dest="topological", help="Reverse the collection ordering for topological sorts (helps reveal dependency issues)")
parser.add_option("--serverside", action="store_true", dest="serverside", help="Turn on server side cursors for PG")
parser.add_option("--require", action="append", dest="require", help="Require a particular driver or module version", default=[])
+ parser.add_option("--mysql-engine", action="store", dest="mysql_engine", help="Use the specified MySQL storage engine for all tables, default is a db-default/InnoDB combo.", default=None)
+ parser.add_option("--table-option", action="append", dest="tableopts", help="Add a dialect-specific table option, key=value", default=[])
(options, args) = parser.parse_args()
sys.argv[1:] = args
if not db_uri:
raise "Could not create engine. specify --db <sqlite|sqlite_file|postgres|mysql|oracle|oracle8|mssql|firebird> to test runner."
+ global table_options
+ for spec in options.tableopts:
+ key, value = spec.split('=')
+ table_options[key] = value
+
+ if options.mysql_engine:
+ table_options['mysql_engine'] = options.mysql_engine
+
if not options.nothreadlocal:
__import__('sqlalchemy.mods.threadlocal')
sqlalchemy.mods.threadlocal.uninstall_plugin()
return lala
return decorate
+def Table(*args, **kw):
+ """A schema.Table wrapper/hook for dialect-specific tweaks."""
+
+ test_opts = dict([(k,kw.pop(k)) for k in kw.keys()
+ if k.startswith('test_')])
+
+ kw.update(table_options)
+
+ if db.engine.name == 'mysql':
+ if 'mysql_engine' not in kw and 'mysql_type' not in kw:
+ if 'test_needs_fk' in test_opts or 'test_needs_acid' in test_opts:
+ kw['mysql_engine'] = 'InnoDB'
+
+ return schema.Table(*args, **kw)
+
+def Column(*args, **kw):
+ """A schema.Column wrapper/hook for dialect-specific tweaks."""
+
+ # TODO: a Column that creates a Sequence automatically for PK columns,
+ # which would help Oracle tests
+ return schema.Column(*args, **kw)
class PersistTest(unittest.TestCase):
from sqlalchemy import *
+from testbase import Table, Column
metadata = MetaData()
"""application table metadata objects are described here."""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
