Viewing file:  test_tablecreate.py (3.26 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
"""
Tests for table-creation.
"""
from axiom import item
from axiom import attributes
from axiom import store
from twisted.trial.unittest import TestCase
from twisted.python import filepath
class A(item.Item):
typeName = 'test_table_creator'
schemaVersion = 1
attr = attributes.integer(default=3)
class SomeError(Exception):
"""
Dummy error for testing.
"""
def createAndRaise(s):
"""
Create an A item, then raise a L{SomeError} (which will revert the
transaction). This is because there is no direct API for creating tables.
"""
A(store=s)
raise SomeError()
class TableCreationTest(TestCase):
"""
Tests for table creation.
"""
def test_committedTableCreation(self):
"""
When tables are created in a transaction which is committed, they should
persist in both Axiom's in-memory schema representation and within the
on-disk SQL store.
"""
storedir = filepath.FilePath(self.mktemp())
s1 = store.Store(storedir)
s1.transact(A, store=s1)
self.assertIn(A, s1.typeToTableNameCache)
s1.close()
s2 = store.Store(storedir)
self.assertIn(A, s2.typeToTableNameCache)
s2.close()
def test_revertedTableCreation(self):
"""
When tables are created in a transaction which is reverted, they should
persist in neither the SQL store nor the in-memory schema
representation.
"""
storedir = self.mktemp()
s1 = store.Store(storedir)
self.assertRaises(SomeError, s1.transact, createAndRaise, s1)
self.assertNotIn(A, s1.typeToTableNameCache)
s1.close()
s2 = store.Store(storedir)
self.assertNotIn(A, s2.typeToTableNameCache)
def test_differentStoreTableCreation(self):
"""
If two different stores are opened before a given table is created, and
one creates it, this should be transparent to both item creation and
queries made from either store.
"""
storedir = self.mktemp()
s1 = store.Store(storedir)
s2 = store.Store(storedir)
a1 = A(store=s1)
a2 = A(store=s2)
self.assertEquals(list(s1.query(
A, sort=A.storeID.ascending).getColumn("storeID")),
[a1.storeID, a2.storeID])
self.assertEquals(list(s2.query(
A, sort=A.storeID.ascending).getColumn("storeID")),
[a1.storeID, a2.storeID])
def test_dontReadTheSchemaSoMuch(self):
"""
This is a regression test for a bug in Axiom where the schema was
refreshed from SQL every time a table needed to be created, regardless
of whether the schema needed to be refreshed or not. In addition to
being logically incorrect, this error severely hurt performance.
The schema should only be re-read when a change is detected, by way of
a table being created in two different Store objects, as in the test
above in L{TableCreationTest.test_differentStoreTableCreation}.
"""
s1 = store.Store(filepath.FilePath(self.mktemp()))
def die():
self.fail("schema refreshed unnecessarily called too much")
s1._startup = die
A(store=s1)
|