Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Manager, Engineer = self.classes("Manager", "Engineer")
self._mapping_fixture(False, passive_updates)
sess = sa.orm.sessionmaker()()
m1 = Manager(name="dogbert", paperwork="lots")
e1, e2 = (
Engineer(name="dilbert", primary_language="java", boss=m1),
Engineer(name="wally", primary_language="c++", boss=m1),
)
sess.add_all([e1, e2, m1])
sess.commit()
eq_(e1.boss_name, "dogbert")
eq_(e2.boss_name, "dogbert")
eq_(
sess.execute(
self.tables.engineer.select().order_by(Engineer.name)
).fetchall(),
[("dilbert", "java", "dogbert"), ("wally", "c++", "dogbert")],
)
sess.expire_all()
m1.name = "pointy haired"
e1.primary_language = "scala"
e2.primary_language = "cobol"
sess.commit()
eq_(e1.boss_name, "pointy haired")
sess.add(o1)
sess.commit()
o1.name = "pointy haired"
o1.owner_name = "pointy"
sess.commit()
eq_(
sess.execute(self.tables.manager.select()).fetchall(),
[("pointy haired", None)],
)
eq_(
sess.execute(self.tables.owner.select()).fetchall(),
[("pointy haired", "pointy")],
)
eq_(o1.name, "pointy haired")
o1.name = "catbert"
sess.commit()
eq_(o1.name, "catbert")
eq_(
sess.execute(self.tables.manager.select()).fetchall(),
[("catbert", None)],
)
eq_(
sess.execute(self.tables.owner.select()).fetchall(),
[("catbert", "pointy")],
)
def test_clear(self):
sess = Session()
f1 = Foo(data={"a": "b"})
sess.add(f1)
sess.commit()
f1.data.clear()
sess.commit()
eq_(f1.data, {})
Engineer,
engineers_table,
inherits=employee_mapper,
concrete=True,
polymorphic_identity="engineer",
)
session = create_session()
tom = ManagerWHybrid("Tom", "mgrdata")
# mapping did not impact the engineer_info
# hybrid in any way
eq_(test_calls.mock_calls, [])
eq_(tom.engineer_info, "mgrdata")
eq_(test_calls.mock_calls, [mock.call.engineer_info_instance()])
session.add(tom)
session.flush()
session.close()
tom = (
session.query(ManagerWHybrid)
.filter(ManagerWHybrid.engineer_info == "mgrdata")
.one()
)
eq_(
test_calls.mock_calls,
[
mock.call.engineer_info_instance(),
mock.call.engineer_info_class(),
lazy_posts.return_value = attributes.PASSIVE_NO_RESULT
b = Blog("blog 1")
p = Post("post 1")
p2 = Post("post 2")
p.blog = b
eq_(lazy_posts.call_count, 1)
lazy_posts.return_value = [p, p2]
# lazy loaded + pending get added together.
# This isn't seen often with the ORM due
# to usual practices surrounding the
# load/flush/load cycle.
eq_(b.posts, [p, p2, p])
eq_(lazy_posts.call_count, 2)
users, addresses, dingalings = self.tables.users, \
self.tables.email_addresses, self.tables.dingalings
table_names = ['users', 'email_addresses']
if table_type == 'view':
table_names = ['users_v', 'email_addresses_v']
insp = inspect(meta.bind)
for table_name, table in zip(table_names, (users,
addresses)):
schema_name = schema
cols = insp.get_columns(table_name, schema=schema_name)
self.assert_(len(cols) > 0, len(cols))
# should be in order
for i, col in enumerate(table.columns):
eq_(col.name, cols[i]['name'])
ctype = cols[i]['type'].__class__
ctype_def = col.type
if isinstance(ctype_def, sa.types.TypeEngine):
ctype_def = ctype_def.__class__
# Oracle returns Date for DateTime.
if testing.against('oracle') and ctype_def \
in (sql_types.Date, sql_types.DateTime):
ctype_def = sql_types.Date
# assert that the desired type and return type share
# a base within one of the generic types.
self.assert_(len(set(ctype.__mro__).
intersection(ctype_def.__mro__).
def test_populate_dict(self):
uowcommit, a1, b1, a_mapper, b_mapper = self._fixture()
a1.obj().id = 10
pairs = [(a_mapper.c.id, b_mapper.c.id)]
dest = {}
sync.populate_dict(a1, a_mapper, dest, pairs)
eq_(dest, {"id": 10})
]
),
"Columns of reflected table didn't equal expected " "columns",
)
eq_(table.c.question.primary_key, True)
# disabled per http://www.sqlalchemy.org/trac/ticket/1660
# eq_(table.c.question.sequence.name, 'gen_testtable_id')
assert isinstance(table.c.question.type, Integer)
eq_(table.c.question.server_default.arg.text, "42")
assert isinstance(table.c.answer.type, String)
assert table.c.answer.type.length == 255
eq_(table.c.answer.server_default.arg.text, "'no answer'")
assert isinstance(table.c.remark.type, Text)
eq_(table.c.remark.server_default.arg.text, "''")
assert isinstance(table.c.photo.type, BLOB)
assert table.c.redundant.server_default is None
# The following assume a Dialect 3 database
assert isinstance(table.c.d.type, Date)
assert isinstance(table.c.t.type, Time)
assert isinstance(table.c.dt.type, DateTime)
def test_anon_aliased_overlapping(self):
text1 = self.tables.text1
c1 = text1.c.a.label(None)
c2 = text1.alias().c.a
c3 = text1.alias().c.a.label(None)
c4 = text1.c.a.label(None)
stmt = text("select a, b, c, d from text1").columns(c1, c2, c3, c4)
result = testing.db.execute(stmt)
row = result.first()
eq_(row[c1], "a1")
eq_(row[c2], "b1")
eq_(row[c3], "c1")
eq_(row[c4], "d1")
def test_legacy_accept(self):
canary = Mock()
@event.listens_for(self.TargetOne, "event_three")
def handler1(x, y):
canary(x, y)
self.TargetOne().dispatch.event_three(4, 5, 6, 7)
eq_(canary.mock_calls, [call(4, 5)])