def test_rowmethod():
session = Session()
with pytest.raises(ValueError) as e:
ModelStub.my_row_method(0, 1, 2)
c, s, a = ModelStub.my_row_method(0, session, 1)
assert c is 0
assert s is session
assert a == 1
i = ModelStub()
c, s, a = i.my_row_method(session, 2)
assert c is i
assert s is session
assert a == 2
python类Session()的实例源码
def create_session(self):
"""
Create a session context that communicates with the database.
Commits all changes to the database before closing the session, and if an exception is raised,
rollback the session.
"""
session = Session(self.connection)
logger.info("Create session {0} with {1}".format(
id(session), self._public_db_uri(str(self.engine.url))))
try:
yield session
session.commit()
logger.info("Commit transactions to database")
except Exception:
session.rollback()
logger.exception("Database transactions rolled back")
finally:
logger.info("Session {0} with {1} closed".format(
id(session), self._public_db_uri(str(self.engine.url))))
session.close()
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 7), self.mock_sess)
m_account = Mock(name='foo')
type(m_account).name = 'foo'
m_budget = Mock(name='bar')
type(m_budget).name = 'bar'
self.m_st = Mock(
spec_set=ScheduledTransaction,
id=123,
description='desc',
amount=Decimal('123.45'),
account_id=2,
account=m_account,
budget_id=3,
budget=m_budget
)
def init_empty_site(self, dbsession:Session, user:UserMixin):
"""When the first user signs up build the admin groups and make the user member of it.
Make the first member of the site to be admin and superuser.
"""
# Try to reflect related group class based on User model
i = inspection.inspect(user.__class__)
Group = i.relationships["groups"].mapper.entity
# Do we already have any groups... if we do we probably don'¨t want to init again
if dbsession.query(Group).count() > 0:
return
g = Group(name=Group.DEFAULT_ADMIN_GROUP_NAME)
dbsession.add(g)
g.users.append(user)
def filled_database(file_database) -> curse.Database:
"""Database with some mods filled in."""
# Create structure
addon.AddonBase.metadata.create_all(file_database.engine)
# Add few mods
session = SQLSession(bind=file_database.engine)
session.add_all([
addon.Mod(id=42, name='tested', summary="Mod under test"),
addon.Mod(id=45, name='tester', summary="Validate tested mod"),
addon.Mod(id=3, name='unrelated', summary="Dummy"),
])
session.commit()
return file_database
def find(cls, connection: SQLSession, name: str) -> 'Mod':
"""Find exactly one Mod named NAME.
Keyword Arguments:
connection: Database connection to ask on.
name: The name of the mod to search for.
Returns:
The requested mod.
Raises:
NoResultsFound: The name does not match any known mod.
MultipleResultsFound: The name is too ambiguous,
multiple matching mods found.
"""
query = SQLBakery(lambda conn: conn.query(cls))
query += lambda q: q.filter(cls.name.like(bindparam('name')))
return query(connection).params(name='{}'.format(name)).one()
def with_id(cls, connection: SQLSession, id: int) -> 'Mod':
"""Fetch mod with id from database.
Keyword arguments:
connection: Database connection to fetch from.
id: The id field of the mod to get.
Returns:
The requested mod.
Raises:
NoResultFound: Mod with specified id does not exist.
"""
query = SQLBakery(lambda conn: conn.query(cls))
query += lambda q: q.filter(cls.id == bindparam('id'))
return query(connection).params(id=id).one()
def get_tm_session(session_factory, transaction_manager):
"""
Get a ``sqlalchemy.orm.Session`` instance backed by a transaction.
This function will hook the session to the transaction manager which
will take care of committing any changes.
- When using pyramid_tm it will automatically be committed or aborted
depending on whether an exception is raised.
- When using scripts you should wrap the session in a manager yourself.
For example::
import transaction
engine = get_engine(settings)
session_factory = get_session_factory(engine)
with transaction.manager:
dbsession = get_tm_session(session_factory, transaction.manager)
"""
dbsession = session_factory()
zope.sqlalchemy.register(
dbsession, transaction_manager=transaction_manager)
return dbsession
def _delete_all_attached(session: Session, machine: Machine):
"""
Delete all resources attached to a machine
As we don't need performance we can avoid heuristics by dropping and re-creating theses needed resources
The discovery data is the reference of the reality
:param session:
:param machine:
:return:
"""
session.query(MachineDisk) \
.filter(MachineDisk.machine_id == machine.id) \
.delete()
all_mi = session.query(MachineInterface) \
.filter(MachineInterface.machine_id == machine.id)
for i in all_mi:
session.query(ChassisPort) \
.filter(ChassisPort.machine_interface == i.id) \
.delete()
session.delete(i)
session.flush()
def test_modelmethod():
session = Session()
with pytest.raises(ValueError) as e:
ModelStub.my_model_method(1, 2)
c, s, a = ModelStub.my_model_method(session, 1)
assert c is ModelStub
assert s is session
assert a == 1
def test_default_store(self):
with StoreManager(Session) as manager1:
# Hacking StoreManager to raise error
StoreManager._default = None
self.assertRaises(DefaultStoreError, manager1.get)
# making it default again
StoreManager.make_default('dummy')
self.assertIsNotNone(manager1.get())
# unregister
StoreManager.unregister('dummy')
def test_context_stack(self):
self.assertRaises(ContextError, StoreManager.get_current_store_manager)
with StoreManager(Session) as manager1:
store1 = manager1.get()
self.assertIs(store1, manager1.default_store)
with StoreManager(Session) as manager2:
store2 = manager2.get()
self.assertIs(store2, manager2.default_store)
self.assertIsNot(manager1, manager2)
self.assertIsNot(store1, store2)
def test_empty_blog(web_server: str, browser: DriverAPI, dbsession: Session):
"""We can render empty blog."""
# Direct Splinter browser to the website
b = browser
b.visit(web_server + "/blog/")
# After login we see a profile link to our profile
assert b.is_element_visible_by_css("#blog-no-posts")
def test_no_unpublished_in_blog_roll(web_server: str, browser: DriverAPI, dbsession: Session, unpublished_post_id):
"""Visitors should not see unpublished posts in blog roll."""
# Direct Splinter browser to the website
b = browser
b.visit(web_server + "/blog/")
# After login we see a profile link to our profile
assert b.is_element_visible_by_css("#blog-no-posts")
def test_published_excerpt(web_server: str, browser: DriverAPI, dbsession: Session, published_post_id):
"""When posts are published they become visible in blog roll."""
# Direct Splinter browser to the website
b = browser
b.visit(web_server + "/blog/")
# After login we see a profile link to our profile
assert b.is_element_visible_by_css(".excerpt")
def test_published_post(web_server: str, browser: DriverAPI, dbsession: Session, published_post_id):
"""User can view blog posts."""
b = browser
b.visit(web_server + "/blog/")
assert b.is_element_present_by_css("#heading-blog")
# Go to the published post from the roll
b.find_by_css(".post-link").click()
assert b.is_element_present_by_css("#heading-post")
def test_empty_tag_roll(web_server: str, browser: DriverAPI, dbsession: Session):
"""We can render empty tag list."""
# Direct Splinter browser to the website
b = browser
b.visit(web_server + "/blog/tag/xxx")
assert b.is_element_present_by_css("#blog-no-posts")
def create_session(self,
bind,
options=None,
session_class=Session,
query_class=SQLQuery):
"""Factory function to create a scoped session using `bind`.
Args:
bind (Engine|Connection): Database engine or connection instance.
options (dict, optional): Session configuration options.
session_class (obj, optional): Session class to use when creating
new session instances. Defaults to :class:`.Session`.
query_class (obj, optional): Query class used for ``session.query``
instances. Defaults to :class:`.SQLQuery`.
Returns:
Session: SQLAlchemy session instance bound to `bind`.
"""
if options is None: # pragma: no cover
options = {}
if query_class:
options['query_cls'] = query_class
scopefunc = options.pop('scopefunc', None)
session_factory = orm.sessionmaker(bind=bind,
class_=session_class,
**options)
return orm.scoped_session(session_factory, scopefunc=scopefunc)
def session(self, Session):
"""Set private :attr:`_Session`."""
self._Session = Session
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 17), self.mock_sess)
def setup(self):
self.mock_sess = Mock(spec_set=Session)
self.cls = BiweeklyPayPeriod(date(2017, 3, 7), self.mock_sess)
