apm-wo-analysis/tests/test_blockkit.py

"""Offline unit tests for the Block Kit surface builders.

No AWS, no network, no slack_sdk. All fixtures are plain dicts.

Run with the repo venv:

    ./.venv/bin/python -m pytest tests/test_blockkit.py -q
"""

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(
    0, str(Path(__file__).resolve().parent.parent / "lambdas" / "slack_post")
)

import blockkit  # noqa: E402
from blockkit import (  # noqa: E402
    build_daily_summary,
    build_escalation_alert,
    build_wo_modal,
)

# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------

DASHBOARD_URL = "https://grafana.seahaven.internal/d/apm-wo"

CATEGORY_COUNTS_TYPICAL: dict[str, int] = {
    "3rd Escalation": 4,
    "2nd Escalation": 8,
    "1st Escalation": 12,
    "SIM Ticket": 2,
    "Vendor No-Show": 3,
    "Awaiting Scheduling": 25,
    "Report / Docs Needed": 18,
    "Awaiting Report / Invoice": 7,
    "Awaiting Vendor / Parts": 5,
    "Status Inquiry": 6,
    "Other Escalation": 1,
    "Schedule Confirmed": 90,
    "Weekly WO Scheduled": 40,
    "Completed / Pending Close": 22,
    "Acknowledgement / No-op": 10,
    "Cancelled": 5,
    "On Hold": 3,
    "Rescheduled": 4,
    "Avetta Project Created": 2,
    "Other": 15,
}

TODAY_SUMMARY: dict = {
    "dt": "2026-05-28",
    "classified_total": 282,
    "blank_comment_rows": 68,
    "category_counts": CATEGORY_COUNTS_TYPICAL,
    "escalation_total": 27,
    "third_escalation_count": 4,
    "action_needed": 91,
    "routine": 191,
    "top_sites": [
        {"site": "ABQ5", "count": 45},
        {"site": "ACY9", "count": 38},
        {"site": "BOS1", "count": 31},
        {"site": "DFW7", "count": 28},
        {"site": "LAX9", "count": 22},
    ],
    "mismatches": [
        {
            "wo_number": "WO-001",
            "category": "Completed / Pending Close",
            "mismatch": "comment claims completion while on REPORT hold",
        },
        {
            "wo_number": "WO-002",
            "category": "Schedule Confirmed",
            "mismatch": "comment confirms schedule while on SCHEDULING hold",
        },
    ],
    "generated_at": "2026-05-28T06:00:00Z",
}

YESTERDAY_SUMMARY: dict = {
    "dt": "2026-05-27",
    "classified_total": 270,
    "blank_comment_rows": 80,
    "category_counts": {},
    "escalation_total": 30,
    "third_escalation_count": 6,
    "action_needed": 95,
    "routine": 175,
    "top_sites": [],
    "mismatches": [],
    "generated_at": "2026-05-27T06:00:00Z",
}


def _make_wo(i: int) -> dict:
    return {
        "wo_number": f"WO-{i:04d}",
        "wo_description": f"Repair HVAC unit at dock {i}",
        "site": "ABQ5",
        "department": "SSP",
        "category": "3rd Escalation",
        "last_comment": f"3rd attempt to schedule vendor for dock {i} repair.",
        "is_escalation": True,
        "is_action": True,
        "mismatch": None,
    }


# ---------------------------------------------------------------------------
# build_daily_summary
# ---------------------------------------------------------------------------


class TestDailySummaryDeltas:
    def test_with_yesterday_shows_deltas(self):
        blocks = build_daily_summary(TODAY_SUMMARY, YESTERDAY_SUMMARY, DASHBOARD_URL)
        # Collect all text from section/context blocks.
        all_text = _extract_text(blocks)
        # classified went 270 -> 282 (+12), escalation 30 -> 27 (-3).
        assert "▲+12" in all_text, "expected classified delta ▲+12"
        assert "▼-3" in all_text, "expected escalation delta ▼-3"

    def test_without_yesterday_omits_deltas(self):
        """yesterday=None must not raise and must omit delta symbols."""
        blocks = build_daily_summary(TODAY_SUMMARY, None, DASHBOARD_URL)
        all_text = _extract_text(blocks)
        assert "▲" not in all_text
        assert "▼" not in all_text
        assert "No prior day" in all_text

    def test_without_yesterday_returns_blocks(self):
        blocks = build_daily_summary(TODAY_SUMMARY, None, DASHBOARD_URL)
        assert isinstance(blocks, list)
        assert len(blocks) > 0

    def test_zero_delta_shows_tilde(self):
        yesterday_same = {**YESTERDAY_SUMMARY, "escalation_total": 27}
        blocks = build_daily_summary(TODAY_SUMMARY, yesterday_same, DASHBOARD_URL)
        all_text = _extract_text(blocks)
        assert "~0" in all_text


class TestDailySummaryStructure:
    def test_has_header_block(self):
        blocks = build_daily_summary(TODAY_SUMMARY, YESTERDAY_SUMMARY, DASHBOARD_URL)
        assert any(b.get("type") == "header" for b in blocks)

    def test_header_contains_date(self):
        blocks = build_daily_summary(TODAY_SUMMARY, YESTERDAY_SUMMARY, DASHBOARD_URL)
        header = next(b for b in blocks if b.get("type") == "header")
        assert "2026" in header["text"]["text"] or "May" in header["text"]["text"]

    def test_mismatch_callout_present_when_mismatches_nonzero(self):
        blocks = build_daily_summary(TODAY_SUMMARY, YESTERDAY_SUMMARY, DASHBOARD_URL)
        all_text = _extract_text(blocks)
        assert "mismatch" in all_text.lower()

    def test_mismatch_callout_absent_when_no_mismatches(self):
        today_no_mm = {**TODAY_SUMMARY, "mismatches": []}
        blocks = build_daily_summary(today_no_mm, YESTERDAY_SUMMARY, DASHBOARD_URL)
        all_text = _extract_text(blocks)
        assert "mismatch" not in all_text.lower()

    def test_dashboard_link_button_url(self):
        blocks = build_daily_summary(TODAY_SUMMARY, YESTERDAY_SUMMARY, DASHBOARD_URL)
        url = _find_link_button_url(blocks)
        assert url == DASHBOARD_URL, f"expected {DASHBOARD_URL!r}, got {url!r}"

    def test_dashboard_button_has_no_action_id(self):
        """The Open-dashboard button must be a url button, not an action button."""
        blocks = build_daily_summary(TODAY_SUMMARY, YESTERDAY_SUMMARY, DASHBOARD_URL)
        for block in blocks:
            if block.get("type") != "actions":
                continue
            for elem in block.get("elements", []):
                if elem.get("url") == DASHBOARD_URL:
                    assert "action_id" not in elem, (
                        "dashboard link button must not carry an action_id"
                    )

    def test_drill_buttons_have_action_id(self):
        blocks = build_daily_summary(TODAY_SUMMARY, YESTERDAY_SUMMARY, DASHBOARD_URL)
        drill_found = False
        for block in blocks:
            if block.get("type") != "actions":
                continue
            for elem in block.get("elements", []):
                if elem.get("action_id") == "drill_category":
                    drill_found = True
                    assert "value" in elem
        assert drill_found, "expected at least one drill_category action button"

    def test_3rd_escalation_highlighted(self):
        blocks = build_daily_summary(TODAY_SUMMARY, YESTERDAY_SUMMARY, DASHBOARD_URL)
        all_text = _extract_text(blocks)
        # 3rd Escalation should appear in bold mrkdwn.
        assert "*3rd Escalation" in all_text or "3rd Escalation*" in all_text


class TestDailySummaryBlockLimit:
    def _large_summary(self) -> dict:
        """Summary with 25+ categories, 30 sites, and 200 mismatches."""
        big_cats: dict[str, int] = {f"Category {i}": i + 1 for i in range(30)}
        big_cats.update(CATEGORY_COUNTS_TYPICAL)
        big_sites = [{"site": f"S{i:03d}", "count": 30 - i} for i in range(30)]
        big_mismatches = [
            {"wo_number": f"WO-{i:04d}", "category": "X", "mismatch": "test mm"}
            for i in range(200)
        ]
        return {
            **TODAY_SUMMARY,
            "category_counts": big_cats,
            "top_sites": big_sites,
            "mismatches": big_mismatches,
        }

    def test_stays_under_100_blocks_large_input(self):
        big = self._large_summary()
        blocks = build_daily_summary(big, YESTERDAY_SUMMARY, DASHBOARD_URL)
        assert len(blocks) < 100, f"daily summary exceeded 100 blocks: {len(blocks)}"

    def test_reports_block_count(self, capsys):
        big = self._large_summary()
        blocks = build_daily_summary(big, YESTERDAY_SUMMARY, DASHBOARD_URL)
        with capsys.disabled():
            print(f"\n[test] daily summary (large input): {len(blocks)} blocks")


# ---------------------------------------------------------------------------
# build_escalation_alert
# ---------------------------------------------------------------------------


class TestEscalationAlert:
    def test_returns_none_on_empty_list(self):
        assert build_escalation_alert([]) is None

    def test_returns_blocks_on_non_empty(self):
        escalations = [_make_wo(i) for i in range(3)]
        result = build_escalation_alert(escalations)
        assert result is not None
        assert isinstance(result, list)
        assert len(result) > 0

    def test_exactly_one_here_mention(self):
        escalations = [_make_wo(i) for i in range(5)]
        blocks = build_escalation_alert(escalations)
        assert blocks is not None
        all_text = _extract_text(blocks)
        here_count = all_text.count("<!here>")
        assert here_count == 1, f"expected exactly 1 <!here>, found {here_count}"

    def test_here_in_section_not_header(self):
        """@here must be in a section block, not the header."""
        escalations = [_make_wo(i) for i in range(2)]
        blocks = build_escalation_alert(escalations)
        assert blocks is not None
        for block in blocks:
            if block.get("type") == "header":
                header_text = block["text"]["text"]
                assert "<!here>" not in header_text

    def test_no_here_on_empty(self):
        result = build_escalation_alert([])
        assert result is None

    def test_overflow_note_when_many_wos(self):
        """When list exceeds ALERT_MAX_WOS, a '+M more' context block is added."""
        many = [_make_wo(i) for i in range(blockkit.ALERT_MAX_WOS + 10)]
        blocks = build_escalation_alert(many)
        assert blocks is not None
        all_text = _extract_text(blocks)
        assert "more" in all_text.lower()

    def test_no_overflow_note_when_within_limit(self):
        few = [_make_wo(i) for i in range(blockkit.ALERT_MAX_WOS - 1)]
        blocks = build_escalation_alert(few)
        assert blocks is not None
        all_text = _extract_text(blocks)
        # Should not have a "+N more" line.
        assert " more —" not in all_text

    def test_single_wo_grammatically_correct(self):
        """'WO' (singular) when count is 1."""
        blocks = build_escalation_alert([_make_wo(1)])
        assert blocks is not None
        header = next(b for b in blocks if b.get("type") == "header")
        assert "WOs" not in header["text"]["text"], "expected singular 'WO' not 'WOs'"

    def test_stays_under_100_blocks(self):
        huge = [_make_wo(i) for i in range(500)]
        blocks = build_escalation_alert(huge)
        assert blocks is not None
        assert len(blocks) < 100, f"alert exceeded 100 blocks: {len(blocks)}"

    def test_wo_number_appears_in_blocks(self):
        escalations = [_make_wo(42)]
        blocks = build_escalation_alert(escalations)
        assert blocks is not None
        all_text = _extract_text(blocks)
        assert "WO-0042" in all_text


# ---------------------------------------------------------------------------
# build_wo_modal
# ---------------------------------------------------------------------------


class TestWoModal:
    def test_returns_dict_with_modal_type(self):
        modal = build_wo_modal("Test Modal", [_make_wo(1)], DASHBOARD_URL)
        assert isinstance(modal, dict)
        assert modal.get("type") == "modal"

    def test_modal_has_close_button(self):
        modal = build_wo_modal("Test", [_make_wo(1)], DASHBOARD_URL)
        assert "close" in modal
        assert modal["close"]["type"] == "plain_text"

    def test_title_truncated_to_24_chars(self):
        long_title = "A" * 50
        modal = build_wo_modal(long_title, [_make_wo(1)], DASHBOARD_URL)
        assert len(modal["title"]["text"]) <= 24

    def test_500_wos_truncated_to_under_100_blocks(self, capsys):
        wos = [_make_wo(i) for i in range(500)]
        modal = build_wo_modal("3rd Escalations", wos, DASHBOARD_URL)
        block_count = len(modal["blocks"])
        with capsys.disabled():
            print(f"\n[test] 500-WO modal: {block_count} blocks")
        assert block_count <= 100, (
            f"modal with 500 WOs has {block_count} blocks, exceeds 100"
        )

    def test_500_wos_has_overflow_note(self):
        wos = [_make_wo(i) for i in range(500)]
        modal = build_wo_modal("3rd Escalations", wos, DASHBOARD_URL)
        all_text = _extract_text(modal["blocks"])
        assert "more" in all_text.lower(), "expected '+M more' overflow note"

    def test_overflow_note_contains_dashboard_url(self):
        wos = [_make_wo(i) for i in range(500)]
        modal = build_wo_modal("3rd Escalations", wos, DASHBOARD_URL)
        all_text = _extract_text(modal["blocks"])
        assert DASHBOARD_URL in all_text, "overflow note must link to the dashboard URL"

    def test_overflow_count_correct(self):
        """The overflow note's M value must equal len(wos) - rendered_count."""
        n = 500
        wos = [_make_wo(i) for i in range(n)]
        modal = build_wo_modal("3rd Escalations", wos, DASHBOARD_URL)
        blocks = modal["blocks"]
        # Last block should be context with the overflow note.
        last_text = _extract_text([blocks[-1]])
        # The overflow note pattern is "+M more".
        import re

        match = re.search(r"\+(\d+) more", last_text)
        assert match, f"no '+M more' pattern in last block: {last_text!r}"
        reported_overflow = int(match.group(1))
        # Block count: total blocks = visible WO blocks + 1 overflow block.
        rendered_wo_blocks = len(blocks) - 1
        assert reported_overflow == n - rendered_wo_blocks, (
            f"overflow note says +{reported_overflow} but "
            f"{n} - {rendered_wo_blocks} = {n - rendered_wo_blocks}"
        )

    def test_empty_wo_list_returns_placeholder(self):
        modal = build_wo_modal("Empty", [], DASHBOARD_URL)
        all_text = _extract_text(modal["blocks"])
        assert "no work orders" in all_text.lower()

    def test_wo_number_appears_in_blocks(self):
        modal = build_wo_modal("Test", [_make_wo(7)], DASHBOARD_URL)
        all_text = _extract_text(modal["blocks"])
        assert "WO-0007" in all_text

    def test_snippet_truncated_to_150_chars(self):
        wo = _make_wo(1)
        wo["last_comment"] = "X" * 300
        modal = build_wo_modal("Test", [wo], DASHBOARD_URL)
        all_text = _extract_text(modal["blocks"])
        # The snippet in the text should not be 300 X's.
        assert "X" * 200 not in all_text

    def test_no_apm_deep_links(self):
        """Modal must not contain any APM application deep-links."""
        wos = [_make_wo(i) for i in range(10)]
        modal = build_wo_modal("3rd Escalations", wos, DASHBOARD_URL)
        all_text = _extract_text(modal["blocks"])
        # APM URLs would start with typical patterns — confirm none present.
        assert "apm://" not in all_text
        assert "app.apm" not in all_text


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _extract_text(blocks: list[dict]) -> str:
    """Collect all text strings from blocks for assertion convenience."""
    parts: list[str] = []
    for block in blocks:
        btype = block.get("type")
        if btype == "header":
            parts.append(block.get("text", {}).get("text", ""))
        elif btype == "section":
            text_obj = block.get("text")
            if text_obj:
                parts.append(text_obj.get("text", ""))
            for field in block.get("fields", []):
                parts.append(field.get("text", ""))
        elif btype == "context":
            for elem in block.get("elements", []):
                parts.append(elem.get("text", ""))
        elif btype == "actions":
            for elem in block.get("elements", []):
                txt = elem.get("text", {})
                if isinstance(txt, dict):
                    parts.append(txt.get("text", ""))
    return "\n".join(parts)


def _find_link_button_url(blocks: list[dict]) -> str | None:
    """Return the URL from the first link button found in any actions block."""
    for block in blocks:
        if block.get("type") != "actions":
            continue
        for elem in block.get("elements", []):
            if "url" in elem:
                return elem["url"]
    return None