135 lines
5.3 KiB
Python
135 lines
5.3 KiB
Python
"""Saved resource coverage acts as an automatic billing rule for time only."""
|
|
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from app.workflows.engine import evaluate
|
|
|
|
|
|
def agreement(coverage=None, resource_id=41):
|
|
return {'id': resource_id, 'workflow_coverage': coverage if coverage is not None else {'tag_ids': [7]}}
|
|
|
|
|
|
def card(coverage=None, resource_id=51, balance=10):
|
|
return {'id': resource_id, 'remaining_hours': balance,
|
|
'workflow_coverage': coverage if coverage is not None else {'tag_ids': [7]}}
|
|
|
|
|
|
def context(**values):
|
|
return dict({'customer_id': '24', 'hours': 0.5, 'tag_ids': [7],
|
|
'concrete_tag_ids': [7], 'agreements': [agreement()]}, **values)
|
|
|
|
|
|
def workflow(action, *, stop=False):
|
|
return {'id': 2, 'definition': {'trigger': 'time.created',
|
|
'stop_after_match': stop,
|
|
'branches': [{'condition': None, 'actions': [action]}]}}
|
|
|
|
|
|
@pytest.mark.parametrize('trigger', ['time.created', 'time.updated'])
|
|
def test_saved_aba_coverage_selects_fixed_price_agreement(trigger):
|
|
result = evaluate([], context(), trigger)
|
|
assert result['status'] == 'decided'
|
|
assert result['method'] == 'subscription'
|
|
assert result['decision']['agreement_id'] == 41
|
|
|
|
|
|
def test_matching_card_selects_prepaid():
|
|
result = evaluate([], context(agreements=[], cards=[card()]), 'time.created')
|
|
assert result['method'] == 'prepaid'
|
|
assert result['decision']['card_id'] == 51
|
|
|
|
|
|
@pytest.mark.parametrize('coverage', [
|
|
{'tag_ids': [8]},
|
|
{'tag_ids': [7], 'excluded_tag_ids': [7]},
|
|
{'tag_ids': [7], 'auto_billing': False},
|
|
{},
|
|
])
|
|
def test_mismatch_exclusion_disabled_and_untouched_coverage_do_not_select(coverage):
|
|
result = evaluate([], context(agreements=[agreement(coverage)]), 'time.created')
|
|
assert result['status'] == 'manual_review'
|
|
assert result['method'] is None
|
|
|
|
|
|
@pytest.mark.parametrize('include_descendants,expected', [(False, None), (True, 'subscription')])
|
|
def test_descendant_coverage_requires_opt_in(include_descendants, expected):
|
|
resource = agreement({'tag_ids': [7], 'include_descendants': include_descendants})
|
|
result = evaluate([], context(agreements=[resource], tag_ids=[7, 8], concrete_tag_ids=[8]), 'time.created')
|
|
assert result['method'] == expected
|
|
|
|
|
|
@pytest.mark.parametrize('resources', [
|
|
{'agreements': [agreement(resource_id=41), agreement(resource_id=42)]},
|
|
{'agreements': [agreement()], 'cards': [card()]},
|
|
{'agreements': [], 'cards': [card(resource_id=51), card(resource_id=52)]},
|
|
])
|
|
def test_multiple_matching_resources_require_review(resources):
|
|
result = evaluate([], context(**resources), 'time.created')
|
|
assert result['status'] == 'manual_review'
|
|
assert result['reason'] == 'ambiguous_resource_coverage'
|
|
|
|
|
|
def test_explicit_billing_rule_precedes_automatic_coverage():
|
|
rule = workflow({'type': 'set_billing', 'method': 'invoice'})
|
|
result = evaluate([rule], context(), 'time.created')
|
|
assert result['method'] == 'invoice'
|
|
|
|
|
|
def test_stop_rule_without_billing_blocks_automatic_coverage():
|
|
rule = workflow({'type': 'add_tag', 'tag_id': 9}, stop=True)
|
|
result = evaluate([rule], context(), 'time.created')
|
|
assert result['status'] == 'manual_review'
|
|
assert result['method'] is None
|
|
assert result['actions'][0]['tag_id'] == 9
|
|
|
|
|
|
def test_targeted_workflow_does_not_use_unrelated_automatic_coverage():
|
|
result = evaluate([], context(target_workflow_id=99), 'time.created')
|
|
assert result['method'] is None
|
|
|
|
|
|
@pytest.mark.parametrize('trigger', ['case.created', 'case.updated', 'tag.approved'])
|
|
def test_non_time_events_do_not_choose_billing(trigger):
|
|
assert evaluate([], context(), trigger)['method'] is None
|
|
|
|
|
|
def test_saved_empty_coverage_explicitly_enables_all_tags():
|
|
result = evaluate([], context(agreements=[agreement({'auto_billing': True, 'tag_ids': []})]), 'time.created')
|
|
assert result['method'] == 'subscription'
|
|
|
|
|
|
def test_exclusions_only_saved_coverage_is_an_automatic_rule():
|
|
result = evaluate([], context(agreements=[agreement({'excluded_tag_ids': [8]})]), 'time.created')
|
|
assert result['method'] == 'subscription'
|
|
|
|
|
|
def test_pending_tag_suggestions_block_automatic_billing():
|
|
result = evaluate([], context(pending_billing_tag_suggestions=True), 'time.created')
|
|
assert result['reason'] == 'pending_tag_approval'
|
|
|
|
|
|
def test_automatic_card_keeps_existing_balance_check():
|
|
result = evaluate([], context(agreements=[], cards=[card(balance=0.25)]), 'time.created')
|
|
assert result['reason'] == 'insufficient_card_balance'
|
|
|
|
|
|
def test_automatic_card_keeps_rounding_balance_check():
|
|
resource = dict(card(balance=0.2), rounding_minutes=15)
|
|
result = evaluate([], context(hours=0.1, agreements=[], cards=[resource]), 'time.created')
|
|
assert result['reason'] == 'insufficient_card_balance'
|
|
|
|
|
|
def test_automatic_evaluation_does_not_mutate_resources_or_context():
|
|
supplied = context()
|
|
before = deepcopy(supplied)
|
|
evaluate([], supplied, 'time.created')
|
|
assert supplied == before
|
|
|
|
@pytest.mark.parametrize('trigger', ['tags.added', 'tags.removed', 'tags.changed'])
|
|
def test_time_tag_events_retain_automatic_coverage(trigger):
|
|
context = {'entity_type': 'time_entry', 'tag_ids': [7], 'agreements': [{'id': 2, 'workflow_coverage': {'tag_ids': [7]}}]}
|
|
assert evaluate([], context, trigger)['method'] == 'subscription'
|
|
context['tag_ids'] = []
|
|
assert evaluate([], context, trigger)['status'] == 'manual_review'
|