def test_transaction_aware_task_success(celery_worker, task_app_request, dbsession, demo_user):
"""Transaction aware tasks works in eager mode.."""
with transaction.manager:
# Do a dummy database write
u = dbsession.query(User).first()
demotasks.modify_username.apply_async([u.id], tm=transaction.manager)
# Let the task travel in Celery queue
time.sleep(2.0)
# Task should not fire unless we exit the transaction
assert u.username != "set by celery"
# Let the transaction commit
time.sleep(0.5)
# Task has now fired after transaction was committed
with transaction.manager:
u = dbsession.query(User).get(1)
assert u.username == "set by celery"
python类manager()的实例源码
def test_transaction_manager_abort(celery_worker, task_app_request, dbsession, demo_user):
"""Test that transaction are not executed to schedule if transaction manager aborts."""
try:
# Tasks do not execute if transaction is never commit because of exception
with transaction.manager:
assert dbsession.query(User).count() == 1
assert dbsession.query(User).get(1).username == "test"
demotasks.modify_username.apply_async(args=[1], tm=transaction.manager)
raise RuntimeError("aargh")
except RuntimeError:
pass
# Let the transaction commit (which should not happen)
time.sleep(0.5)
# Because of exception, Celery never fires
with transaction.manager:
u = dbsession.query(User).get(1)
assert u.username == "test"
def test_manual_transaction(celery_worker, task_app_request, dbsession, demo_user):
"""Manual transaction lifecycles within task work."""
with transaction.manager:
# Do a dummy database write
u = dbsession.query(User).first()
demotasks.modify_username_manual_transaction.apply_async(kwargs={"user_id": u.id}, tm=transaction.manager)
# Let the task travel in Celery queue
time.sleep(1.0)
# Task should not fire unless we exit the transaction
assert u.username != "set by celery"
# Let the transaction commit
time.sleep(0.5)
# Task has now fired after transaction was committed
with transaction.manager:
u = dbsession.query(User).first()
assert u.username == "set by celery"
def test_login(web_server, browser, dbsession, init):
"""Login an user."""
with transaction.manager:
create_user(dbsession, init.config.registry)
b = browser
b.visit(web_server)
b.click_link_by_text("Sign in")
assert b.is_element_present_by_css("#login-form")
b.fill("username", EMAIL)
b.fill("password", PASSWORD)
b.find_by_name("login_email").click()
# After login we see a profile link to our profile
assert b.is_element_present_by_css("#nav-logout")
def test_logout(web_server, browser, dbsession, init):
"""Log out."""
with transaction.manager:
create_user(dbsession, init.config.registry)
b = browser
b.visit("{}/{}".format(web_server, "login"))
assert b.is_element_present_by_css("#login-form")
b.fill("username", EMAIL)
b.fill("password", PASSWORD)
b.find_by_name("login_email").click()
assert b.is_element_present_by_css("#msg-you-are-logged-in")
b.find_by_css("#nav-logout").click()
# Anonynous again
assert b.is_element_present_by_css("#msg-logged-out")
assert not b.is_element_present_by_css("#nav-logout")
# We should see the log in form
assert b.is_element_present_by_css("#login-form")
def test_forget_password_disabled_user(web_server, browser, dbsession, init):
"""Reset password by email."""
with transaction.manager:
u = create_user(dbsession, init.config.registry)
u.enabled = False
b = browser
b.visit(web_server + "/login")
assert b.is_element_present_by_css("#login-form")
b.click_link_by_text("Forgot your password?")
assert b.is_element_present_by_css("#forgot-password-form")
b.fill("email", EMAIL)
b.find_by_name("submit").click()
assert b.is_element_present_by_css("#msg-cannot-reset-password")
def test_login_forget_password_email_send(web_server, browser, dbsession, init):
"""Send out the reset password by email, but do not answer to it, instead directly login."""
with transaction.manager:
create_user(dbsession, init.config.registry)
b = browser
b.visit(web_server)
b.find_by_css("#nav-sign-in").click()
assert b.is_element_present_by_css("#login-form")
b.click_link_by_text("Forgot your password?")
assert b.is_element_present_by_css("#forgot-password-form")
b.fill("email", EMAIL)
b.find_by_name("submit").click()
b.visit("{}/login".format(web_server))
b.fill("username", EMAIL)
b.fill("password", PASSWORD)
b.find_by_name("login_email").click()
assert b.is_element_present_by_css("#msg-you-are-logged-in")
def test_forget_password_expired_token(web_server, browser, dbsession, init):
"""Reset password by email."""
with transaction.manager:
create_user(dbsession, init.config.registry)
b = browser
b.visit(web_server + "/forgot-password")
assert b.is_element_present_by_css("#forgot-password-form")
b.fill("email", EMAIL)
b.find_by_name("submit").click()
assert b.is_element_present_by_css("#msg-check-email")
with transaction.manager:
user = get_user(dbsession)
activation = user.activation
activation.expires_at = now() - timedelta(days=365)
activation_code = activation.code
b.visit("{}/reset-password/{}".format(web_server, activation_code))
assert b.is_element_present_by_css("#not-found")
def test_enter_admin(web_server, browser, dbsession, init):
"""The first user can open the admin page."""
with transaction.manager:
u = create_user(dbsession, init.config.registry)
site_creator = get_site_creator(init.config.registry)
site_creator.init_empty_site(dbsession, u)
assert u.is_admin()
b = browser
b.visit(web_server + "/login")
b.fill("username", EMAIL)
b.fill("password", PASSWORD)
b.find_by_name("login_email").click()
assert b.is_element_visible_by_css("#nav-admin")
b.find_by_css("#nav-admin").click()
assert b.is_element_present_by_css("#admin-main")
def create_logged_in_user(dbsession:Session, registry:Registry, web_server:str, browser:DriverAPI, admin:bool=False, email:str=EMAIL, password:str=PASSWORD):
"""For a web browser test session, creates a new user and log it in inside the test browser."""
# Catch some common argument misordering issues
assert isinstance(registry, Registry)
assert isinstance(web_server, str)
with transaction.manager:
create_user(dbsession, registry, admin=admin, email=email, password=password)
b = browser
b.visit("{}/{}".format(web_server, "login"))
assert b.is_element_present_by_css("#login-form")
b.fill("username", email)
b.fill("password", password)
b.find_by_name("login_email").click()
# After login we log out link to confirm login has succeeded
assert b.is_element_present_by_css("#nav-logout")
def test_add_choice_choose_no_question(browser: DriverAPI, registry, web_server, dbsession):
from .tutorial import Question
from .tutorial import Choice
with transaction.manager:
q = Question(question_text="What is love")
dbsession.add(q)
dbsession.flush()
b = browser
create_logged_in_user(dbsession, registry, web_server, browser, admin=True)
b.visit(web_server)
b.find_by_css("#nav-admin").click()
b.find_by_css("#btn-panel-add-choice").click()
b.fill("choice_text", "Baby don't hurt me")
b.find_by_name("add").click()
assert b.is_element_present_by_css("#msg-item-added")
with transaction.manager:
assert dbsession.query(Choice).first().question is None
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
