Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def testRun(self):
db = DAL(DEFAULT_URI, check_reserved=['all'])
db.define_table('person', Field('name', default="Michael"),Field('uuid'))
db.define_table('pet',Field('friend',db.person),Field('name'))
dbdict = db.as_dict(flat=True, sanitize=False)
assert isinstance(dbdict, dict)
uri = dbdict["uri"]
assert isinstance(uri, basestring) and uri
assert len(dbdict["tables"]) == 2
assert len(dbdict["tables"][0]["fields"]) == 3
assert dbdict["tables"][0]["fields"][1]["type"] == db.person.name.type
assert dbdict["tables"][0]["fields"][1]["default"] == db.person.name.default
db2 = DAL(**dbdict)
assert len(db.tables) == len(db2.tables)
assert hasattr(db2, "pet") and isinstance(db2.pet, Table)
assert hasattr(db2.pet, "friend") and isinstance(db2.pet.friend, Field)
drop(db.pet)
db.commit()
db2.commit()
have_serializers = True
try:
import serializers
dbjson = db.as_json(sanitize=False)
assert isinstance(dbjson, basestring) and len(dbjson) > 0
unicode_keys = True
if sys.version < "2.6.5":
assert not (str(db5) in ("", None))
dbdict6 = {
"uri": DEFAULT_URI,
"tables": [
{"tablename": "staff"},
{
"tablename": "tvshow",
"fields": [
{"fieldname": "name"},
{"fieldname": "rating", "type": "double"},
],
},
],
}
db6 = DAL(**dbdict6)
assert len(db6["staff"].fields) == 1
assert "name" in db6["tvshow"].fields
assert db6.staff.insert() is not None
assert db6(db6.staff).select().first().id == 1
db6.staff.drop()
db6.tvshow.drop()
db6.commit()
db.close()
db2.close()
db4.close()
db5.close()
db6.close()
def testRun(self):
# check connection is no longer active after close
db = DAL(DEFAULT_URI, check_reserved=["all"])
connection = db._adapter.connection
db.close()
self.assertRaises(Exception, connection.commit)
# check connection are reused with pool_size
connections = set()
for a in range(10):
db2 = DAL(DEFAULT_URI, check_reserved=["all"], pool_size=5)
c = db2._adapter.connection
connections.add(c)
db2.close()
self.assertEqual(len(connections), 1)
c = connections.pop()
c.commit()
c.close()
# check correct use of pool_size
def testSelect(self):
db = DAL(DEFAULT_URI, check_reserved=['all'])
db.define_table(
'a_table',
Field('b_field'),
Field('a_field'),
)
db.a_table.insert(a_field="aa1", b_field="bb1")
rtn = db(db.a_table).select(db.a_table.id, db.a_table.b_field, db.a_table.a_field).as_list()
self.assertEqual(rtn[0]['b_field'], 'bb1')
keys = rtn[0].keys()
self.assertEqual(len(keys), 3)
self.assertEqual(("id" in keys, "b_field" in keys, "a_field" in keys), (True, True, True))
drop(db.a_table)
db.close()
def testJoin(self):
db = self.connect()
rname = "this_is_field_aa"
rname2 = "this_is_field_b"
db.define_table("t1", Field("aa", rname=rname))
db.define_table("t2", Field("aa", rname=rname), Field("b", db.t1, rname=rname2))
i1 = db.t1.insert(aa="1")
i2 = db.t1.insert(aa="2")
i3 = db.t1.insert(aa="3")
db.t2.insert(aa="4", b=i1)
db.t2.insert(aa="5", b=i2)
db.t2.insert(aa="6", b=i2)
self.assertEqual(
len(db(db.t1.id == db.t2.b).select(orderby=db.t1.aa | db.t2.aa)), 3
)
self.assertEqual(
db(db.t1.id == db.t2.b).select(orderby=db.t1.aa | db.t2.aa)[2].t1.aa, "2"
)
self.assertEqual(
db(db.t1.id == db.t2.b).select(orderby=db.t1.aa | db.t2.aa)[2].t2.aa, "6"
)
self.assertEqual(
def testListReference(self):
db = self.connect()
db.define_table("t0", Field("aa", "string"))
db.define_table("tt", Field("t0_id", "list:reference t0"))
id_a1 = db.t0.insert(aa="test1")
id_a2 = db.t0.insert(aa="test2")
ref1 = [id_a1]
ref2 = [id_a2]
ref3 = [id_a1, id_a2]
db.tt.insert(t0_id=ref1)
self.assertEqual(db(db.tt).select(db.tt.t0_id).last()[db.tt.t0_id], ref1)
db.tt.insert(t0_id=ref2)
self.assertEqual(db(db.tt).select(db.tt.t0_id).last()[db.tt.t0_id], ref2)
db.tt.insert(t0_id=ref3)
self.assertEqual(db(db.tt).select(db.tt.t0_id).last()[db.tt.t0_id], ref3)
self.assertEqual(db(db.tt.t0_id == ref3).count(), 1)
def testRowExtra(self):
db = self.connect(check_reserved=None, lazy_tables=True)
tt = db.define_table("tt", Field("value", "integer"))
db.tt.insert(value=1)
row = db(db.tt).select("value").first()
self.assertEqual(row.value, 1)
def testRun(self):
db = DAL(None)
db.define_table("no_table", Field("aa"))
self.assertIsInstance(db.no_table.aa, Field)
self.assertIsInstance(db.no_table["aa"], Field)
db.close()
rtn = db(db.easy_name.a_field == "c").count()
self.assertEqual(rtn, 1)
rtn = db(db.easy_name.a_field != "c").count()
self.assertEqual(rtn, 1)
avg = db.easy_name.id.avg()
rtn = db(db.easy_name.id > 0).select(avg)
self.assertEqual(rtn[0][avg], 3)
avg = db.easy_name.rating.avg()
rtn = db(db.easy_name.id > 0).select(avg)
self.assertEqual(rtn[0][avg], 2)
rname = "this_is_the_person_name"
db.define_table(
"person",
Field("id", type="id", rname="fooid"),
Field("name", default="Michael", rname=rname),
Field("uuid"),
)
rname = "this_is_the_pet_name"
db.define_table(
"pet", Field("friend", "reference person"), Field("name", rname=rname)
)
michael = db.person.insert() # default insert
john = db.person.insert(name="John")
luke = db.person.insert(name="Luke")
# michael owns Phippo
phippo = db.pet.insert(friend=michael, name="Phippo")
# john owns Dunstin and Gertie
dunstin = db.pet.insert(friend=john, name="Dunstin")
gertie = db.pet.insert(friend=john, name="Gertie")
def testRun(self):
db = DAL(DEFAULT_URI, check_reserved=['all'])
db.define_table('tt', Field('aa', 'reference'))
with self.assertRaises(ValueError):
db.tt.insert(aa='x')
with self.assertRaises(ValueError):
db.tt.insert(aa='_')
with self.assertRaises(TypeError):
db.tt.insert(aa=3.1)
self.assertEqual(isinstance(db.tt.insert(aa=''), long), True)
self.assertEqual(isinstance(db.tt.insert(aa='1'), long), True)
self.assertEqual(isinstance(db.tt.insert(aa='0x1'), long), True)
with self.assertRaises(RuntimeError):
db(db.tt.aa+1==1).update(aa=0)
drop(db.tt)
db.define_table('tt', Field('aa', 'date'))
self.assertEqual(isinstance(db.tt.insert(aa=None), long), True)