agenda/tests/test_geomob_recording.py

132 lines
5.2 KiB
Python

"""Check automatic Geomob recording and failure notifications."""
import subprocess
from pathlib import Path
from datetime import date
from unittest.mock import patch
import pytest
import yaml
from agenda.geomob import GeomobEvent, conference_page_links, geomob_email, record_event
EVENT = GeomobEvent(date(2026, 10, 15), "/post/oct-2026", "#geomobLON")
BASE_URL = "https://thegeomob.com"
def test_record_event_success() -> None:
"""Validate before committing, and restrict the commit to conferences."""
with (
patch("agenda.geomob.subprocess.run") as run,
patch("agenda.add_new_conference.add_new_conference", return_value=True) as add,
):
run.return_value.stdout = ""
result = record_event(EVENT, BASE_URL)
commands = [call.args[0] for call in run.call_args_list]
assert commands[1] == ["git", "pull", "--rebase"]
assert commands[2][0] == "python3"
assert commands[2][1].endswith("/validate_yaml.py")
assert commands[3] == ["git", "add", "--", "conferences.yaml"]
assert commands[4][0:3] == ["git", "commit", "--only"]
assert commands[-1] == ["git", "push"]
add.assert_called_once()
assert "committed and pushed" in result
@pytest.mark.parametrize("failed_call", [1, 2, 3, 4, 5])
def test_command_failure_stops_workflow(failed_call: int) -> None:
"""Every command failure appears in the email and stops later commands."""
success = subprocess.CompletedProcess(["git"], 0, stdout="", stderr="")
error = subprocess.CalledProcessError(1, ["failed"], stderr="failure details")
with (
patch(
"agenda.geomob.subprocess.run",
side_effect=[success] * failed_call + [error],
) as run,
patch("agenda.add_new_conference.add_new_conference", return_value=True),
):
result = record_event(EVENT, BASE_URL)
assert run.call_count == failed_call + 1
assert "FAILED" in result
assert "failure details" in result
_, body = geomob_email([EVENT], BASE_URL, {EVENT: result})
assert result in body
def test_dirty_repository() -> None:
"""Do not pull or add a conference over existing edits."""
with (
patch("agenda.geomob.subprocess.run") as run,
patch("agenda.add_new_conference.add_new_conference") as add,
):
run.return_value.stdout = " M conferences.yaml"
assert "uncommitted changes" in record_event(EVENT, BASE_URL)
assert run.call_count == 1
add.assert_not_called()
def test_add_failure() -> None:
"""Generation errors are reported without validating or committing."""
with (
patch("agenda.geomob.subprocess.run") as run,
patch(
"agenda.add_new_conference.add_new_conference",
side_effect=ValueError("bad event"),
),
):
run.return_value.stdout = ""
assert "FAILED at add conference: bad event" in record_event(EVENT, BASE_URL)
assert run.call_count == 2
def test_duplicate_event() -> None:
"""Existing events still validate and push without an empty commit."""
with (
patch("agenda.geomob.subprocess.run") as run,
patch("agenda.add_new_conference.add_new_conference", return_value=False),
):
run.return_value.stdout = ""
assert "Already recorded" in record_event(EVENT, BASE_URL)
assert run.call_count == 4
assert run.call_args.args[0] == ["git", "push"]
def test_email_keeps_event_url_and_adds_conference_page(tmp_path: Path) -> None:
"""A saved event links to its canonical agenda page alongside the original URL."""
event = GeomobEvent(
date(2026, 10, 28), "/post/oct-28th-2026-geomobcgn-details", "#geomobCGN"
)
conference = {
"name": "Geomob Cologne",
"location": "Cologne",
"country": "de",
"dates": {"status": "exact", "start": event.date, "end": event.date},
"url": BASE_URL + event.href,
}
(tmp_path / "conferences.yaml").write_text(yaml.safe_dump([conference]))
links = conference_page_links([event], BASE_URL, str(tmp_path))
expected = (
"https://edwardbetts.com/agenda/conference/2026-10-28/geomob-cologne-cologne"
)
assert links == {event: expected}
subject, body = geomob_email(
[event], BASE_URL, {event: "Recorded and pushed."}, links
)
assert subject == "1 New Geomob Event(s) Announced"
assert "URL: https://thegeomob.com/post/oct-28th-2026-geomobcgn-details" in body
assert f"Conference page: {expected}" in body
assert "Recording: Recorded and pushed." in body
changed_date = GeomobEvent(date(2026, 10, 29), event.href, event.hashtag)
assert conference_page_links([changed_date], BASE_URL, str(tmp_path)) == {}
def test_missing_conference_data_does_not_hide_event(tmp_path: Path) -> None:
"""Unrecorded events still generate their original notification without a broken link."""
links = conference_page_links([EVENT], BASE_URL, str(tmp_path))
assert links == {}
_, body = geomob_email(
[EVENT], BASE_URL, {EVENT: "FAILED at add conference"}, links
)
assert "URL: " + BASE_URL + EVENT.href in body
assert "FAILED at add conference" in body
assert "Conference page:" not in body