openjobradar/tests/test_notify_delivery.py at dependabot/github_actions/actions/setup-python-7.0.0 · ghzhost/openjobradar · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
"""Alert rendering and SES delivery (roadmap M4/M7, ADR-0028)."""
from __future__ import annotations

import boto3
import pytest
from moto import mock_aws

from openjobradar.notify.decision import NotifyDecision
from openjobradar.notify.render import render_alert_body, render_alert_subject
from openjobradar.notify.ses_sender import deliver
from openjobradar.poll.models import Posting
from openjobradar.scoring.protocol import ScoreResult

POSTING = Posting(
    org_slug="gitlab",
    ats_type="greenhouse",
    job_id="123",
    title="Staff Data Engineer",
    location_raw="Remote",
    apply_url="https://job-boards.greenhouse.io/gitlab/jobs/123",
    jd_text="Build data pipelines. " * 50,
    org_name="GitLab",
)
SCORE = ScoreResult(
    score=82, verdict="alert", facets={"craft_depth": 90}, rationale="Strong overlap.",
    rubric_id="data_analytics_ml", scorer_id="stub-v0",
)


def _decision(**overrides) -> NotifyDecision:
    defaults = {
        "verdict": "alert",
        "channels": ("email",),
        "is_dry_run": False,
        "deferred_until_hour_utc": None,
        "reason": "score meets alert threshold",
    }
    return NotifyDecision(**{**defaults, **overrides})


# --- rendering ---------------------------------------------------------------


def test_render_subject_includes_title_org_and_score() -> None:
    subject = render_alert_subject(POSTING, SCORE)
    assert "Staff Data Engineer" in subject
    assert "GitLab" in subject
    assert "82" in subject


def test_render_body_includes_apply_url_and_rationale() -> None:
    body = render_alert_body(POSTING, SCORE)
    assert POSTING.apply_url in body
    assert SCORE.rationale in body
    assert "82/100" in body


def test_render_body_truncates_long_jd_text() -> None:
    body = render_alert_body(POSTING, SCORE)
    assert "…" in body
    assert len(body) < len(POSTING.jd_text)


def test_render_body_handles_missing_jd_text() -> None:
    bare = Posting(
        org_slug="acme", ats_type="lever", job_id="1", title="Engineer",
        location_raw="", apply_url="https://jobs.lever.co/acme/1",
    )
    body = render_alert_body(bare, SCORE)
    assert "not specified" in body  # location fallback
    assert "excerpt" not in body.lower()  # no jd section when there's nothing to show


# --- delivery ------------------------------------------------------------------


@pytest.fixture
def ses_client():
    with mock_aws():
        client = boto3.client("ses", region_name="us-east-1")
        client.verify_email_identity(EmailAddress="alerts@example.com")
        yield client


def test_alert_verdict_actually_sends(ses_client) -> None:
    result = deliver(
        ses_client, _decision(), POSTING, SCORE,
        from_email="alerts@example.com", to_email="user@example.com",
    )
    assert result.sent is True
    assert result.reason == "sent"
    assert result.message_id
    assert result.subject and result.body


def test_suppressed_verdict_never_sends() -> None:
    result = deliver(
        object(), _decision(verdict="suppressed", channels=()), POSTING, SCORE,  # type: ignore[arg-type]
        from_email="alerts@example.com", to_email="user@example.com",
    )
    assert result.sent is False
    assert "suppressed" in result.reason
    assert result.subject and result.body  # still rendered, for a future preview UI


def test_digest_verdict_never_sends_as_an_immediate_alert() -> None:
    result = deliver(
        object(), _decision(verdict="digest"), POSTING, SCORE,  # type: ignore[arg-type]
        from_email="alerts@example.com", to_email="user@example.com",
    )
    assert result.sent is False
    assert "digest" in result.reason


def test_dry_run_alert_renders_but_never_sends() -> None:
    """Never-silent-death (ADR-0017/0019): a dry-run alert still gets real rendered content --
    it just never reaches SES."""
    result = deliver(
        object(), _decision(is_dry_run=True), POSTING, SCORE,  # type: ignore[arg-type]
        from_email="alerts@example.com", to_email="user@example.com",
    )
    assert result.sent is False
    assert result.reason == "dry_run"
    assert POSTING.title in result.subject
    assert result.message_id == ""


def test_deferred_quiet_hours_alert_never_sends_yet() -> None:
    result = deliver(
        object(), _decision(deferred_until_hour_utc=6), POSTING, SCORE,  # type: ignore[arg-type]
        from_email="alerts@example.com", to_email="user@example.com",
    )
    assert result.sent is False
    assert result.reason == "deferred_quiet_hours"


def test_email_channel_not_enabled_never_sends() -> None:
    result = deliver(
        object(), _decision(channels=("webhook",)), POSTING, SCORE,  # type: ignore[arg-type]
        from_email="alerts@example.com", to_email="user@example.com",
    )
    assert result.sent is False
    assert "email" in result.reason