Source code for sqlobject.tests.test_delete

from sqlobject import OR, RelatedJoin, SQLObject, StringCol, cache
from sqlobject.tests.dbtest import setupClass
from .test_basic import SOTestSO1, setupGetters


########################################
# Delete during select
########################################


[docs]def testSelect():
    setupGetters(SOTestSO1)
    for obj in SOTestSO1.select('all'):
        obj.destroySelf()
    assert list(SOTestSO1.select('all')) == []


########################################
# Delete many rows at once
########################################


[docs]def testDeleteMany():
    setupGetters(SOTestSO1)
    SOTestSO1.deleteMany(OR(SOTestSO1.q.name == "bob",
                            SOTestSO1.q.name == "fred"))
    assert len(list(SOTestSO1.select('all'))) == 2


[docs]def testDeleteBy():
    setupGetters(SOTestSO1)
    SOTestSO1.deleteBy(name="dave")
    assert len(list(SOTestSO1.select())) == 3


########################################
# Delete without caching
########################################


[docs]class NoCache(SQLObject):
    name = StringCol()


[docs]def testDestroySelf():
    setupClass(NoCache)
    old = NoCache._connection.cache
    NoCache._connection.cache = cache.CacheSet(cache=False)
    value = NoCache(name='test')
    value.destroySelf()
    NoCache._connection.cache = old


########################################
# Delete from related joins
########################################


[docs]class Service(SQLObject):
    groups = RelatedJoin("ServiceGroup")


[docs]class ServiceGroup(SQLObject):
    services = RelatedJoin("Service")


[docs]def testDeleteRelatedJoins():
    setupClass([Service, ServiceGroup])
    service = Service()
    service_group = ServiceGroup()
    service.addServiceGroup(service_group)
    service.destroySelf()
    service_group = ServiceGroup.get(service_group.id)
    assert len(service_group.services) == 0