procurement-ingest/lambdas/wo/email_processor/tests/test_wo_merge.py

"""Moto-backed merge-semantics tests for save_work_order (constraint 5d).

A real in-memory work-orders table is stood up with moto so these exercise the
actual DynamoDB UpdateExpression build -- the reserved-word status->wo_status
mapping, the if_not_exists created_at guard, and the None-field-dropping logic --
against a real store rather than a stub. Mirrors tests/test_po_merge.py's
approach, placed beside the WO code (same constraint-3 principle that moved
test_po_merge out of the repo-root tests/).

The table name comes from ``wo_persistence.WORK_ORDERS_TABLE`` (default the
literal ``"WorkOrders"``, NOT a kebab-case name) so the fixture and the code
under test resolve the same name from one source. persistence.py is untouched --
this is a pure new-test addition pinning four existing-but-untested behaviors.
"""

import boto3
import pytest
from moto import mock_aws


@pytest.fixture
def wo_table(wo_persistence):
    with mock_aws():
        resource = boto3.resource("dynamodb", region_name="us-east-1")
        table = resource.create_table(
            TableName=wo_persistence.WORK_ORDERS_TABLE,
            KeySchema=[{"AttributeName": "work_order_id", "KeyType": "HASH"}],
            AttributeDefinitions=[
                {"AttributeName": "work_order_id", "AttributeType": "S"},
            ],
            BillingMode="PAY_PER_REQUEST",
        )
        table.wait_until_exists()
        # Point the lazily-cached module resource at the moto-mocked one.
        wo_persistence.dynamodb = resource
        try:
            yield table
        finally:
            wo_persistence.dynamodb = None


def _item(table, work_order_id):
    return table.get_item(Key={"work_order_id": work_order_id}).get("Item")


def _parsed(work_order_id, **overrides):
    base = {
        "work_order_id": work_order_id,
        "email_type": "update",
        "description": None,
        "status": None,
        "site_code": None,
        "building": None,
        "address": None,
        "severity": None,
        "priority": None,
        "date_reported": None,
        "scheduled_start": None,
        "due_date": None,
        "assigned_to": None,
    }
    base.update(overrides)
    return base


def test_null_status_never_clobbers_wo_status(wo_persistence, wo_table):
    """A later email with status=None must NOT overwrite an established
    wo_status (None fields are dropped from the SET clause)."""
    wo_table.put_item(Item={"work_order_id": "WO-A", "wo_status": "assigned"})

    wo_persistence.save_work_order(_parsed("WO-A", status=None), "s3://b/k")

    item = _item(wo_table, "WO-A")
    assert item["wo_status"] == "assigned"  # not clobbered


def test_created_at_immutable_via_if_not_exists(wo_persistence, wo_table):
    """created_at is set with if_not_exists, so the first write's value survives
    every subsequent upsert."""
    wo_persistence.save_work_order(_parsed("WO-B"), "s3://b/k1")
    first = _item(wo_table, "WO-B")["created_at"]

    wo_persistence.save_work_order(_parsed("WO-B", description="later"), "s3://b/k2")
    second = _item(wo_table, "WO-B")

    assert second["created_at"] == first  # immutable
    assert second["description"] == "later"  # other fields still enriched
    assert second["updated_at"] >= first  # updated_at is always refreshed


def test_status_maps_to_wo_status_reserved_word(wo_persistence, wo_table):
    """'status' is a DynamoDB reserved word -- the source field maps to the
    'wo_status' attribute, and no attribute literally named 'status' is written."""
    wo_persistence.save_work_order(_parsed("WO-C", status="completed"), "s3://b/k")

    item = _item(wo_table, "WO-C")
    assert item["wo_status"] == "completed"
    assert "status" not in item


def test_none_fields_absent_from_set_clause(wo_persistence, wo_table):
    """None source fields are dropped from the SET clause: a pre-existing
    attribute is retained rather than being overwritten with None."""
    wo_table.put_item(
        Item={"work_order_id": "WO-D", "building": "B12", "priority": "high"}
    )

    wo_persistence.save_work_order(_parsed("WO-D", description="fix door"), "s3://b/k")

    item = _item(wo_table, "WO-D")
    assert item["description"] == "fix door"  # the one non-None field written
    assert item["building"] == "B12"  # pre-existing attrs retained
    assert item["priority"] == "high"


def test_record_type_only_when_email_type_present(wo_persistence, wo_table):
    """record_type is written only when email_type is present (truthy); a falsy
    email_type leaves the attribute absent."""
    wo_persistence.save_work_order(_parsed("WO-E", email_type="new_work_order"), "s")
    assert _item(wo_table, "WO-E")["record_type"] == "new_work_order"

    wo_persistence.save_work_order(_parsed("WO-F", email_type=""), "s")
    assert "record_type" not in _item(wo_table, "WO-F")