42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""The transactional write must issue both statements on one connection."""
|
||
|
|
import inspect
|
||
|
|
|
||
|
|
from adclaw.campaigns import repo as repo_module
|
||
|
|
|
||
|
|
|
||
|
|
def test_transactional_method_exists():
|
||
|
|
assert hasattr(repo_module.CampaignRepository, "update_campaign_with_event")
|
||
|
|
|
||
|
|
|
||
|
|
def test_transactional_method_opens_exactly_one_connection():
|
||
|
|
source = inspect.getsource(
|
||
|
|
repo_module.CampaignRepository.update_campaign_with_event,
|
||
|
|
)
|
||
|
|
assert source.count("connect_database()") == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_transactional_method_commits_once_after_both_statements():
|
||
|
|
source = inspect.getsource(
|
||
|
|
repo_module.CampaignRepository.update_campaign_with_event,
|
||
|
|
)
|
||
|
|
assert source.count("await conn.commit()") == 1
|
||
|
|
assert "maskanx_campaign_events" in source
|
||
|
|
assert "UPDATE maskanx_campaigns" in source
|
||
|
|
assert source.index("UPDATE maskanx_campaigns") < source.index("await conn.commit()")
|
||
|
|
assert source.index("maskanx_campaign_events") < source.index("await conn.commit()")
|
||
|
|
|
||
|
|
|
||
|
|
def test_transactional_method_rolls_back_on_error():
|
||
|
|
source = inspect.getsource(
|
||
|
|
repo_module.CampaignRepository.update_campaign_with_event,
|
||
|
|
)
|
||
|
|
assert "await conn.rollback()" in source
|
||
|
|
|
||
|
|
|
||
|
|
def test_transactional_method_raises_lookup_error_for_unknown_id():
|
||
|
|
source = inspect.getsource(
|
||
|
|
repo_module.CampaignRepository.update_campaign_with_event,
|
||
|
|
)
|
||
|
|
assert "LookupError" in source
|