junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Trigger MagicPlan on outcome "surveyed" transition 🟥

Browse source
This commit is contained in:
Daniel Roth 2026-05-11 09:25:41 +00:00
parent 6fd95d20af
commit 2049553176
2 changed files with 27 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
etl/hubspot/hubspot_deal_differ.py
View file

@ -198,12 +198,7 @@ class HubspotDealDiffer:
def check_for_magicplan_trigger(
new_deal: Dict[str, str], old_deal: HubspotDealData
) -> bool:
new_status = (new_deal.get("coordination_status__stage_1_") or "").lower()
old_status = (old_deal.coordination_status or "").lower()
return (
new_status in HubspotDealDiffer.COORDINATION_COMPLETE
and old_status not in HubspotDealDiffer.COORDINATION_COMPLETE
)
raise NotImplementedError
@staticmethod
def _lodgement_completed(

49
etl/hubspot/tests/test_hubspot_deal_differ.py
View file

@ -275,15 +275,12 @@ def test_pashub_trigger__coordination_design_lodgement_not_completed_and_pashub_
# ==========================
def test_magicplan_trigger__transitions_to_coordination_complete__returns_true() -> None:
def test_magicplan_trigger__outcome_transitions_to_surveyed__returns_true() -> None:
deal_id = uuid.uuid4()
# Arrange
old_deal = make_old_deal(id=deal_id, coordination_status="in progress")
new_deal = make_new_deal(
deal_id,
**{"coordination_status__stage_1_": "(v1) ioe/mtp complete"},
)
old_deal = make_old_deal(id=deal_id, outcome="assessed")
new_deal = make_new_deal(deal_id, outcome="surveyed")
# Act
result = HubspotDealDiffer.check_for_magicplan_trigger(
@ -295,20 +292,12 @@ def test_magicplan_trigger__transitions_to_coordination_complete__returns_true()
assert result is True
def test_magicplan_trigger__already_in_coordination_complete_unrelated_change__returns_false() -> None:
def test_magicplan_trigger__outcome_already_surveyed__returns_false() -> None:
deal_id = uuid.uuid4()
# Arrange
old_deal = make_old_deal(
id=deal_id,
coordination_status="(v1) ioe/mtp complete",
outcome="pending",
)
new_deal = make_new_deal(
deal_id,
**{"coordination_status__stage_1_": "(v1) ioe/mtp complete"},
outcome="won",
)
old_deal = make_old_deal(id=deal_id, outcome="surveyed")
new_deal = make_new_deal(deal_id, outcome="surveyed")
# Act
result = HubspotDealDiffer.check_for_magicplan_trigger(
@ -320,15 +309,12 @@ def test_magicplan_trigger__already_in_coordination_complete_unrelated_change__r
assert result is False
def test_magicplan_trigger__transitions_to_non_complete_coordination_status__returns_false() -> None:
def test_magicplan_trigger__outcome_transitions_to_non_surveyed__returns_false() -> None:
deal_id = uuid.uuid4()
# Arrange
old_deal = make_old_deal(id=deal_id, coordination_status="in progress")
new_deal = make_new_deal(
deal_id,
**{"coordination_status__stage_1_": "design submitted"},
)
old_deal = make_old_deal(id=deal_id, outcome="assessed")
new_deal = make_new_deal(deal_id, outcome="assessed")
# Act
result = HubspotDealDiffer.check_for_magicplan_trigger(
@ -340,6 +326,23 @@ def test_magicplan_trigger__transitions_to_non_complete_coordination_status__ret
assert result is False
def test_magicplan_trigger__outcome_surveyed_uppercase__returns_true() -> None:
deal_id = uuid.uuid4()
# Arrange
old_deal = make_old_deal(id=deal_id, outcome="assessed")
new_deal = make_new_deal(deal_id, outcome="SURVEYED")
# Act
result = HubspotDealDiffer.check_for_magicplan_trigger(
new_deal=new_deal,
old_deal=old_deal,
)
# Assert
assert result is True
# =======================
# DB UPDATE TRIGGER TESTS
# =======================