import urllib.error from email.message import Message from datetime import UTC, datetime from zoneinfo import ZoneInfo from beaver_agent.hands.calendars import ( Calendars, Feed, calendars, events_in, feeds, is_routine, short, unfold, ) ZONE = ZoneInfo("Europe/Warsaw") NOW = datetime(2026, 9, 1, 12, 0, tzinfo=UTC) # 14:00 в Варшаве MCPS = ( "personal=https://calendar-mcp.com/api/mcp?email=b%40b.com" "&icsUrl=https%3A%2F%2Fcalendar.google.com%2Fical%2Fprivate-abc%2Fbasic.ics," "student=https://calendar-mcp.com/api/mcp?icsUrl=https%3A%2F%2Fusos.pl%2Fics%3Fkey%3Dz," "bare=https://calendar-mcp.com/mcp/AAAA" ) def ics(*events: str) -> str: body = "".join(f"BEGIN:VEVENT\n{e.strip()}\nEND:VEVENT\n" for e in events) return f"BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//test//EN\n{body}END:VCALENDAR\n" def starts(text: str, *, days: int = 7) -> list[str]: found = events_in(text, zone=ZONE, since=NOW.astimezone(ZONE), days=days) return [ f"{e.start.astimezone(ZONE):%m-%d %H:%M} {e.title}" for e in sorted(found, key=lambda e: e.start) ] def test_feeds_take_the_ical_url_out_of_the_same_mcp_urls(): assert [f.name for f in feeds(MCPS)] == ["personal", "student"] assert feeds(MCPS)[0].url == ( "https://calendar.google.com/ical/private-abc/basic.ics" ) assert feeds(MCPS)[1].url == "https://usos.pl/ics?key=z" assert feeds("") == () assert feeds("plain=https://example.com/plan.ics")[0].url.endswith("plan.ics") # MCP-рука остаётся у модели, даже если ical-фида в записи нет assert [m.name for m in calendars(MCPS)] == [ "calendar-personal", "calendar-student", "calendar-bare", ] def test_folded_lines_are_glued_back(): assert unfold("SUMMARY:длин\r\n ное\r\nUID:x") == ["SUMMARY:длинное", "UID:x"] def test_window_sorts_by_start_and_keeps_only_the_next_days(): text = ics( "SUMMARY:позже\nDTSTART:20260904T080000Z", "SUMMARY:раньше\nDTSTART:20260902T080000Z", "SUMMARY:за окном\nDTSTART:20260920T080000Z", "SUMMARY:уже прошло\nDTSTART:20260801T080000Z", ) assert starts(text) == ["09-02 10:00 раньше", "09-04 10:00 позже"] def test_all_day_event_of_today_still_counts_as_upcoming(): text = ics("SUMMARY:отпуск\nDTSTART;VALUE=DATE:20260901\nDTEND;VALUE=DATE:20260902") found = events_in(text, zone=ZONE, since=NOW.astimezone(ZONE), days=7) assert len(found) == 1 assert found[0].all_day assert f"{found[0].start:%H:%M}" == "00:00" def test_place_takes_the_room_from_the_description_and_the_address_from_location(): text = ics( "SUMMARY:Analiza matematyczna II - Egzamin\n" "DTSTART;VALUE=DATE-TIME:20260903T110000\n" "DESCRIPTION:Sala: 3\\nA23 - CW - Centrum Wykładowe\\n\n" "LOCATION:ul. Piotrowo 2\\, 61-138 Poznań\\, Polska" ) found = events_in(text, zone=ZONE, since=NOW.astimezone(ZONE), days=7) assert found[0].place == "Sala: 3, ul. Piotrowo 2, 61-138 Poznań, Polska" # плавающее время читается в зоне агента, а не в UTC assert f"{found[0].start:%H:%M}" == "11:00" def test_cancelled_events_are_skipped(): text = ics("SUMMARY:отменено\nDTSTART:20260902T080000Z\nSTATUS:CANCELLED") assert starts(text) == [] def test_weekly_repeat_expands_and_exdate_with_moved_instance_are_subtracted(): series = ( "UID:w1\nSUMMARY:польский\nDTSTART;TZID=Europe/Warsaw:20260824T100000\n" "RRULE:FREQ=WEEKLY;BYDAY=MO,WE" ) assert starts(ics(series)) == ["09-02 10:00 польский", "09-07 10:00 польский"] moved = ics( series + "\nEXDATE;TZID=Europe/Warsaw:20260902T100000", "UID:w1\nSUMMARY:польский (перенос)\n" "RECURRENCE-ID;TZID=Europe/Warsaw:20260907T100000\n" "DTSTART;TZID=Europe/Warsaw:20260907T173000", ) assert starts(moved) == ["09-07 17:30 польский (перенос)"] def test_daily_repeat_stops_on_until_and_count(): until = ics( "SUMMARY:завтрак\nDTSTART:20260801T060000Z\n" "RRULE:FREQ=DAILY;UNTIL=20260903T060000Z" ) assert starts(until) == ["09-02 08:00 завтрак", "09-03 08:00 завтрак"] counted = ics("SUMMARY:курс\nDTSTART:20260902T060000Z\nRRULE:FREQ=DAILY;COUNT=3") assert len(starts(counted)) == 3 yearly = ics( "SUMMARY:часы\nDTSTART:20200906T060000Z\nRRULE:FREQ=YEARLY;BYMONTH=9;BYDAY=1SU" ) assert starts(yearly) == ["09-06 08:00 часы"] def test_a_working_week_is_routine_but_one_weekday_a_week_is_an_event(): text = ics( "SUMMARY:пара\nDTSTART;TZID=Europe/Warsaw:20260831T100000\n" "RRULE:FREQ=WEEKLY;BYDAY=MO", "SUMMARY:зарядка\nDTSTART;TZID=Europe/Warsaw:20260831T070000\n" "RRULE:FREQ=WEEKLY;BYDAY=MO,TU,WE,TH,FR", ) found = events_in(text, zone=ZONE, since=NOW.astimezone(ZONE), days=7) assert {e.title for e in found if e.routine} == {"зарядка"} assert {e.title for e in found if not e.routine} == {"пара"} assert len([e for e in found if e.routine]) == 5 def test_routine_is_a_rule_that_fires_once_a_day_or_more_often(): assert is_routine({"FREQ": "DAILY"}) assert is_routine({"FREQ": "WEEKLY", "BYDAY": "MO,TU,WE,TH,FR,SA"}) assert not is_routine({"FREQ": "WEEKLY", "BYDAY": "MO,WE,FR"}) assert not is_routine({"FREQ": "WEEKLY"}) assert not is_routine({"FREQ": "MONTHLY"}) # через день - это уже не быт, такое видно поимённо assert not is_routine({"FREQ": "DAILY", "INTERVAL": "2"}) def test_a_single_event_is_never_routine(): text = ics("SUMMARY:Egzamin\nDTSTART:20260903T090000Z") assert not events_in(text, zone=ZONE, since=NOW.astimezone(ZONE), days=7)[0].routine def test_a_feed_that_fails_becomes_a_line_not_an_exception(): class Broken(Calendars): def fetch(self, url: str) -> str: raise urllib.error.HTTPError(url, 403, "Forbidden", Message(), None) broken = Broken((Feed("student", "https://usos.pl/ics"),), tz="Europe/Warsaw") found = broken.upcoming(NOW) assert found.read == 0 assert found.events == () assert found.errors == ("student: HTTP 403",) def test_short_reason_is_one_line_and_never_carries_the_url(): assert short(TimeoutError()) == "таймаут" assert short(urllib.error.URLError("no route")).startswith("сеть:") assert short(ValueError("а\nб")) == "ValueError: а б"