edward/agenda
1
0
Fork 0
Code Issues 9 Pull requests Projects Releases Packages Wiki Activity

Automatically record new Geomob events in personal data

Browse source
This commit is contained in:
Edward Betts 2026-10-01 12:46:33 +01:00
parent 51b8bf6ecb
commit 22976868ce
3 changed files with 210 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

83
agenda/geomob.py
View file

@ -1,15 +1,18 @@
"""Geomob events."""
import os
import subprocess
from pathlib import Path
from dataclasses import dataclass
from datetime import date, datetime
from typing import List
import dateutil.parser
import flask
import lxml.html
import lxml.html # type: ignore[import-untyped]
import requests
import agenda.add_new_conference
import agenda.mail
import agenda.utils
@ -49,7 +52,11 @@ def find_new_events(
return list(set(cur) - set(prev))
def geomob_email(new_events: list[GeomobEvent], base_url: str) -> tuple[str, str]:
def geomob_email(
new_events: list[GeomobEvent],
base_url: str,
recording_results: dict[GeomobEvent, str] | None = None,
) -> tuple[str, str]:
"""Generate email subject and body for new events.
Args:
@ -73,6 +80,8 @@ def geomob_email(new_events: list[GeomobEvent], base_url: str) -> tuple[str, str
else:
assert "//" not in url, f"Double slash found in URL: {url}"
event_details = f"Date: {event.date}\nURL: {url}\nHashtag: {event.hashtag}\n"
if recording_results is not None:
event_details += f"Recording: {recording_results[event]}\n"
body_lines.append(event_details)
body_lines.append("-" * 40)
@ -80,6 +89,73 @@ def geomob_email(new_events: list[GeomobEvent], base_url: str) -> tuple[str, str
return (subject, body)
def record_event(event: GeomobEvent, base_url: str) -> str:
"""Record an event, reporting failures without suppressing its notification.
Leave failed changes in place for inspection. Only conferences.yaml is staged
and committed, and an initially dirty repository is never modified.
"""
repository = Path("~/src/personal-data").expanduser()
yaml_path = repository / "conferences.yaml"
validator = Path(__file__).resolve().parent.parent / "validate_yaml.py"
stage = "check repository"
recorded = False
def run(command: list[str]) -> str:
result = subprocess.run(
command,
cwd=repository,
capture_output=True,
text=True,
timeout=300,
check=True,
)
return result.stdout.strip()
try:
if run(["git", "status", "--porcelain"]):
return "FAILED at check repository: uncommitted changes; no steps run."
stage = "git pull --rebase"
run(["git", "pull", "--rebase"])
stage = "add conference"
added = agenda.add_new_conference.add_new_conference(
base_url + event.href, str(yaml_path)
)
recorded = True
stage = "validate_yaml.py"
run(["python3", str(validator)])
if added:
stage = "git add conferences.yaml"
run(["git", "add", "--", "conferences.yaml"])
stage = "git commit"
run(
[
"git",
"commit",
"--only",
"-m",
f"Add {event.hashtag} event on {event.date}",
"--",
"conferences.yaml",
]
)
stage = "git push"
run(["git", "push"])
return (
"Recorded in conferences.yaml, validated, committed and pushed."
if added
else "Already recorded in conferences.yaml; validated and pushed."
)
except Exception as exc:
details = str(exc)
if isinstance(exc, subprocess.CalledProcessError):
details += f"\n{exc.stdout or ''}\n{exc.stderr or ''}"
elif isinstance(exc, subprocess.TimeoutExpired):
details += f"\n{exc.stdout!r}\n{exc.stderr!r}"
state = "Recorded locally; " if recorded else ""
return f"{state}FAILED at {stage}: {details.strip()}"
def get_cached_upcoming_events_list(geomob_dir: str) -> list[GeomobEvent]:
"""Get known geomob events."""
filename = agenda.utils.get_most_recent_file(geomob_dir, "html")
@ -106,5 +182,6 @@ def update(config: flask.config.Config) -> None:
return
base_url = "https://thegeomob.com"
subject, body = geomob_email(new_events, base_url)
recording_results = {event: record_event(event, base_url) for event in new_events}
subject, body = geomob_email(new_events, base_url, recording_results)
agenda.mail.send_mail(config, subject, body)

78
tests/test_geomob.py
View file

@ -3,7 +3,8 @@
from datetime import date
from unittest.mock import Mock, mock_open, patch
import lxml.html
import flask
import lxml.html # type: ignore[import-untyped]
import pytest
from agenda.geomob import (
@ -16,7 +17,7 @@ from agenda.geomob import (
)
def test_geomob_event_dataclass():
def test_geomob_event_dataclass() -> None:
"""Test GeomobEvent dataclass creation and properties."""
event = GeomobEvent(
date=date(2024, 7, 15), href="/event/london-2024-07-15", hashtag="#geomobLDN"
@ -27,17 +28,17 @@ def test_geomob_event_dataclass():
assert event.hashtag == "#geomobLDN"
def test_geomob_event_frozen():
def test_geomob_event_frozen() -> None:
"""Test that GeomobEvent is frozen (immutable)."""
event = GeomobEvent(
date=date(2024, 7, 15), href="/event/london-2024-07-15", hashtag="#geomobLDN"
)
with pytest.raises(AttributeError):
event.date = date(2024, 8, 15)
event.date = date(2024, 8, 15) # type: ignore[misc]
def test_extract_events_with_valid_html():
def test_extract_events_with_valid_html() -> None:
"""Test extracting events from valid HTML."""
html_content = """
<html>
@ -64,7 +65,7 @@ def test_extract_events_with_valid_html():
assert events[1].hashtag == "#geomobBER"
def test_extract_events_empty_list():
def test_extract_events_empty_list() -> None:
"""Test extracting events from HTML with no events."""
html_content = """
<html>
@ -81,7 +82,7 @@ def test_extract_events_empty_list():
assert events == []
def test_extract_events_no_event_list():
def test_extract_events_no_event_list() -> None:
"""Test extracting events from HTML with no event list."""
html_content = """
<html>
@ -97,7 +98,7 @@ def test_extract_events_no_event_list():
assert events == []
def test_find_new_events_with_new_events():
def test_find_new_events_with_new_events() -> None:
"""Test finding new events when there are some."""
prev_events = [
GeomobEvent(date(2024, 7, 15), "/event/london-2024-07-15", "#geomobLDN"),
@ -118,7 +119,7 @@ def test_find_new_events_with_new_events():
assert new_events[0].hashtag == "#geomobPAR"
def test_find_new_events_no_new_events():
def test_find_new_events_no_new_events() -> None:
"""Test finding new events when there are none."""
events = [
GeomobEvent(date(2024, 7, 15), "/event/london-2024-07-15", "#geomobLDN"),
@ -129,7 +130,7 @@ def test_find_new_events_no_new_events():
assert new_events == []
def test_find_new_events_empty_previous():
def test_find_new_events_empty_previous() -> None:
"""Test finding new events when previous list is empty."""
cur_events = [
GeomobEvent(date(2024, 7, 15), "/event/london-2024-07-15", "#geomobLDN"),
@ -141,7 +142,7 @@ def test_find_new_events_empty_previous():
assert set(new_events) == set(cur_events)
def test_geomob_email_single_event():
def test_geomob_email_single_event() -> None:
"""Test generating email for a single new event."""
events = [GeomobEvent(date(2024, 7, 15), "/event/london-2024-07-15", "#geomobLDN")]
@ -156,7 +157,7 @@ def test_geomob_email_single_event():
assert "----------------------------------------" in body
def test_geomob_email_single_event_no_protocol():
def test_geomob_email_single_event_no_protocol() -> None:
"""Test generating email for a single new event with base URL without protocol."""
events = [GeomobEvent(date(2024, 7, 15), "/event/london-2024-07-15", "#geomobLDN")]
@ -171,7 +172,7 @@ def test_geomob_email_single_event_no_protocol():
assert "----------------------------------------" in body
def test_geomob_email_multiple_events():
def test_geomob_email_multiple_events() -> None:
"""Test generating email for multiple new events."""
events = [
GeomobEvent(date(2024, 7, 15), "/event/london-2024-07-15", "#geomobLDN"),
@ -186,7 +187,7 @@ def test_geomob_email_multiple_events():
assert body.count("----------------------------------------") == 2
def test_geomob_email_no_double_slash():
def test_geomob_email_no_double_slash() -> None:
"""Test that URL construction doesn't create double slashes."""
events = [GeomobEvent(date(2024, 7, 15), "/event/london-2024-07-15", "#geomobLDN")]
@ -196,13 +197,13 @@ def test_geomob_email_no_double_slash():
assert "https://thegeomob.com/event/london-2024-07-15" in body
def test_geomob_email_empty_list():
def test_geomob_email_empty_list() -> None:
"""Test that geomob_email raises assertion error with empty list."""
with pytest.raises(AssertionError):
geomob_email([], "https://thegeomob.com")
def test_geomob_email_detects_double_slash_in_path():
def test_geomob_email_detects_double_slash_in_path() -> None:
"""Test that the function detects double slashes in the URL path."""
events = [GeomobEvent(date(2024, 7, 15), "//event/london-2024-07-15", "#geomobLDN")]
@ -213,7 +214,9 @@ def test_geomob_email_detects_double_slash_in_path():
@patch("agenda.utils.get_most_recent_file")
@patch("lxml.html.parse")
def test_get_cached_upcoming_events_list_with_file(mock_parse, mock_get_file):
def test_get_cached_upcoming_events_list_with_file(
mock_parse: Mock, mock_get_file: Mock
) -> None:
"""Test getting cached events when file exists."""
mock_get_file.return_value = "/path/to/recent.html"
@ -237,7 +240,7 @@ def test_get_cached_upcoming_events_list_with_file(mock_parse, mock_get_file):
@patch("agenda.utils.get_most_recent_file")
def test_get_cached_upcoming_events_list_no_file(mock_get_file):
def test_get_cached_upcoming_events_list_no_file(mock_get_file: Mock) -> None:
"""Test getting cached events when no file exists."""
mock_get_file.return_value = None
@ -254,16 +257,16 @@ def test_get_cached_upcoming_events_list_no_file(mock_get_file):
@patch("builtins.open", new_callable=mock_open)
@patch("agenda.geomob.datetime")
def test_update_with_new_events(
mock_datetime,
mock_open_file,
mock_join,
mock_get_cached,
mock_requests,
mock_send_mail,
):
mock_datetime: Mock,
mock_open_file: Mock,
mock_join: Mock,
mock_get_cached: Mock,
mock_requests: Mock,
mock_send_mail: Mock,
) -> None:
"""Test update function when there are new events."""
# Mock config
config = {"DATA_DIR": "/data"}
config = flask.config.Config("/", {"DATA_DIR": "/data"})
# Mock datetime
mock_now = Mock()
@ -293,7 +296,8 @@ def test_update_with_new_events(
]
mock_extract.return_value = cur_events
update(config)
with patch("agenda.geomob.record_event", return_value="Recorded and pushed."):
update(config)
# Verify file was written
mock_open_file.assert_called_once_with(
@ -309,9 +313,9 @@ def test_update_with_new_events(
@patch("requests.get")
@patch("agenda.geomob.get_cached_upcoming_events_list")
def test_update_no_changes(mock_get_cached, mock_requests):
def test_update_no_changes(mock_get_cached: Mock, mock_requests: Mock) -> None:
"""Test update function when there are no changes."""
config = {"DATA_DIR": "/data"}
config = flask.config.Config("/", {"DATA_DIR": "/data"})
# Mock identical events
events = [GeomobEvent(date(2024, 7, 15), "/event/london-2024-07-15", "#geomobLDN")]
@ -341,15 +345,15 @@ def test_update_no_changes(mock_get_cached, mock_requests):
@patch("builtins.open", new_callable=mock_open)
@patch("agenda.geomob.datetime")
def test_update_events_changed_but_no_new(
mock_datetime,
mock_open_file,
mock_join,
mock_get_cached,
mock_requests,
mock_send_mail,
):
mock_datetime: Mock,
mock_open_file: Mock,
mock_join: Mock,
mock_get_cached: Mock,
mock_requests: Mock,
mock_send_mail: Mock,
) -> None:
"""Test update function when events changed but no new ones added."""
config = {"DATA_DIR": "/data"}
config = flask.config.Config("/", {"DATA_DIR": "/data"})
# Mock datetime
mock_now = Mock()

89
tests/test_geomob_recording.py Normal file
View file

@ -0,0 +1,89 @@
"""Check automatic Geomob recording and failure notifications."""
import subprocess
from datetime import date
from unittest.mock import patch
import pytest
from agenda.geomob import GeomobEvent, 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"]