39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
from unittest.mock import MagicMock
|
|
import pytest
|
|
from app.workflows import service
|
|
|
|
def install(monkeypatch):
|
|
conn=MagicMock()
|
|
release=MagicMock()
|
|
monkeypatch.setattr(service,'get_db_connection',lambda:conn)
|
|
monkeypatch.setattr(service,'release_db_connection',release)
|
|
return conn,release
|
|
|
|
def test_single_commit_and_release(monkeypatch):
|
|
conn,release=install(monkeypatch)
|
|
with service.transaction() as cursor:
|
|
cursor.execute('first')
|
|
cursor.execute('second')
|
|
conn.commit.assert_not_called()
|
|
conn.commit.assert_called_once()
|
|
conn.rollback.assert_not_called()
|
|
release.assert_called_once_with(conn)
|
|
|
|
def test_all_writes_rollback_on_failure(monkeypatch):
|
|
conn,release=install(monkeypatch)
|
|
with pytest.raises(RuntimeError):
|
|
with service.transaction() as cursor:
|
|
cursor.execute('first')
|
|
raise RuntimeError('failed second effect')
|
|
conn.commit.assert_not_called()
|
|
conn.rollback.assert_called_once()
|
|
release.assert_called_once_with(conn)
|
|
|
|
def test_enqueue_is_deduplicated_by_revision(monkeypatch):
|
|
cursor=MagicMock()
|
|
cursor.fetchone.return_value=None
|
|
assert service.emit_time_event_on_connection(cursor,42,7) is None
|
|
sql,values=cursor.execute.call_args.args
|
|
assert 'ON CONFLICT(event_key) DO NOTHING' in sql
|
|
assert values[1]=='time:42:7'
|