import codecs

import pytest

from sqlobject import BoolCol, ForeignKey, IntCol, KeyCol, SQLObject, \

StringCol, connectionForURI, sqlhub

from sqlobject.tests.dbtest import inserts, raises, setupClass, supports

class SOTestSO1(SQLObject):

name = StringCol(length=50, dbName='name_col')

name.title = 'Your Name'

name.foobar = 1

passwd = StringCol(length=10)

class sqlmeta:

cacheValues = False

def _set_passwd(self, passwd):

self._SO_set_passwd(codecs.encode(passwd, 'rot13'))

def setupGetters(cls):

setupClass(cls)

inserts(cls, [('bob', 'god'), ('sally', 'sordid'),

('dave', 'dremel'), ('fred', 'forgo')],

'name passwd')

def test_case1():

setupGetters(SOTestSO1)

bob = SOTestSO1.selectBy(name='bob')[0]

assert bob.name == 'bob'

assert bob.passwd == codecs.encode('god', 'rot13')

bobs = SOTestSO1.selectBy(name='bob')[:10]

assert len(list(bobs)) == 1

def test_newline():

setupGetters(SOTestSO1)

bob = SOTestSO1.selectBy(name='bob')[0]

testString = 'hey\nyou\\can\'t you see me?\t'

bob.name = testString

bob.expire()

assert bob.name == testString

def test_count():

setupGetters(SOTestSO1)

assert SOTestSO1.selectBy(name=None).count() == 0

assert SOTestSO1.selectBy(name='bob').count() == 1

assert SOTestSO1.select(SOTestSO1.q.name == 'bob').count() == 1

assert SOTestSO1.select().count() == len(list(SOTestSO1.select()))

def test_getset():

setupGetters(SOTestSO1)

bob = SOTestSO1.selectBy(name='bob')[0]

assert bob.name == 'bob'

bob.name = 'joe'

assert bob.name == 'joe'

bob.set(name='joebob', passwd='testtest')

assert bob.name == 'joebob'

def test_extra_vars():

setupGetters(SOTestSO1)

col = SOTestSO1.sqlmeta.columns['name']

assert col.title == 'Your Name'

assert col.foobar == 1

assert getattr(SOTestSO1.sqlmeta.columns['passwd'], 'title', None) is None

class SOTestSO2(SQLObject):

name = StringCol(length=50, dbName='name_col')

passwd = StringCol(length=10)

def _set_passwd(self, passwd):

self._SO_set_passwd(codecs.encode(passwd, 'rot13'))

def test_case2():

setupGetters(SOTestSO2)

bob = SOTestSO2.selectBy(name='bob')[0]

assert bob.name == 'bob'

assert bob.passwd == codecs.encode('god', 'rot13')

class Student(SQLObject):

is_smart = BoolCol()

def test_boolCol():

setupClass(Student)

student = Student(is_smart=False)

assert not student.is_smart

student2 = Student(is_smart=1)

assert student2.is_smart

class SOTestSO3(SQLObject):

name = StringCol(length=10, dbName='name_col')

other = ForeignKey('SOTestSO4', default=None)

other2 = KeyCol(foreignKey='SOTestSO4', default=None)

class SOTestSO4(SQLObject):

me = StringCol(length=10)

def test_foreignKey():

setupClass([SOTestSO4, SOTestSO3])

test3_order = [col.name for col in SOTestSO3.sqlmeta.columnList]

assert test3_order == ['name', 'otherID', 'other2ID']

tc3 = SOTestSO3(name='a')

assert tc3.other is None

assert tc3.other2 is None

assert tc3.otherID is None

assert tc3.other2ID is None

tc4a = SOTestSO4(me='1')

tc3.other = tc4a

assert tc3.other == tc4a

assert tc3.otherID == tc4a.id

tc4b = SOTestSO4(me='2')

tc3.other = tc4b.id

assert tc3.other == tc4b

assert tc3.otherID == tc4b.id

tc4c = SOTestSO4(me='3')

tc3.other2 = tc4c

assert tc3.other2 == tc4c

assert tc3.other2ID == tc4c.id

tc4d = SOTestSO4(me='4')

tc3.other2 = tc4d.id

assert tc3.other2 == tc4d

assert tc3.other2ID == tc4d.id

tcc = SOTestSO3(name='b', other=tc4a)

assert tcc.other == tc4a

tcc2 = SOTestSO3(name='c', other=tc4a.id)

assert tcc2.other == tc4a

def test_selectBy():

setupClass([SOTestSO4, SOTestSO3])

tc4 = SOTestSO4(me='another')

tc3 = SOTestSO3(name='sel', other=tc4)

SOTestSO3(name='not joined')

assert tc3.other == tc4

assert list(SOTestSO3.selectBy(other=tc4)) == [tc3]

assert list(SOTestSO3.selectBy(otherID=tc4.id)) == [tc3]

assert SOTestSO3.selectBy(otherID=tc4.id)[0] == tc3

assert list(SOTestSO3.selectBy(otherID=tc4.id)[:10]) == [tc3]

assert list(SOTestSO3.selectBy(other=tc4)[:10]) == [tc3]

class SOTestSO5(SQLObject):

name = StringCol(length=10, dbName='name_col')

other = ForeignKey('SOTestSO6', default=None, cascade=True)

another = ForeignKey('SOTestSO7', default=None, cascade=True)

class SOTestSO6(SQLObject):

name = StringCol(length=10, dbName='name_col')

other = ForeignKey('SOTestSO7', default=None, cascade=True)

class SOTestSO7(SQLObject):

name = StringCol(length=10, dbName='name_col')

def test_foreignKeyDestroySelfCascade():

setupClass([SOTestSO7, SOTestSO6, SOTestSO5])

tc5 = SOTestSO5(name='a')

tc6a = SOTestSO6(name='1')

tc5.other = tc6a

tc7a = SOTestSO7(name='2')

tc6a.other = tc7a

tc5.another = tc7a

assert tc5.other == tc6a

assert tc5.otherID == tc6a.id

assert tc6a.other == tc7a

assert tc6a.otherID == tc7a.id

assert tc5.other.other == tc7a

assert tc5.other.otherID == tc7a.id

assert tc5.another == tc7a

assert tc5.anotherID == tc7a.id

assert tc5.other.other == tc5.another

assert SOTestSO5.select().count() == 1

assert SOTestSO6.select().count() == 1

assert SOTestSO7.select().count() == 1

tc6b = SOTestSO6(name='3')

tc6c = SOTestSO6(name='4')

tc7b = SOTestSO7(name='5')

tc6b.other = tc7b

tc6c.other = tc7b

assert SOTestSO5.select().count() == 1

assert SOTestSO6.select().count() == 3

assert SOTestSO7.select().count() == 2

tc6b.destroySelf()

assert SOTestSO5.select().count() == 1

assert SOTestSO6.select().count() == 2

assert SOTestSO7.select().count() == 2

tc7b.destroySelf()

assert SOTestSO5.select().count() == 1

assert SOTestSO6.select().count() == 1

assert SOTestSO7.select().count() == 1

tc7a.destroySelf()

assert SOTestSO5.select().count() == 0

assert SOTestSO6.select().count() == 0

assert SOTestSO7.select().count() == 0

def testForeignKeyDropTableCascade():

if not supports('dropTableCascade'):

pytest.skip("dropTableCascade isn't supported")

setupClass(SOTestSO7)

setupClass(SOTestSO6)

setupClass(SOTestSO5)

tc5a = SOTestSO5(name='a')

tc6a = SOTestSO6(name='1')

tc5a.other = tc6a

tc7a = SOTestSO7(name='2')

tc6a.other = tc7a

tc5a.another = tc7a

tc5b = SOTestSO5(name='b')

tc5c = SOTestSO5(name='c')

tc6b = SOTestSO6(name='3')

tc5c.other = tc6b

assert SOTestSO5.select().count() == 3

assert SOTestSO6.select().count() == 2

assert SOTestSO7.select().count() == 1

SOTestSO7.dropTable(cascade=True)

assert SOTestSO5.select().count() == 3

assert SOTestSO6.select().count() == 2

tc6a.destroySelf()

assert SOTestSO5.select().count() == 2

assert SOTestSO6.select().count() == 1

tc6b.destroySelf()

assert SOTestSO5.select().count() == 1

assert SOTestSO6.select().count() == 0

assert next(iter(SOTestSO5.select())) == tc5b

tc6c = SOTestSO6(name='3')

tc5b.other = tc6c

assert SOTestSO5.select().count() == 1

assert SOTestSO6.select().count() == 1

tc6c.destroySelf()

assert SOTestSO5.select().count() == 0

assert SOTestSO6.select().count() == 0

class SOTestSO8(SQLObject):

name = StringCol(length=10, dbName='name_col')

other = ForeignKey('SOTestSO9', default=None, cascade=False)

class SOTestSO9(SQLObject):

name = StringCol(length=10, dbName='name_col')

def testForeignKeyDestroySelfRestrict():

setupClass([SOTestSO9, SOTestSO8])

tc8a = SOTestSO8(name='a')

tc9a = SOTestSO9(name='1')

tc8a.other = tc9a

tc8b = SOTestSO8(name='b')

tc9b = SOTestSO9(name='2')

assert tc8a.other == tc9a

assert tc8a.otherID == tc9a.id

assert SOTestSO8.select().count() == 2

assert SOTestSO9.select().count() == 2

raises(Exception, tc9a.destroySelf)

tc9b.destroySelf()

assert SOTestSO8.select().count() == 2

assert SOTestSO9.select().count() == 1

tc8a.destroySelf()

tc8b.destroySelf()

tc9a.destroySelf()

assert SOTestSO8.select().count() == 0

assert SOTestSO9.select().count() == 0

class SOTestSO10(SQLObject):

name = StringCol()

class SOTestSO11(SQLObject):

name = StringCol()

other = ForeignKey('SOTestSO10', default=None, cascade='null')

def testForeignKeySetNull():

setupClass([SOTestSO10, SOTestSO11])

obj1 = SOTestSO10(name='foo')

obj2 = SOTestSO10(name='bar')

dep1 = SOTestSO11(name='xxx', other=obj1)

dep2 = SOTestSO11(name='yyy', other=obj1)

dep3 = SOTestSO11(name='zzz', other=obj2)

for name in 'xxx', 'yyy', 'zzz':

assert len(list(SOTestSO11.selectBy(name=name))) == 1

obj1.destroySelf()

for name in 'xxx', 'yyy', 'zzz':

assert len(list(SOTestSO11.selectBy(name=name))) == 1

assert dep1.other is None

assert dep2.other is None

assert dep3.other is obj2

def testAsDict():

setupGetters(SOTestSO1)

bob = SOTestSO1.selectBy(name='bob')[0]

assert bob.sqlmeta.asDict() == {

'passwd': 'tbq', 'name': 'bob', 'id': bob.id}

def test_nonexisting_attr():

setupClass(Student)

raises(AttributeError, getattr, Student.q, 'nonexisting')

class SOTestSO12(SQLObject):

name = StringCol()

so_value = IntCol(defaultSQL='1')

def test_defaultSQL():

setupClass(SOTestSO12)

test = SOTestSO12(name="test")

assert test.so_value == 1

def test_connection_override():

sqlhub.processConnection = connectionForURI('sqlite:///db1')

class SOTestSO13(SQLObject):

_connection = connectionForURI('sqlite:///db2')

assert SOTestSO13._connection.uri() == 'sqlite:///db2'

del sqlhub.processConnection

def _test_wrong_sqlmeta_idType():

class SOTestSO13(SQLObject):

class sqlmeta:

idType = dict

def test_wrong_sqlmeta_idType():

pytest.raises(TypeError, _test_wrong_sqlmeta_idType)

def _test_wrong_sqlmeta_idSize():

class SOTestSO14(SQLObject):

class sqlmeta:

idSize = 'DEFAULT'

def test_wrong_sqlmeta_idSize():

pytest.raises(ValueError, _test_wrong_sqlmeta_idSize)