summaryrefslogtreecommitdiff
path: root/test/ext/test_sqlsoup.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2010-07-11 13:41:38 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2010-07-11 13:41:38 -0400
commitf2b43da1a8e3fa8f2afc49e04ec16479a6e7da64 (patch)
tree90f12adf613f4a97c8ad5861560f86abd8263ccb /test/ext/test_sqlsoup.py
parenta6b62cc3fed5f06d3428b1f6ee13756175ded61b (diff)
downloadsqlalchemy-f2b43da1a8e3fa8f2afc49e04ec16479a6e7da64.tar.gz
tidy test/base, test/ex, test/ext
Diffstat (limited to 'test/ext/test_sqlsoup.py')
-rw-r--r--test/ext/test_sqlsoup.py378
1 files changed, 179 insertions, 199 deletions
diff --git a/test/ext/test_sqlsoup.py b/test/ext/test_sqlsoup.py
index 9216af96f..7fe8ab178 100644
--- a/test/ext/test_sqlsoup.py
+++ b/test/ext/test_sqlsoup.py
@@ -1,15 +1,14 @@
-#!coding:utf-8
-
from sqlalchemy.ext import sqlsoup
from sqlalchemy.test.testing import TestBase, eq_, assert_raises
-from sqlalchemy import create_engine, or_, desc, select, func, exc, Table,\
- util
+from sqlalchemy import create_engine, or_, desc, select, func, exc, \
+ Table, util
from sqlalchemy.orm import scoped_session, sessionmaker
import datetime
class SQLSoupTest(TestBase):
- __requires__ = ('sqlite', )
+
+ __requires__ = 'sqlite',
@classmethod
def setup_class(cls):
@@ -17,205 +16,198 @@ class SQLSoupTest(TestBase):
engine = create_engine('sqlite://')
for sql in _ddl:
engine.execute(sql)
-
+
@classmethod
def teardown_class(cls):
engine.dispose()
-
+
def setup(self):
for sql in _data:
engine.execute(sql)
-
+
def teardown(self):
sqlsoup.Session.remove()
for sql in _teardown:
engine.execute(sql)
-
+
def test_bad_names(self):
db = sqlsoup.SqlSoup(engine)
-# print db.bad_names.c.id
+
+ # print db.bad_names.c.id
+
print db.bad_names.c.query
-
-
+
def test_load(self):
db = sqlsoup.SqlSoup(engine)
MappedUsers = db.users
users = db.users.all()
users.sort()
- eq_(
- users,
- [
- MappedUsers(
- name=u'Joe Student',
- email=u'student@example.edu',
- password=u'student',classname=None,admin=0),
- MappedUsers(
- name=u'Bhargan Basepair',
- email=u'basepair@example.edu',
- password=u'basepair',classname=None,admin=1)
- ]
- )
-
+ eq_(users, [MappedUsers(name=u'Joe Student',
+ email=u'student@example.edu', password=u'student',
+ classname=None, admin=0),
+ MappedUsers(name=u'Bhargan Basepair',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1)])
+
def test_order_by(self):
db = sqlsoup.SqlSoup(engine)
MappedUsers = db.users
users = db.users.order_by(db.users.name).all()
- eq_(
- users,
- [
- MappedUsers(
- name=u'Bhargan Basepair',
- email=u'basepair@example.edu',
- password=u'basepair',classname=None,admin=1),
- MappedUsers(
- name=u'Joe Student',
- email=u'student@example.edu',
- password=u'student',classname=None,admin=0),
- ]
- )
-
+ eq_(users, [MappedUsers(name=u'Bhargan Basepair',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1), MappedUsers(name=u'Joe Student',
+ email=u'student@example.edu', password=u'student',
+ classname=None, admin=0)])
+
def test_whereclause(self):
db = sqlsoup.SqlSoup(engine)
MappedUsers = db.users
-
- where = or_(db.users.name=='Bhargan Basepair', db.users.email=='student@example.edu')
- users = db.users.filter(where).order_by(desc(db.users.name)).all()
- eq_(
- users,
- [MappedUsers(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0), MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1)]
- )
-
+ where = or_(db.users.name == 'Bhargan Basepair', db.users.email
+ == 'student@example.edu')
+ users = \
+ db.users.filter(where).order_by(desc(db.users.name)).all()
+ eq_(users, [MappedUsers(name=u'Joe Student',
+ email=u'student@example.edu', password=u'student',
+ classname=None, admin=0),
+ MappedUsers(name=u'Bhargan Basepair',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1)])
+
def test_first(self):
db = sqlsoup.SqlSoup(engine)
MappedUsers = db.users
-
- user = db.users.filter(db.users.name=='Bhargan Basepair').one()
- eq_(user,
- MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1)
- )
+ user = db.users.filter(db.users.name == 'Bhargan Basepair'
+ ).one()
+ eq_(user, MappedUsers(name=u'Bhargan Basepair',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1))
db.rollback()
-
user = db.users.get('Bhargan Basepair')
- eq_(user, MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1))
+ eq_(user, MappedUsers(name=u'Bhargan Basepair',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1))
db.rollback()
-
user = db.users.filter_by(name='Bhargan Basepair').one()
- eq_(user, MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1))
+ eq_(user, MappedUsers(name=u'Bhargan Basepair',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1))
db.rollback()
def test_crud(self):
+
# note we're testing autoflush here too...
+
db = sqlsoup.SqlSoup(engine)
MappedLoans = db.loans
user = db.users.filter_by(name='Bhargan Basepair').one()
- book_id = db.books.filter_by(title='Regional Variation in Moss').first().id
- loan_insert = db.loans.insert(book_id=book_id, user_name=user.name)
-
- loan = db.loans.filter_by(book_id=2, user_name='Bhargan Basepair').one()
+ book_id = db.books.filter_by(title='Regional Variation in Moss'
+ ).first().id
+ loan_insert = db.loans.insert(book_id=book_id,
+ user_name=user.name)
+ loan = db.loans.filter_by(book_id=2,
+ user_name='Bhargan Basepair').one()
eq_(loan, loan_insert)
- l2 = MappedLoans(book_id=2,user_name=u'Bhargan Basepair',loan_date=loan.loan_date)
+ l2 = MappedLoans(book_id=2, user_name=u'Bhargan Basepair',
+ loan_date=loan.loan_date)
eq_(loan, l2)
db.expunge(l2)
-
db.delete(loan)
- loan = db.loans.filter_by(book_id=2, user_name='Bhargan Basepair').first()
+ loan = db.loans.filter_by(book_id=2,
+ user_name='Bhargan Basepair').first()
assert loan is None
-
+
def test_cls_crud(self):
db = sqlsoup.SqlSoup(engine)
MappedUsers = db.users
- db.users.filter_by(name='Bhargan Basepair').update(dict(name='Some New Name'))
- u1= db.users.filter_by(name='Some New Name').one()
- eq_(
- u1,
- MappedUsers(name=u'Some New Name',
- email=u'basepair@example.edu',
- password=u'basepair',classname=None,admin=1)
- )
-
+ db.users.filter_by(name='Bhargan Basepair'
+ ).update(dict(name='Some New Name'))
+ u1 = db.users.filter_by(name='Some New Name').one()
+ eq_(u1, MappedUsers(name=u'Some New Name',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1))
+
def test_map_table(self):
db = sqlsoup.SqlSoup(engine)
users = Table('users', db._metadata, autoload=True)
MappedUsers = db.map(users)
-
users = MappedUsers.order_by(db.users.name).all()
- eq_(
- users,
- [
- MappedUsers(
- name=u'Bhargan Basepair',
- email=u'basepair@example.edu',
- password=u'basepair',classname=None,admin=1),
- MappedUsers(
- name=u'Joe Student',
- email=u'student@example.edu',
- password=u'student',classname=None,admin=0),
- ]
- )
-
+ eq_(users, [MappedUsers(name=u'Bhargan Basepair',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1), MappedUsers(name=u'Joe Student',
+ email=u'student@example.edu', password=u'student',
+ classname=None, admin=0)])
+
def test_mapped_join(self):
db = sqlsoup.SqlSoup(engine)
-
join1 = MappedJoin = db.join(db.users, db.loans, isouter=True)
mj = join1.filter_by(name='Joe Student').all()
- eq_(
- mj,
- [MappedJoin(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0,book_id=1,user_name=u'Joe Student',loan_date=datetime.datetime(2006, 7, 12, 0, 0))]
- )
-
+ eq_(mj, [MappedJoin(
+ name=u'Joe Student',
+ email=u'student@example.edu',
+ password=u'student',
+ classname=None,
+ admin=0,
+ book_id=1,
+ user_name=u'Joe Student',
+ loan_date=datetime.datetime(2006, 7, 12, 0, 0),
+ )])
db.rollback()
-
join2 = MappedJoin = db.join(join1, db.books)
mj = join2.all()
- eq_(mj, [MappedJoin(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0,book_id=1,user_name=u'Joe Student',loan_date=datetime.datetime(2006, 7, 12, 0, 0),id=1,title=u'Mustards I Have Known',published_year=u'1989',authors=u'Jones')])
-
- eq_(
- db.with_labels(join1).c.keys(),
- [u'users_name', u'users_email', u'users_password',
- u'users_classname', u'users_admin',
- u'loans_book_id', u'loans_user_name',
- u'loans_loan_date']
- )
-
+ eq_(mj, [MappedJoin(
+ name=u'Joe Student',
+ email=u'student@example.edu',
+ password=u'student',
+ classname=None,
+ admin=0,
+ book_id=1,
+ user_name=u'Joe Student',
+ loan_date=datetime.datetime(2006, 7, 12, 0, 0),
+ id=1,
+ title=u'Mustards I Have Known',
+ published_year=u'1989',
+ authors=u'Jones',
+ )])
+ eq_(db.with_labels(join1).c.keys(), [
+ u'users_name',
+ u'users_email',
+ u'users_password',
+ u'users_classname',
+ u'users_admin',
+ u'loans_book_id',
+ u'loans_user_name',
+ u'loans_loan_date',
+ ])
labeled_loans = db.with_labels(db.loans)
- eq_(
- db.join(db.users, labeled_loans, isouter=True).c.keys(),
- [u'name', u'email', u'password', u'classname',
- u'admin', u'loans_book_id', u'loans_user_name', u'loans_loan_date']
- )
-
+ eq_(db.join(db.users, labeled_loans, isouter=True).c.keys(), [
+ u'name',
+ u'email',
+ u'password',
+ u'classname',
+ u'admin',
+ u'loans_book_id',
+ u'loans_user_name',
+ u'loans_loan_date',
+ ])
+
def test_relations(self):
db = sqlsoup.SqlSoup(engine)
db.users.relate('loans', db.loans)
-
MappedLoans = db.loans
MappedUsers = db.users
-
- eq_(
- db.users.get('Joe Student').loans,
- [MappedLoans(
- book_id=1,
- user_name=u'Joe Student',
- loan_date=datetime.datetime(2006, 7, 12, 0, 0))
- ]
- )
+ eq_(db.users.get('Joe Student').loans, [MappedLoans(book_id=1,
+ user_name=u'Joe Student', loan_date=datetime.datetime(2006,
+ 7, 12, 0, 0))])
db.rollback()
-
- eq_(
- db.users.filter(~db.users.loans.any()).all(),
- [MappedUsers(
- name=u'Bhargan Basepair',
- email='basepair@example.edu',
- password=u'basepair',
- classname=None,admin=1)
- ]
- )
+ eq_(db.users.filter(~db.users.loans.any()).all(),
+ [MappedUsers(name=u'Bhargan Basepair',
+ email='basepair@example.edu', password=u'basepair',
+ classname=None, admin=1)])
db.rollback()
-
del db._cache['users']
- db.users.relate('loans', db.loans,
- order_by=db.loans.loan_date, cascade='all, delete-orphan')
-
+ db.users.relate('loans', db.loans, order_by=db.loans.loan_date,
+ cascade='all, delete-orphan')
+
def test_explicit_session(self):
Session = scoped_session(sessionmaker())
db = sqlsoup.SqlSoup(engine, session=Session)
@@ -223,62 +215,46 @@ class SQLSoupTest(TestBase):
MappedUsers = db.users
sess = Session()
assert db.users._query.session is db.users.session is sess
-
row = db.users.insert(name='new name', email='new email')
assert row in sess
finally:
sess.rollback()
sess.close()
-
+
def test_selectable(self):
db = sqlsoup.SqlSoup(engine)
MappedBooks = db.books
b = db.books._table
-
- s = select(
- [b.c.published_year, func.count('*').label('n')],
- from_obj=[b], group_by=[b.c.published_year])
-
+ s = select([b.c.published_year, func.count('*').label('n')],
+ from_obj=[b], group_by=[b.c.published_year])
s = s.alias('years_with_count')
years_with_count = db.map(s, primary_key=[s.c.published_year])
-
- eq_(
- years_with_count.filter_by(published_year='1989').all(),
- [MappedBooks(published_year=u'1989',n=1)]
- )
-
+ eq_(years_with_count.filter_by(published_year='1989').all(),
+ [MappedBooks(published_year=u'1989', n=1)])
+
def test_raw_sql(self):
db = sqlsoup.SqlSoup(engine)
rp = db.execute('select name, email from users order by name')
- eq_(
- rp.fetchall(),
- [('Bhargan Basepair', 'basepair@example.edu'),
- ('Joe Student', 'student@example.edu')]
- )
-
- # test that execute() shares the same transactional
- # context as the session
+ eq_(rp.fetchall(), [('Bhargan Basepair', 'basepair@example.edu'
+ ), ('Joe Student', 'student@example.edu')])
+
+ # test that execute() shares the same transactional context as
+ # the session
+
db.execute("update users set email='foo bar'")
- eq_(
- db.execute("select distinct email from users").fetchall(),
- [('foo bar',)]
- )
+ eq_(db.execute('select distinct email from users').fetchall(),
+ [('foo bar', )])
db.rollback()
- eq_(
- db.execute("select distinct email from users").fetchall(),
- [(u'basepair@example.edu',), (u'student@example.edu',)]
- )
-
+ eq_(db.execute('select distinct email from users').fetchall(),
+ [(u'basepair@example.edu', ), (u'student@example.edu', )])
+
def test_connection(self):
db = sqlsoup.SqlSoup(engine)
conn = db.connection()
rp = conn.execute('select name, email from users order by name')
- eq_(
- rp.fetchall(),
- [('Bhargan Basepair', 'basepair@example.edu'),
- ('Joe Student', 'student@example.edu')]
- )
-
+ eq_(rp.fetchall(), [('Bhargan Basepair', 'basepair@example.edu'
+ ), ('Joe Student', 'student@example.edu')])
+
def test_entity(self):
db = sqlsoup.SqlSoup(engine)
tablename = 'loans'
@@ -287,40 +263,40 @@ class SQLSoupTest(TestBase):
def test_entity_with_different_base(self):
class subclass(object):
pass
+
db = sqlsoup.SqlSoup(engine, base=subclass)
assert issubclass(db.entity('loans'), subclass)
-
+
def test_filter_by_order_by(self):
db = sqlsoup.SqlSoup(engine)
MappedUsers = db.users
- users = db.users.filter_by(classname=None).order_by(db.users.name).all()
- eq_(
- users,
- [MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1), MappedUsers(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0)]
- )
-
+ users = \
+ db.users.filter_by(classname=None).order_by(db.users.name).all()
+ eq_(users, [MappedUsers(name=u'Bhargan Basepair',
+ email=u'basepair@example.edu', password=u'basepair',
+ classname=None, admin=1), MappedUsers(name=u'Joe Student',
+ email=u'student@example.edu', password=u'student',
+ classname=None, admin=0)])
+
def test_no_pk(self):
db = sqlsoup.SqlSoup(engine)
assert_raises(sqlsoup.PKNotFoundError, getattr, db, 'nopk')
-
+
def test_nosuchtable(self):
db = sqlsoup.SqlSoup(engine)
assert_raises(exc.NoSuchTableError, getattr, db, 'nosuchtable')
-
+
def test_dont_persist_alias(self):
db = sqlsoup.SqlSoup(engine)
MappedBooks = db.books
b = db.books._table
-
- s = select(
- [b.c.published_year, func.count('*').label('n')],
- from_obj=[b], group_by=[b.c.published_year])
-
+ s = select([b.c.published_year, func.count('*').label('n')],
+ from_obj=[b], group_by=[b.c.published_year])
s = s.alias('years_with_count')
years_with_count = db.map(s, primary_key=[s.c.published_year])
-
- assert_raises(exc.InvalidRequestError, years_with_count.insert, published_year='2007', n=1)
-
+ assert_raises(exc.InvalidRequestError, years_with_count.insert,
+ published_year='2007', n=1)
+
def test_clear(self):
db = sqlsoup.SqlSoup(engine)
eq_(db.loans.count(), 1)
@@ -328,8 +304,10 @@ class SQLSoupTest(TestBase):
db.expunge_all()
db.flush()
eq_(db.loans.count(), 1)
-
-_ddl = u"""
+
+
+_ddl = \
+ u"""
CREATE TABLE books (
id integer PRIMARY KEY, -- auto-increments in sqlite
title text NOT NULL,
@@ -360,10 +338,10 @@ CREATE TABLE bad_names (
id int primary key,
query varchar(100)
)
-""".split(';')
-
-
-_data = """
+""".split(';'
+ )
+_data = \
+ """
insert into users(name, email, password, admin)
values('Bhargan Basepair', 'basepair@example.edu', 'basepair', 1);
insert into users(name, email, password, admin)
@@ -380,11 +358,13 @@ values (
(select name from users where name like 'Joe%'),
'2006-07-12 0:0:0')
;
-""".split(";")
-
-_teardown = """
+""".split(';'
+ )
+_teardown = \
+ """
delete from loans;
delete from books;
delete from users;
delete from nopk;
-""".split(";")
+""".split(';'
+ )