Add template render tester

.gitignore

@@ -12,3 +12,8 @@ crash.log
 .vscode/
 __pycache__/
 .coverage
+
+# Python packaging artifacts
+*.egg
+*.egg-info/
+.eggs/

src/lambdacron/notifications/__init__.py

@@ -1,5 +1,6 @@
 from lambdacron.notifications.base import (
     EnvVarTemplateProvider,
+    FileTemplateProvider,
     RenderedTemplateNotificationHandler,
     TemplateProvider,
 )
@@ -9,6 +10,7 @@
 __all__ = [
     "EmailNotificationHandler",
     "EnvVarTemplateProvider",
+    "FileTemplateProvider",
     "PrintNotificationHandler",
     "RenderedTemplateNotificationHandler",
     "TemplateProvider",

src/lambdacron/notifications/base.py

@@ -2,6 +2,7 @@
 import logging
 import os
 from abc import ABC, abstractmethod
+from pathlib import Path
 from typing import Any, Mapping, Optional
 
 from jinja2 import Environment, StrictUndefined
@@ -58,6 +59,31 @@ def get_template(self) -> str:
         return template
 
 
+class FileTemplateProvider(TemplateProvider):
+    """
+    Load a notification template from a file on disk.
+
+    Parameters
+    ----------
+    path : pathlib.Path
+        Path to a file containing the template string.
+    """
+
+    def __init__(self, path: Path) -> None:
+        self.path = path
+
+    def get_template(self) -> str:
+        """
+        Return a Jinja2 template string from the configured file path.
+
+        Returns
+        -------
+        str
+            Template contents as a string.
+        """
+        return self.path.read_text(encoding="utf-8")
+
+
 class RenderedTemplateNotificationHandler(ABC):
     """
     Base class for SQS-driven notifications using Jinja2 templates.

src/lambdacron/render.py

@@ -0,0 +1,122 @@
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any, Mapping, TextIO
+
+from jinja2 import TemplateError
+
+from lambdacron.notifications.base import (
+    FileTemplateProvider,
+    RenderedTemplateNotificationHandler,
+)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Render a notification template using LambdaCron task output data."
+    )
+    parser.add_argument(
+        "output_json",
+        metavar="output.json",
+        nargs="?",
+        default="-",
+        help=(
+            "JSON file containing the direct output from `_perform_task` "
+            "(a result-type-to-payload object). Use '-' or omit this argument "
+            "to read from stdin."
+        ),
+    )
+    parser.add_argument(
+        "-t",
+        "--template",
+        required=True,
+        type=Path,
+        help="Path to the Jinja2 template file.",
+    )
+    parser.add_argument(
+        "-r",
+        "--result-type",
+        required=True,
+        help="Result type key to emulate from the LambdaCron task output.",
+    )
+    return parser
+
+
+class RenderNotificationHandler(RenderedTemplateNotificationHandler):
+    def __init__(self, *, template_path: Path, stream: TextIO | None = None) -> None:
+        super().__init__(
+            template_providers={"body": FileTemplateProvider(template_path)},
+            include_result_type=True,
+        )
+        self.stream = stream or sys.stdout
+
+    def notify(
+        self,
+        *,
+        result: Mapping[str, Any],
+        rendered: Mapping[str, str],
+        record: Mapping[str, Any],
+    ) -> None:
+        print(rendered["body"], file=self.stream)
+
+    def render_payload(self, *, payload_json: str, result_type: str) -> None:
+        event = {
+            "Records": [
+                {
+                    "body": payload_json,
+                    "eventSource": "aws:sqs",
+                    "messageAttributes": {
+                        "result_type": {
+                            "DataType": "String",
+                            "StringValue": result_type,
+                        }
+                    },
+                }
+            ]
+        }
+        self.lambda_handler(event=event, context=None)
+
+
+def read_payload_json(source: str, *, stdin: TextIO) -> str:
+    if source == "-":
+        payload_json = stdin.read()
+        if not payload_json.strip():
+            raise ValueError("stdin did not contain JSON content")
+        return payload_json
+    return Path(source).read_text(encoding="utf-8")
+
+
+def extract_result_payload(payload_json: str, *, result_type: str) -> str:
+    try:
+        payload = json.loads(payload_json)
+    except json.JSONDecodeError as exc:
+        raise ValueError("Task output must be valid JSON") from exc
+    if not isinstance(payload, dict):
+        raise ValueError("Task output must be a JSON object keyed by result type")
+    selected = payload.get(result_type)
+    if not isinstance(selected, dict):
+        raise ValueError(
+            f"Result payload for type '{result_type}' must be a JSON object"
+        )
+    return json.dumps(selected)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        payload_json = read_payload_json(args.output_json, stdin=sys.stdin)
+        payload_json = extract_result_payload(
+            payload_json, result_type=args.result_type
+        )
+        handler = RenderNotificationHandler(template_path=args.template)
+        handler.render_payload(payload_json=payload_json, result_type=args.result_type)
+    except (OSError, ValueError, TemplateError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/notifications/test_base.py

@@ -5,6 +5,7 @@
 
 from lambdacron.notifications.base import (
     EnvVarTemplateProvider,
+    FileTemplateProvider,
     RenderedTemplateNotificationHandler,
 )
 
@@ -51,6 +52,22 @@ def test_env_var_template_provider_requires_value(monkeypatch):
         provider.get_template()
 
 
+def test_file_template_provider_reads_template(tmp_path):
+    template_path = tmp_path / "template.jinja2"
+    template_path.write_text("Hello {{ name }}", encoding="utf-8")
+
+    provider = FileTemplateProvider(template_path)
+
+    assert provider.get_template() == "Hello {{ name }}"
+
+
+def test_file_template_provider_raises_for_missing_file(tmp_path):
+    provider = FileTemplateProvider(tmp_path / "missing.jinja2")
+
+    with pytest.raises(FileNotFoundError):
+        provider.get_template()
+
+
 def test_notification_handler_parses_sqs_json_body(monkeypatch):
     monkeypatch.setenv("TEMPLATE", "Status {{ status }}")
     handler = CapturingHandler(template_providers={"body": EnvVarTemplateProvider()})

tests/test_render.py

@@ -0,0 +1,165 @@
+import json
+from io import StringIO
+
+import pytest
+
+from lambdacron import render
+
+
+def test_render_main_renders_with_long_flags(tmp_path, capsys):
+    template_path = tmp_path / "template.jinja2"
+    output_path = tmp_path / "output.json"
+    template_path.write_text(
+        "Status {{ status }} ({{ result_type }})", encoding="utf-8"
+    )
+    output_path.write_text(json.dumps({"success": {"status": "ok"}}), encoding="utf-8")
+
+    code = render.main(
+        [
+            "--template",
+            str(template_path),
+            "--result-type",
+            "success",
+            str(output_path),
+        ]
+    )
+
+    captured = capsys.readouterr()
+    assert code == 0
+    assert captured.out == "Status ok (success)\n"
+    assert captured.err == ""
+
+
+def test_render_main_short_flags_preserve_payload_result_type(tmp_path, capsys):
+    template_path = tmp_path / "template.jinja2"
+    output_path = tmp_path / "output.json"
+    template_path.write_text("{{ result_type }}", encoding="utf-8")
+    output_path.write_text(
+        json.dumps({"attribute": {"status": "ok", "result_type": "payload"}}),
+        encoding="utf-8",
+    )
+
+    code = render.main(["-t", str(template_path), "-r", "attribute", str(output_path)])
+
+    captured = capsys.readouterr()
+    assert code == 0
+    assert captured.out == "payload\n"
+    assert captured.err == ""
+
+
+def test_render_main_rejects_non_object_json(tmp_path, capsys):
+    template_path = tmp_path / "template.jinja2"
+    output_path = tmp_path / "output.json"
+    template_path.write_text("{{ value }}", encoding="utf-8")
+    output_path.write_text(json.dumps(["not", "an", "object"]), encoding="utf-8")
+
+    code = render.main(["-t", str(template_path), "-r", "success", str(output_path)])
+
+    captured = capsys.readouterr()
+    assert code == 1
+    assert "Task output must be a JSON object keyed by result type" in captured.err
+
+
+def test_render_main_uses_strict_undefined(tmp_path, capsys):
+    template_path = tmp_path / "template.jinja2"
+    output_path = tmp_path / "output.json"
+    template_path.write_text("{{ missing }}", encoding="utf-8")
+    output_path.write_text(json.dumps({"success": {"status": "ok"}}), encoding="utf-8")
+
+    code = render.main(["-t", str(template_path), "-r", "success", str(output_path)])
+
+    captured = capsys.readouterr()
+    assert code == 1
+    assert "undefined" in captured.err.lower()
+
+
+def test_render_main_requires_template_and_result_type(tmp_path):
+    output_path = tmp_path / "output.json"
+    output_path.write_text(json.dumps({"success": {"status": "ok"}}), encoding="utf-8")
+
+    with pytest.raises(SystemExit) as exc_info:
+        render.main([str(output_path)])
+
+    assert exc_info.value.code == 2
+
+
+def test_render_main_reads_from_stdin_when_file_is_omitted(
+    tmp_path, capsys, monkeypatch
+):
+    template_path = tmp_path / "template.jinja2"
+    template_path.write_text(
+        "Status {{ status }} ({{ result_type }})", encoding="utf-8"
+    )
+    monkeypatch.setattr(
+        "sys.stdin", StringIO(json.dumps({"success": {"status": "ok"}}))
+    )
+
+    code = render.main(["-t", str(template_path), "-r", "success"])
+
+    captured = capsys.readouterr()
+    assert code == 0
+    assert captured.out == "Status ok (success)\n"
+    assert captured.err == ""
+
+
+def test_render_main_reads_from_stdin_with_dash(tmp_path, capsys, monkeypatch):
+    template_path = tmp_path / "template.jinja2"
+    template_path.write_text("{{ result_type }}", encoding="utf-8")
+    monkeypatch.setattr(
+        "sys.stdin", StringIO(json.dumps({"success": {"status": "ok"}}))
+    )
+
+    code = render.main(["-t", str(template_path), "-r", "success", "-"])
+
+    captured = capsys.readouterr()
+    assert code == 0
+    assert captured.out == "success\n"
+    assert captured.err == ""
+
+
+def test_render_main_extracts_payload_from_result_map_stdin(
+    tmp_path, capsys, monkeypatch
+):
+    template_path = tmp_path / "template.jinja2"
+    template_path.write_text(
+        "Status {{ status }} ({{ result_type }})", encoding="utf-8"
+    )
+    monkeypatch.setattr(
+        "sys.stdin",
+        StringIO(
+            json.dumps({"success": {"status": "ok"}, "failure": {"status": "bad"}})
+        ),
+    )
+
+    code = render.main(["-t", str(template_path), "-r", "success"])
+
+    captured = capsys.readouterr()
+    assert code == 0
+    assert captured.out == "Status ok (success)\n"
+    assert captured.err == ""
+
+
+def test_render_main_rejects_task_output_missing_result_type(tmp_path, capsys):
+    template_path = tmp_path / "template.jinja2"
+    output_path = tmp_path / "output.json"
+    template_path.write_text("{{ status }}", encoding="utf-8")
+    output_path.write_text(json.dumps({"failure": {"status": "bad"}}), encoding="utf-8")
+
+    code = render.main(["-t", str(template_path), "-r", "success", str(output_path)])
+
+    captured = capsys.readouterr()
+    assert code == 1
+    assert "Result payload for type 'success' must be a JSON object" in captured.err
+
+
+def test_render_main_rejects_non_object_result_payload(tmp_path, capsys):
+    template_path = tmp_path / "template.jinja2"
+    output_path = tmp_path / "output.json"
+    template_path.write_text("{{ status }}", encoding="utf-8")
+    output_path.write_text(json.dumps({"success": "ok"}), encoding="utf-8")
+
+    code = render.main(["-t", str(template_path), "-r", "success", str(output_path)])
+
+    captured = capsys.readouterr()
+    assert code == 1
+    assert "Result payload for type 'success' must be a JSON object" in captured.err