28 lines
1.5 KiB
Python
28 lines
1.5 KiB
Python
|
import pytest
|
||
|
|
from app.workflows.service import is_incoming_mail
|
||
|
|
from test_workflow_db import db
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('context',[{'folder':'Sent Items','status':'new'},{'folder':'INBOX','status':'sent'},{'folder':'Drafts'},{'folder':'Outbox'},{'direction':'outbound'},{'direction':'outgoing'}])
|
||
|
|
def test_outbound_mail_never_triggers_received_workflow(context):
|
||
|
|
assert not is_incoming_mail(context)
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('context',[{'folder':'INBOX','status':'new'},{'folder':'Processed','status':'processed'},{'folder':'Archive'},{}])
|
||
|
|
def test_incoming_mail_can_trigger_received_workflow(context):
|
||
|
|
assert is_incoming_mail(context)
|
||
|
|
|
||
|
|
def test_database_capture_excludes_sent_mail_but_includes_inbound(db):
|
||
|
|
with db.cursor() as cursor:
|
||
|
|
cursor.execute('UPDATE bmc_workflow_runtime SET enabled=TRUE WHERE id=1')
|
||
|
|
db.autocommit=False
|
||
|
|
try:
|
||
|
|
with db.cursor() as cursor:
|
||
|
|
for folder,status,count in [('Sent Items','sent',0),('INBOX','new',1)]:
|
||
|
|
cursor.execute('INSERT INTO email_messages(message_id,subject,folder,status) VALUES(%s,%s,%s,%s) RETURNING id',(str(uuid.uuid4())+'@workflow.invalid','Workflow source test',folder,status))
|
||
|
|
email_id=cursor.fetchone()[0]
|
||
|
|
cursor.execute("SELECT COUNT(*) FROM bmc_workflow_events WHERE entity_type='mail' AND entity_id=%s AND trigger='mail.received'",(str(email_id),))
|
||
|
|
assert cursor.fetchone()[0]==count
|
||
|
|
finally:
|
||
|
|
db.rollback()
|
||
|
|
db.autocommit=True
|