import datetime import hashlib import operator from copy import deepcopy from random import shuffle from typing import Union, List, Dict, Optional, AnyStr import dateutil.parser import pytest from pytz import timezone, utc from sync_ics2gcal import CalendarSync, DateDateTime from sync_ics2gcal.gcal import EventDateOrDateTime, EventData, EventList def sha1(s: AnyStr) -> str: h = hashlib.sha1() h.update(str(s).encode("utf8") if isinstance(s, str) else s) return h.hexdigest() def gen_events( start: int, stop: int, start_time: DateDateTime, no_time: bool = False, ) -> EventList: duration: datetime.timedelta date_key: str date_end: str if no_time: start_time = datetime.date(start_time.year, start_time.month, start_time.day) duration = datetime.date(1, 1, 2) - datetime.date(1, 1, 1) date_key = "date" date_end = "" else: start_time = utc.normalize(start_time.astimezone(utc)).replace(tzinfo=None) # type: ignore duration = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1) date_key = "dateTime" date_end = "Z" result: EventList = [] for i in range(start, stop): event_start = start_time + (duration * i) event_end = event_start + duration updated: DateDateTime = event_start if no_time: updated = datetime.datetime( updated.year, updated.month, updated.day, 0, 0, 0, 1, tzinfo=utc ) event: EventData = { "summary": "test event __ {}".format(i), "location": "la la la {}".format(i), "description": "test TEST -- test event {}".format(i), "iCalUID": "{}@test.com".format(sha1("test - event {}".format(i))), "updated": updated.isoformat() + "Z", "created": updated.isoformat() + "Z", "start": {date_key: event_start.isoformat() + date_end}, # type: ignore "end": {date_key: event_end.isoformat() + date_end}, # type: ignore } result.append(event) return result def gen_list_to_compare(start: int, stop: int) -> List[Dict[str, str]]: result: List[Dict[str, str]] = [] for i in range(start, stop): result.append({"iCalUID": "test{:06d}".format(i)}) return result def get_start_date(event: EventData) -> DateDateTime: event_start: EventDateOrDateTime = event["start"] start_date: Optional[str] = None is_date = False if "date" in event_start: start_date = event_start["date"] # type: ignore is_date = True if "dateTime" in event_start: start_date = event_start["dateTime"] # type: ignore result: DateDateTime = dateutil.parser.parse(str(start_date)) if is_date: result = datetime.date(result.year, result.month, result.day) return result def test_compare(): part_len = 20 # [1..2n] lst_src = gen_list_to_compare(1, 1 + part_len * 2) # [n..3n] lst_dst = gen_list_to_compare(1 + part_len, 1 + part_len * 3) lst_src_rnd = deepcopy(lst_src) lst_dst_rnd = deepcopy(lst_dst) shuffle(lst_src_rnd) shuffle(lst_dst_rnd) to_ins, to_upd, to_del = CalendarSync._events_list_compare(lst_src_rnd, lst_dst_rnd) assert len(to_ins) == part_len assert len(to_upd) == part_len assert len(to_del) == part_len assert sorted(to_ins, key=lambda x: x["iCalUID"]) == lst_src[:part_len] assert sorted(to_del, key=lambda x: x["iCalUID"]) == lst_dst[part_len:] to_upd_ok = list(zip(lst_src[part_len:], lst_dst[:part_len])) assert len(to_upd) == len(to_upd_ok) for item in to_upd_ok: assert item in to_upd @pytest.mark.parametrize("no_time", [True, False], ids=["date", "dateTime"]) def test_filter_events_by_date(no_time: bool): msk = timezone("Europe/Moscow") now = utc.localize(datetime.datetime.utcnow()) msk_now = msk.normalize(now.astimezone(msk)) part_len = 5 if no_time: duration = datetime.date(1, 1, 2) - datetime.date(1, 1, 1) else: duration = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1) date_cmp: DateDateTime = msk_now + (duration * part_len) if no_time: date_cmp = datetime.date(date_cmp.year, date_cmp.month, date_cmp.day) events = gen_events(1, 1 + (part_len * 2), msk_now, no_time) shuffle(events) events_pending = CalendarSync._filter_events_by_date(events, date_cmp, operator.ge) events_past = CalendarSync._filter_events_by_date(events, date_cmp, operator.lt) assert len(events_pending) == 1 + part_len assert len(events_past) == part_len - 1 for event in events_pending: assert get_start_date(event) >= date_cmp for event in events_past: assert get_start_date(event) < date_cmp def test_filter_events_to_update(): msk = timezone("Europe/Moscow") now = utc.localize(datetime.datetime.utcnow()) msk_now = msk.normalize(now.astimezone(msk)) one_hour = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1) date_upd = msk_now + (one_hour * 5) count = 10 events_old = gen_events(1, 1 + count, msk_now) events_new = gen_events(1, 1 + count, date_upd) sync1 = CalendarSync(None, None) sync1.to_update = list(zip(events_new, events_old)) sync1._filter_events_to_update() sync2 = CalendarSync(None, None) sync2.to_update = list(zip(events_old, events_new)) sync2._filter_events_to_update() assert len(sync1.to_update) == count assert sync2.to_update == [] def test_filter_events_no_updated(): """ test filtering events that not have 'updated' field such events should always pass the filter """ now = datetime.datetime.utcnow() yesterday = now - datetime.timedelta(days=-1) count = 10 events_old = gen_events(1, 1 + count, now) events_new = gen_events(1, 1 + count, now) # 1/2 updated=yesterday, 1/2 no updated field i = 0 for event in events_new: if 0 == i % 2: event["updated"] = yesterday.isoformat() + "Z" else: del event["updated"] i += 1 sync = CalendarSync(None, None) sync.to_update = list(zip(events_old, events_new)) sync._filter_events_to_update() assert len(sync.to_update) == count // 2