AFFAANh
2e46f8485d
_transition previously called update_campaign then add_event as two separate commits, so a failed add_event left a status change with no audit row. Phase 2 launches campaigns that authorise spend, so a compliance hole like that has to close first. Add CampaignRepository.update_campaign_with_event, which runs the UPDATE and the event INSERT on one connection inside one transaction and commits once. Route campaign transitions (submit/approve/reject) through it instead of the update_campaign + add_event pair; update_campaign and add_event themselves are unchanged for other callers. Also update test_campaign_api.py's FakeRepo/monkeypatch to implement the new repository method, since production code now calls it in the submit/approve/reject path.
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
|