apm-wo-analysis/tests/test_slack_post_handler.py

"""Tests for the slack-post Lambda handler and slackio I/O helpers.

Handler is loaded via importlib to avoid the handler.py name collision between
the classifier and slack-post Lambdas (both are on the pythonpath and both are
named handler.py — bare ``import handler`` would be ambiguous/cached).

Monkeypatching strategy
-----------------------
The handler module shares the same ``slackio`` object that was imported during
module load (``sp_handler.slackio is slackio`` is True), so patching attributes
on the ``slackio`` module is sufficient.

Real ``blockkit`` is used (it is pure — no network, no AWS).

slackio cache isolation
-----------------------
``_creds`` and ``_dashboard_url`` are module-level globals that persist across
tests within the same process.  Each slackio test resets them before running to
guarantee isolation.
"""

from __future__ import annotations

import importlib.util
import json
import sys
from pathlib import Path
# ---------------------------------------------------------------------------
# Shared pythonpath: slack_post dir must be on sys.path for slackio + blockkit.
# conftest.py sets the env vars; we rely on that here.
# ---------------------------------------------------------------------------

_SLACK_POST_DIR = Path(__file__).resolve().parents[1] / "lambdas" / "slack_post"
if str(_SLACK_POST_DIR) not in sys.path:
    sys.path.insert(0, str(_SLACK_POST_DIR))

import slackio  # noqa: E402

# ---------------------------------------------------------------------------
# Load the slack-post handler via importlib (avoids name collision with
# lambdas/classifier/handler.py which is also on pythonpath).
# ---------------------------------------------------------------------------

_HANDLER_PATH = _SLACK_POST_DIR / "handler.py"
_spec = importlib.util.spec_from_file_location("slack_post_handler", _HANDLER_PATH)
sp_handler = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(sp_handler)


# ---------------------------------------------------------------------------
# Fake Slack WebClient
# ---------------------------------------------------------------------------


class _FakeClient:
    """Minimal WebClient stand-in that records chat_postMessage calls."""

    def __init__(self):
        self.calls: list[dict] = []

    def chat_postMessage(self, **kwargs):
        self.calls.append(kwargs)
        return {"ok": True}


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

SUMMARY_ZERO_THIRDS = {
    "dt": "2026-05-28",
    "classified_total": 100,
    "blank_comment_rows": 5,
    "escalation_total": 3,
    "action_needed": 10,
    "routine": 90,
    "third_escalation_count": 0,
    "category_counts": {"Schedule Confirmed": 50, "3rd Escalation": 0},
    "top_sites": [],
    "mismatches": [],
    "generated_at": "2026-05-28T12:00:00Z",
}

SUMMARY_WITH_THIRDS = {
    **SUMMARY_ZERO_THIRDS,
    "third_escalation_count": 2,
    "category_counts": {"3rd Escalation": 2, "Schedule Confirmed": 48},
}

DETAILS_WITH_THIRDS = [
    {
        "wo_number": "WO-001",
        "category": "3rd Escalation",
        "site": "ABQ5",
        "wo_description": "Fix HVAC unit",
    },
    {
        "wo_number": "WO-002",
        "category": "3rd Escalation",
        "site": "ACY9",
        "wo_description": "Repair dock door",
    },
    {
        "wo_number": "WO-003",
        "category": "Schedule Confirmed",  # should NOT appear in the alert
        "site": "ABQ5",
        "wo_description": "Routine PM",
    },
]


# ---------------------------------------------------------------------------
# Section 1: handler._yesterday
# ---------------------------------------------------------------------------


class TestYesterday:
    def test_yesterday_basic(self):
        assert sp_handler._yesterday("2026-05-28") == "2026-05-27"

    def test_yesterday_month_boundary(self):
        assert sp_handler._yesterday("2026-06-01") == "2026-05-31"

    def test_yesterday_year_boundary(self):
        assert sp_handler._yesterday("2026-01-01") == "2025-12-31"


# ---------------------------------------------------------------------------
# Section 2: handler.handler — the main Lambda entry point
# ---------------------------------------------------------------------------


class TestHandler:
    def _make_fake_client(self):
        return _FakeClient()

    def test_no_summary_returns_not_posted_and_does_not_call_slack(self, monkeypatch):
        """When read_meta_json returns None for summary.json, handler returns
        posted=False and must NOT call chat_postMessage."""
        fake_client = self._make_fake_client()

        monkeypatch.setattr(slackio, "read_meta_json", lambda dt, name: None)
        monkeypatch.setattr(slackio, "web_client", lambda: fake_client)
        monkeypatch.setattr(slackio, "channel_id", lambda: "C12345")
        monkeypatch.setattr(
            slackio, "get_dashboard_url", lambda: "https://grafana.example.com"
        )

        result = sp_handler.handler({"dt": "2026-05-28"}, None)

        assert result["posted"] is False
        assert result["dt"] == "2026-05-28"
        assert fake_client.calls == [], "chat_postMessage must not be called"

    def test_summary_zero_thirds_posts_summary_only(self, monkeypatch):
        """When third_escalation_count == 0, exactly one chat_postMessage is made
        (daily summary); the standalone alert is suppressed."""
        fake_client = self._make_fake_client()

        def fake_read(dt, name):
            if name == "summary.json":
                return SUMMARY_ZERO_THIRDS
            return None

        monkeypatch.setattr(slackio, "read_meta_json", fake_read)
        monkeypatch.setattr(slackio, "web_client", lambda: fake_client)
        monkeypatch.setattr(slackio, "channel_id", lambda: "C12345")
        monkeypatch.setattr(
            slackio, "get_dashboard_url", lambda: "https://grafana.example.com"
        )

        result = sp_handler.handler({"dt": "2026-05-28"}, None)

        assert result["posted"] is True
        assert result["summary"] is True
        assert result["alert"] is False
        assert len(fake_client.calls) == 1, "exactly one postMessage (summary)"
        # The message text should reference today's dt.
        assert "2026-05-28" in fake_client.calls[0]["text"]

    def test_summary_with_thirds_posts_summary_and_alert(self, monkeypatch):
        """When third_escalation_count > 0, two chat_postMessage calls are made:
        one for the daily summary and one for the 3rd-escalation alert."""
        fake_client = self._make_fake_client()

        def fake_read(dt, name):
            if name == "summary.json":
                return SUMMARY_WITH_THIRDS
            if name == "details.json":
                return DETAILS_WITH_THIRDS
            return None

        monkeypatch.setattr(slackio, "read_meta_json", fake_read)
        monkeypatch.setattr(slackio, "web_client", lambda: fake_client)
        monkeypatch.setattr(slackio, "channel_id", lambda: "C12345")
        monkeypatch.setattr(
            slackio, "get_dashboard_url", lambda: "https://grafana.example.com"
        )

        result = sp_handler.handler({"dt": "2026-05-28"}, None)

        assert result["posted"] is True
        assert result["summary"] is True
        assert result["alert"] is True
        assert len(fake_client.calls) == 2, "summary + alert = two postMessages"

    def test_alert_filtered_to_3rd_escalation_only(self, monkeypatch):
        """The details.json rows fed to build_escalation_alert must be pre-filtered
        to category == '3rd Escalation'.  WO-003 (Schedule Confirmed) must not
        appear in the alert message."""
        fake_client = self._make_fake_client()

        def fake_read(dt, name):
            if name == "summary.json":
                return SUMMARY_WITH_THIRDS
            if name == "details.json":
                return DETAILS_WITH_THIRDS
            return None

        monkeypatch.setattr(slackio, "read_meta_json", fake_read)
        monkeypatch.setattr(slackio, "web_client", lambda: fake_client)
        monkeypatch.setattr(slackio, "channel_id", lambda: "C12345")
        monkeypatch.setattr(
            slackio, "get_dashboard_url", lambda: "https://grafana.example.com"
        )

        sp_handler.handler({"dt": "2026-05-28"}, None)

        # The alert is the second postMessage call.
        assert len(fake_client.calls) == 2
        alert_call = fake_client.calls[1]
        alert_blocks = alert_call["blocks"]

        # Serialise blocks to text for easy searching.
        alert_text = json.dumps(alert_blocks)

        # Both 3rd-escalation WOs must be present.
        assert "WO-001" in alert_text
        assert "WO-002" in alert_text
        # The non-3rd WO must NOT appear in the alert.
        assert "WO-003" not in alert_text


# ---------------------------------------------------------------------------
# Section 3: slackio unit tests
# ---------------------------------------------------------------------------


class TestSlackioReadMetaJson:
    def test_returns_none_on_no_such_key(self, monkeypatch):
        """read_meta_json must return None (not raise) when S3 returns NoSuchKey."""
        # Build a fake get_object that raises the real NoSuchKey exception.
        NoSuchKey = slackio._s3.exceptions.NoSuchKey

        def fake_get_object(Bucket, Key):
            raise NoSuchKey(
                {"Error": {"Code": "NoSuchKey", "Message": "not found"}},
                "GetObject",
            )

        monkeypatch.setattr(slackio._s3, "get_object", fake_get_object)

        result = slackio.read_meta_json("2026-05-28", "summary.json")
        assert result is None

    def test_returns_parsed_json_on_success(self, monkeypatch):
        """read_meta_json must parse and return the JSON body on a successful get."""
        payload = {"classified_total": 42}

        class _FakeBody:
            def read(self):
                return json.dumps(payload).encode("utf-8")

        monkeypatch.setattr(
            slackio._s3, "get_object", lambda Bucket, Key: {"Body": _FakeBody()}
        )

        result = slackio.read_meta_json("2026-05-28", "summary.json")
        assert result == payload


class TestSlackioCredentialsCaching:
    """get_credentials() and get_dashboard_url() must cache their results and
    only call the underlying boto3 client once per container lifetime."""

    def setup_method(self):
        # Reset module-level caches so each test starts from a cold state.
        slackio._creds = None
        slackio._dashboard_url = None

    def test_get_credentials_calls_secrets_once(self, monkeypatch):
        """Calling get_credentials() twice must invoke get_secret_value exactly
        once (the result is cached after the first call)."""
        creds_payload = {
            "botToken": "xoxb-test",
            "signingSecret": "abc",
            "channelId": "C99",
        }
        call_count = 0

        def fake_get_secret(SecretId):
            nonlocal call_count
            call_count += 1
            return {"SecretString": json.dumps(creds_payload)}

        monkeypatch.setattr(slackio._secrets, "get_secret_value", fake_get_secret)

        first = slackio.get_credentials()
        second = slackio.get_credentials()

        assert call_count == 1, "get_secret_value should be called only once"
        assert first == creds_payload
        assert second is first  # same object from cache

    def test_get_dashboard_url_calls_ssm_once(self, monkeypatch):
        """Calling get_dashboard_url() twice must invoke get_parameter exactly
        once (the URL is cached after the first call)."""
        expected_url = "https://grafana.example.com/d/abc"
        call_count = 0

        def fake_get_parameter(Name):
            nonlocal call_count
            call_count += 1
            return {"Parameter": {"Value": expected_url}}

        monkeypatch.setattr(slackio._ssm, "get_parameter", fake_get_parameter)

        first = slackio.get_dashboard_url()
        second = slackio.get_dashboard_url()

        assert call_count == 1, "get_parameter should be called only once"
        assert first == expected_url
        assert second == expected_url