def test_immutable_entity_view_field_value(self):
     foo1 = ex(db.Foo.t.create(name='foo1'))
     foo2 = ex(db.Foo.t.create(name='foo2'))
     bar = ex(db.Bar.t.create(foo_list=[foo1]))
     assert raises(AttributeError, getattr, bar.foo_list, 'append')
     v = bar.v.default()
     assert raises(AttributeError, getattr, v.foo_list, 'append')

 def test_find_extremes(self):
     bt = BTree()
     assert raises(AssertionError, bt.get_min_item)
     assert raises(AssertionError, bt.get_max_item)
     map(bt.add, range(100))
     assert bt.get_min_item() == (0, True)
     assert bt.get_max_item() == (99, True)

 def test_only_one_top_level(self):
     # Cannot pass more than one top-level transaction.
     tx1 = db.User.t.create(name='1')
     tx2 = db.User.t.create(name='2')
     assert raises(RuntimeError, db.execute, tx1, tx2)
     # Must pass at least one top-level transaction.
     assert raises(RuntimeError, db.execute)

 def test_extra_fields(self):
     # A normal Gender create does not normally have fget fields in
     # the transaction, but in ProblemGender we've inserted them as
     # fields of different types.
     tx = db.ProblemGender.t.create()
     assert sorted(tx.s.field_map().keys()) == [
         'code', 'count', 'foo', 'name']
     # When executing the transaction, the superclass T.Create
     # should ignore .count since it was an fget field.
     tx.code = 'X'
     tx.name = 'xyz'
     tx.count = 'foo'
     tx.foo = 5
     pgender = db.execute(tx)
     # Accessing the count directly should result in a calculated
     # value.
     assert pgender.count == 0
     # Peek in the database to make sure that it didn't get stored
     # in the database as a string though, since in cases where the
     # type of the calculated field is an entity field
     assert db._entity_field(
         'ProblemGender', pgender.s.oid, 'count') != 'foo'
     assert raises(KeyError, db._entity_field, 'ProblemGender',
                   pgender.s.oid, 'foo')
     # Same thing for updates.
     tx = pgender.t.update()
     tx.count = 'bar'
     tx.foo = 10
     pgender = db.execute(tx)
     assert pgender.count == 0
     assert db._entity_field(
         'ProblemGender', pgender.s.oid, 'count') != 'foo'
     assert raises(KeyError, db._entity_field, 'ProblemGender',
                   pgender.s.oid, 'foo')

 def test_transfer_account_suspended(self):
     betty = db.Person.findone(name='Betty Rubble')
     family = db.Account.findone(owner=betty, name='Family')
     savings = db.Account.findone(owner=betty, name='Savings')
     # Attempt transfer from family to savings.
     tx = family.t.transfer()
     tx.to_account = savings
     tx.amount = 0.01
     assert raises(Exception, db.execute, tx)
     # Attempt transfer from savings to family.
     tx = savings.t.transfer()
     tx.to_account = family
     tx.amount = 0.01
     assert raises(Exception, db.execute, tx)

 def test_remove_entity_field_then_readd(self):
     schema1 = fix("""
     class Foo(E.Entity):
         bar = f.entity('Bar')
         baz = f.integer()
     class Bar(E.Entity):
         bof = f.string()
     """)
     schema2 = fix("""
     class Foo(E.Entity):
         baz = f.integer()
     class Bar(E.Entity):
         bof = f.string()
     """)
     schema3 = fix("""
     class Foo(E.Entity):
         bar = f.entity('Bar')
         baz = f.integer()
     class Bar(E.Entity):
         bof = f.string()
     """)
     self.sync(schema1)
     bar = db.execute(db.Bar.t.create(bof='fob'))
     foo = db.execute(db.Foo.t.create(bar=bar, baz=12))
     foo_oid = foo.s.oid
     bar_oid = bar.s.oid
     assert bar.s.count() == 1
     # Evolve.
     self.evolve(schema2, version=2)
     foo = db.Foo[foo_oid]
     bar = db.Bar[bar_oid]
     assert foo.baz == 12
     raises(AttributeError, getattr, foo, 'bar')
     assert bar.s.count() == 0
     # Attempt to update the entity.
     tx = foo.t.update()
     tx.baz = 5
     db.execute(tx)
     assert foo.baz == 5
     # Evolve.
     self.evolve(schema3, version=3)
     foo = db.Foo[foo_oid]
     bar = db.Bar[bar_oid]
     assert foo.bar is UNASSIGNED
     assert foo.baz == 5
     tx = foo.t.update()
     tx.bar = bar
     db.execute(tx)
     assert foo.bar == bar
     assert bar.s.count() == 1

 def test_database_decoration(self):
     # This label is assigned automatically.
     assert label.label(db) == u'Schevo Database'
     # It can be easily overridden.
     label.relabel(db, 'Custom Label')
     assert label.label(db) == u'Custom Label'
     # When reopening the database, the label persists.
     self.reopen()
     assert label.label(db) == u'Custom Label'
     # Cannot change the label during a transaction.
     def fn(db):
         label.relabel(db, u'Custom Label 2')
     tx = transaction.CallableWrapper(fn)
     raises(error.DatabaseExecutingTransaction,
            db.execute, tx)

 def test_pop(self):
     assert raises(KeyError, set_type().pop)
     s1 = set_type('asdf')
     x = s1.pop()
     assert x not in s1
     assert len(s1) == 3
     assert (s1 | set_type(x)) == set_type('asdf')

 def test_remove(self):
     s1 = set_type()
     assert raises(KeyError, s1.remove, 1)
     assert s1 == set_type()
     s1 = set_type('asdf')
     s1.remove('a')
     assert s1 == set_type('sdf')

 def test_delete_cannot_skip_revisions(self):
     tx = db.User.t.create(name='foo')
     user = db.execute(tx)
     tx1 = user.t.update(name='bar')
     tx2 = user.t.delete()
     db.execute(tx1)
     assert raises(error.TransactionExpired, db.execute, tx2)

 def test_a(self):
     filename = tempfile.mktemp()
     s = FileStorage2(filename)
     connection = Connection(s)
     root = connection.get_root()
     root['a'] = Persistent()
     root['a'].b = 1
     connection.commit()
     root['a'].b = 2
     connection.commit()
     s.close()
     hc = HistoryConnection(filename)
     a = hc.get_root()['a']
     assert len(hc.get_storage().index.history) == 4
     assert a.b == 2
     hc.previous()
     assert a.b == 1
     hc.next()
     assert a.b == 2
     hc.previous()
     assert a.b == 1
     hc.previous()
     assert a._p_is_ghost()
     assert not hasattr(a, '__dict__')
     assert isinstance(a, Persistent)
     assert raises(KeyError, getattr, a, 'b')
     assert hc.get(a._p_oid) is a
     hc.next()
     assert a.b == 1
     hc.get_storage().fp.close()
     os.unlink(filename)

 def test_remove_extent_restricted(self):
     schema1 = fix("""
     class Foo(E.Entity):
         bar = f.entity(allow='Bar')
     class Bar(E.Entity):
         baz = f.integer()
     """)
     schema2 = fix("""
     class Foo(E.Entity):
         bar = f.entity(allow='Bar')
     class Bof(E.Entity):
         baz = f.integer()
     """)
     self.sync(schema1)
     raises(error.ExtentDoesNotExist, self.sync, schema2)
     assert db.schema_source == schema1
     assert db.extent_names() == ['Bar', 'Foo']

 def test_transfer_insufficient_funds(self):
     fred = db.Person.findone(name='Fred Flintstone')
     business = db.Account.findone(owner=fred, name='Business')
     personal = db.Account.findone(owner=fred, name='Personal')
     # Attempt transfer from personal to business.
     tx = personal.t.transfer()
     tx.to_account = business
     tx.amount = 205.00
     assert raises(Exception, db.execute, tx)

 def test_field_requirements(self):
     tx = db.User.t.create()
     # User was not specified, so transaction shouldn't succeed.
     assert raises(AttributeError, db.execute, tx)
     # Even if the transaction's fields are modified, the entity's
     # field spec should remain enforced.
     tx.f.name.required = False
     assert raises(AttributeError, db.execute, tx)
     # Age should not be required though.
     tx = db.User.t.create()
     tx.name = 'foo'
     result = db.execute(tx)
     assert result.age is UNASSIGNED
     # When updating, restrictions should still be enforced.
     tx = result.t.update(name=UNASSIGNED)
     assert raises(AttributeError, db.execute, tx)
     tx.f.name.required = False
     assert raises(AttributeError, db.execute, tx)

 def test_unassign_when_unassigned_disallowed(self):
     bar1, bar2, bar3, bar4, bar5 = db.Bar.by('name')
     fee = ex(db.Fee.t.create(
         name='fee',
         bar_list=[bar1, bar2, bar3, bar4, bar5, bar4, bar3],
         ))
     assert list(fee.bar_list) == [bar1, bar2, bar3, bar4, bar5, bar4, bar3]
     call = ex, bar4.t.delete()
     assert raises(ValueError, *call)

 def test_disallow_duplicates(self):
     # Make sure that duplicates are allowed by default.
     foo = ex(db.Foo.t.create(name='foo'))
     bar = ex(db.Bar.t.create(foo_list=[foo, foo]))
     assert list(bar.foo_list) == [foo, foo]
     # Make sure they are disallowed when allow_duplicates is False.
     tx = db.Bee.t.create(foo_list=[foo, foo])
     call = ex, tx
     assert raises(ValueError, *call)

 def test_check_memory_storage(self):
     b = MemoryStorage()
     assert b.new_oid() == p64(1)
     assert b.new_oid() == p64(2)
     assert raises(KeyError, b.load, p64(0))
     record = pack_record(p64(0), 'ok', '')
     b.begin()
     b.store(p64(0), record)
     b.end()
     b.sync()
     b.begin()
     b.store(p64(1), pack_record(p64(1), 'no', ''))
     b.end()
     assert len(list(b.gen_oid_record())) == 2
     assert record == b.load(p64(0))
     records = b.bulk_load([p64(0), p64(1)])
     assert len(list(records)) == 2
     records = b.bulk_load([p64(0), p64(1), p64(2)])
     assert raises(KeyError, list, records)

 def test_allow_unassigned(self):
     # Make sure that UNASSIGNED members are disallowed by default.
     foo = ex(db.Foo.t.create(name='foo'))
     tx = db.Bar.t.create(foo_list=[foo, UNASSIGNED])
     call = ex, tx
     assert raises(ValueError, *call)
     # Make sure they are allowed when allow_unassigned is True.
     tx = db.Boo.t.create(foo_list=[foo, UNASSIGNED])
     boo = db.execute(tx)
     assert list(boo.foo_list) == [foo, UNASSIGNED]

 def test_remove_min_one(self):
     bar1 = db.Bar.findone(name=u'bar 1')
     bar2 = db.Bar.findone(name=u'bar 2')
     foo1 = db.Foo.findone(name=u'foo 1')
     db.execute(bar1.t.delete())
     assert list(foo1.bars) == [bar2]
     # Deleting bar2 should fail due to foo1.bars becoming an empty
     # list, which is not allowed.
     call = db.execute, bar2.t.delete()
     assert raises(ValueError, *call)

 def test_min_size_max_size(self):
     # Make sure that empty lists are allowed by default.
     foo = ex(db.Foo.t.create(name='foo'))
     tx = db.Bar.t.create(foo_list=[])
     bar = db.execute(tx)
     assert list(bar.foo_list) == []
     # Make sure they are not allowed when min_size > 0.
     tx = db.Baz.t.create(foo_list=[])
     call = ex, tx
     assert raises(ValueError, *call)