diff --git a/agenda/geomob.py b/agenda/geomob.py index f7383b3..759dd9e 100644 --- a/agenda/geomob.py +++ b/agenda/geomob.py @@ -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) diff --git a/tests/test_geomob.py b/tests/test_geomob.py index a2d4652..b0bc435 100644 --- a/tests/test_geomob.py +++ b/tests/test_geomob.py @@ -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 = """ @@ -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 = """ @@ -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 = """ @@ -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() diff --git a/tests/test_geomob_recording.py b/tests/test_geomob_recording.py new file mode 100644 index 0000000..fb4c20e --- /dev/null +++ b/tests/test_geomob_recording.py @@ -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"]