MCPcopy
hub / github.com/sqlalchemy/sqlalchemy / _test_instance_guards

Method _test_instance_guards

test/orm/test_session.py:2543–2602  ·  view source on GitHub ↗
(self, user_arg)

Source from the content-addressed store, hash-verified

2541 )
2542
2543 def _test_instance_guards(self, user_arg):
2544 watchdog = set()
2545
2546 def x_raises_(obj, method, *args, **kw):
2547 watchdog.add(method)
2548 callable_ = getattr(obj, method)
2549 assert_raises(
2550 sa.orm.exc.UnmappedInstanceError, callable_, *args, **kw
2551 )
2552
2553 def raises_(method, *args, **kw):
2554 x_raises_(fixture_session(), method, *args, **kw)
2555
2556 for name in [
2557 "__contains__",
2558 "is_modified",
2559 "merge",
2560 "refresh",
2561 "add",
2562 "delete",
2563 "expire",
2564 "expunge",
2565 "enable_relationship_loading",
2566 ]:
2567 raises_(name, user_arg)
2568
2569 for name in ["add_all", "merge_all", "delete_all"]:
2570 raises_(name, (user_arg,))
2571
2572 # flush will no-op without something in the unit of work
2573 def _():
2574 class OK:
2575 pass
2576
2577 self._map_it(OK)
2578
2579 s = fixture_session()
2580 s.add(OK())
2581 with assertions.expect_deprecated(
2582 "The `objects` parameter of `Session.flush` is deprecated"
2583 ):
2584 x_raises_(s, "flush", objects=(user_arg,))
2585
2586 _()
2587
2588 instance_methods = (
2589 self._public_session_methods()
2590 - self._class_methods
2591 - {
2592 "bulk_update_mappings",
2593 "bulk_insert_mappings",
2594 "bulk_save_objects",
2595 }
2596 )
2597
2598 eq_(
2599 watchdog,
2600 instance_methods,

Calls 3

eq_Function · 0.90
symmetric_differenceMethod · 0.45

Tested by

no test coverage detected