Merge branch 'develop'

2025-11-04 04:38:26 +00:00 · 2018-04-17 23:10:30 +03:00
parent 53115e5e34 c759d0b696
commit e16a4828e4
4 changed files with 226 additions and 263 deletions
--- a/gcal_sync/sync.py
+++ b/gcal_sync/sync.py
@@ -77,13 +77,23 @@ class CalendarSync():
        """

        def filter_by_date(event):
+            date_cmp = date
            event_start = event['start']
            event_date = None
+            compare_dates = False
+
            if 'date' in event_start:
                event_date = event_start['date']
-            if 'dateTime' in event_start:
+                compare_dates = True
+            elif 'dateTime' in event_start:
                event_date = event_start['dateTime']
-            return op(dateutil.parser.parse(event_date), date)
+            
+            event_date = dateutil.parser.parse(event_date)
+            if compare_dates:
+                date_cmp = datetime.date(date.year, date.month, date.day)
+                event_date = datetime.date(event_date.year, event_date.month, event_date.day)
+
+            return op(event_date, date_cmp)

        return list(filter(filter_by_date, events))

--- a/tests/test_converter.py
+++ b/tests/test_converter.py
@@ -1,154 +1,93 @@
-import unittest
+import pytest
+
 from gcal_sync import CalendarConverter

-ics_empty = """BEGIN:VCALENDAR
-VERSION:2.0
-PRODID:-//test//test//ES
-CALSCALE:GREGORIAN
-METHOD:PUBLISH
-END:VCALENDAR
-"""
-
-ics_empty_event = """BEGIN:VCALENDAR
-BEGIN:VEVENT
-END:VEVENT
-END:VCALENDAR
-"""
-
-ics_event_no_end = """BEGIN:VCALENDAR
-BEGIN:VEVENT
-UID:uisgtr8tre93wewe0yr8wqy@test.com
+uid = "UID:uisgtr8tre93wewe0yr8wqy@test.com"
+only_start_date = uid + """
 DTSTART;VALUE=DATE:20180215
-END:VEVENT
-END:VCALENDAR
 """
-
-ics_event_date_val = """BEGIN:VCALENDAR
-BEGIN:VEVENT
-UID:uisgtr8tre93wewe0yr8wqy@test.com
-DTSTART;VALUE=DATE:20180215
+date_val = only_start_date + """
 DTEND;VALUE=DATE:20180217
-END:VEVENT
-END:VCALENDAR
 """
-
-ics_event_datetime_utc_val = """BEGIN:VCALENDAR
-BEGIN:VEVENT
-UID:uisgtr8tre93wewe0yr8wqy@test.com
+date_duration = only_start_date + """
+DURATION:P2D
+"""
+datetime_utc_val = uid + """
 DTSTART;VALUE=DATE-TIME:20180319T092001Z
-DTEND:20180319T102001Z
-END:VEVENT
-END:VCALENDAR
+DTEND:20180321T102501Z
 """
-
-ics_event_date_duration = """BEGIN:VCALENDAR
-BEGIN:VEVENT
-UID:uisgtr8tre93wewe0yr8wqy@test.com
-DTSTART;VALUE=DATE:20180215
-DURATION:P3D
-END:VEVENT
-END:VCALENDAR
-"""
-
-ics_event_datetime_utc_duration = """BEGIN:VCALENDAR
-BEGIN:VEVENT
-UID:uisgtr8tre93wewe0yr8wqy@test.com
+datetime_utc_duration = uid + """
 DTSTART;VALUE=DATE-TIME:20180319T092001Z
 DURATION:P2DT1H5M
-END:VEVENT
-END:VCALENDAR
 """
-
-ics_event_created_updated = """BEGIN:VCALENDAR
-BEGIN:VEVENT
-UID:uisgtr8tre93wewe0yr8wqy@test.com
-DTSTART:20180215
-DTEND:20180217
+created_updated = date_val + """
 CREATED:20180320T071155Z
 LAST-MODIFIED:20180326T120235Z
-END:VEVENT
-END:VCALENDAR
 """

-class TestCalendarConverter(unittest.TestCase):
-    def test_empty_calendar(self):
-      converter = CalendarConverter()
-      converter.loads(ics_empty)
-      evnts = converter.events_to_gcal()
-      self.assertEqual(len(evnts), 0)
-    
-    def test_empty_event(self):
-      converter = CalendarConverter()
-      converter.loads(ics_empty_event)
-      with self.assertRaises(KeyError):
-        converter.events_to_gcal()
-    
-    def test_event_no_end(self):
-      converter = CalendarConverter()
-      converter.loads(ics_event_no_end)
-      with self.assertRaises(ValueError):
+
+def ics_test_cal(content):
+    return "BEGIN:VCALENDAR\r\n{}END:VCALENDAR\r\n".format(content)
+
+
+def ics_test_event(content):
+    return ics_test_cal("BEGIN:VEVENT\r\n{}END:VEVENT\r\n".format(content))
+
+
+def test_empty_calendar():
+    converter = CalendarConverter()
+    converter.loads(ics_test_cal(""))
+    evnts = converter.events_to_gcal()
+    assert evnts == []
+
+
+def test_empty_event():
+    converter = CalendarConverter()
+    converter.loads(ics_test_event(""))
+    with pytest.raises(KeyError):
        converter.events_to_gcal()

-    def test_event_date_values(self):
-      converter = CalendarConverter()
-      converter.loads(ics_event_date_val)
-      events = converter.events_to_gcal()
-      self.assertEqual(len(events), 1)
-      event = events[0]
-      self.assertEqual(event['start'], {
-        'date': '2018-02-15'
-      })
-      self.assertEqual(event['end'], {
-        'date': '2018-02-17'
-      })

-    def test_event_datetime_utc_values(self):
-      converter = CalendarConverter()
-      converter.loads(ics_event_datetime_utc_val)
-      events = converter.events_to_gcal()
-      self.assertEqual(len(events), 1)
-      event = events[0]
-      self.assertEqual(event['start'], {
-        'dateTime': '2018-03-19T09:20:01.000001Z'
-      })
-      self.assertEqual(event['end'], {
-        'dateTime': '2018-03-19T10:20:01.000001Z'
-      })
-      
-    def test_event_date_duration(self):
-      converter = CalendarConverter()
-      converter.loads(ics_event_date_duration)
-      events = converter.events_to_gcal()
-      self.assertEqual(len(events), 1)
-      event = events[0]
-      self.assertEqual(event['start'], {
-        'date': '2018-02-15'
-      })
-      self.assertEqual(event['end'], {
-        'date': '2018-02-18'
-      })
+def test_event_no_end():
+    converter = CalendarConverter()
+    converter.loads(ics_test_event(only_start_date))
+    with pytest.raises(ValueError):
+        converter.events_to_gcal()

-    def test_event_datetime_utc_duration(self):
-      converter = CalendarConverter()
-      converter.loads(ics_event_datetime_utc_duration)
-      events = converter.events_to_gcal()
-      self.assertEqual(len(events), 1)
-      event = events[0]
-      self.assertEqual(event['start'], {
-        'dateTime': '2018-03-19T09:20:01.000001Z'
-      })
-      self.assertEqual(event['end'], {
-        'dateTime': '2018-03-21T10:25:01.000001Z'
-      })

-    def test_event_created_updated(self):
-      converter = CalendarConverter()
-      converter.loads(ics_event_created_updated)
-      events = converter.events_to_gcal()
-      self.assertEqual(len(events), 1)
-      event = events[0]
-      self.assertEqual(event['created'], '2018-03-20T07:11:55.000001Z')
-      self.assertEqual(event['updated'], '2018-03-26T12:02:35.000001Z')
+@pytest.fixture(params=[
+    ("date", ics_test_event(date_val), '2018-02-15', '2018-02-17'),
+    ("date", ics_test_event(date_duration), '2018-02-15', '2018-02-17'),
+    ("dateTime", ics_test_event(datetime_utc_val),
+     '2018-03-19T09:20:01.000001Z', '2018-03-21T10:25:01.000001Z'),
+    ("dateTime", ics_test_event(datetime_utc_duration), '2018-03-19T09:20:01.000001Z', '2018-03-21T10:25:01.000001Z')],
+    ids=['date values', 'date duration',
+         'datetime utc values', 'datetime utc duration']
+)
+def param_events_start_end(request):
+    return request.param

-if __name__ == '__main__':
-    unittest.main()
+
+def test_event_start_end(param_events_start_end):
+    (date_type, ics_str, start, end) = param_events_start_end
+    converter = CalendarConverter()
+    converter.loads(ics_str)
+    events = converter.events_to_gcal()
+    assert len(events) == 1
+    event = events[0]
+    assert event['start'] == {
+        date_type: start
+    }
+    assert event['end'] == {
+        date_type: end
+    }
+
+
+def test_event_created_updated():
+    converter = CalendarConverter()
+    converter.loads(ics_test_event(created_updated))
+    events = converter.events_to_gcal()
+    assert len(events) == 1
+    event = events[0]
+    assert event['created'] == '2018-03-20T07:11:55.000001Z'
+    assert event['updated'] == '2018-03-26T12:02:35.000001Z'
--- a/tests/test_imports.py
+++ b/tests/test_imports.py
@@ -1,14 +0,0 @@
-import unittest
-
-class TestImports(unittest.TestCase):
-    def test_imports(self):
-      from gcal_sync import (
-        CalendarConverter, 
-        EventConverter, 
-        GoogleCalendarService, 
-        GoogleCalendar, 
-        CalendarSync
-      )
-
-if __name__ == '__main__':
-    unittest.main()
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -1,149 +1,177 @@
 import datetime
 import hashlib
 import operator
-import unittest
 from copy import deepcopy
 from random import shuffle

 import dateutil.parser
+import pytest
 from pytz import timezone, utc

 from gcal_sync import CalendarSync


-class TestCalendarSync(unittest.TestCase):
-    @staticmethod
-    def sha1(string):
-        if isinstance(string, str):
-            string = string.encode('utf8')
-        h = hashlib.sha1()
-        h.update(string)
-        return h.hexdigest()
+def sha1(string):
+    if isinstance(string, str):
+        string = string.encode('utf8')
+    h = hashlib.sha1()
+    h.update(string)
+    return h.hexdigest()

-    @staticmethod
-    def gen_events(start, stop, start_time):
-        one_hour = datetime.datetime(
+
+def gen_events(start, stop, start_time, no_time=False):
+    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"
+        suff = ''
+    else:
+        start_time = utc.normalize(
+            start_time.astimezone(utc)).replace(tzinfo=None)
+        duration = datetime.datetime(
            1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
-        result = []
-        for i in range(start, stop):
-            event_start = start_time + (one_hour * i)
-            event_end = event_start + one_hour
-            updated = utc.normalize(
-                event_start.astimezone(utc)).replace(tzinfo=None)
-            result.append({
-                'summary': 'test event __ {}'.format(i),
-                'location': 'la la la {}'.format(i),
-                'description': 'test TEST -- test event {}'.format(i),
-                'start': {
-                    'dateTime': event_start.isoformat()
-                },
-                'end': {
-                    'dateTime': event_end.isoformat(),
-                },
-                "iCalUID": "{}@test.com".format(TestCalendarSync.sha1("test - event {}".format(i))),
-                "updated": updated.isoformat() + 'Z',
-                "created": updated.isoformat() + 'Z'
-            })
-        return result
+        date_key = "dateTime"
+        suff = 'Z'

-    @staticmethod
-    def gen_list_to_compare(start, stop):
-        result = []
-        for i in range(start, stop):
-            result.append({'iCalUID': 'test{:06d}'.format(i)})
-        return result
+    result = []
+    for i in range(start, stop):
+        event_start = start_time + (duration * i)
+        event_end = event_start + duration

-    @staticmethod
-    def get_start_date(event):
-        event_start = event['start']
-        start_date = None
-        if 'date' in event_start:
-            start_date = event_start['date']
-        if 'dateTime' in event_start:
-            start_date = event_start['dateTime']
-        return dateutil.parser.parse(start_date)
+        updated = event_start
+        if no_time:
+            updated = datetime.datetime(
+                updated.year, updated.month, updated.day, 0, 0, 0, 1, tzinfo=utc)

-    def test_compare(self):
-        part_len = 20
-        # [1..2n]
-        lst_src = TestCalendarSync.gen_list_to_compare(1, 1 + part_len * 2)
-        # [n..3n]
-        lst_dst = TestCalendarSync.gen_list_to_compare(1 + part_len, 1 + part_len * 3)
+        event = {
+            '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'
+        }
+        event['start'] = {date_key: event_start.isoformat() + suff}
+        event['end'] = {date_key: event_end.isoformat() + suff}
+        result.append(event)
+    return result

-        lst_src_rnd = deepcopy(lst_src)
-        lst_dst_rnd = deepcopy(lst_dst)

-        shuffle(lst_src_rnd)
-        shuffle(lst_dst_rnd)
+def gen_list_to_compare(start, stop):
+    result = []
+    for i in range(start, stop):
+        result.append({'iCalUID': 'test{:06d}'.format(i)})
+    return result

-        to_ins, to_upd, to_del = CalendarSync._events_list_compare(
-            lst_src_rnd, lst_dst_rnd)

-        self.assertEqual(len(to_ins), part_len)
-        self.assertEqual(len(to_upd), part_len)
-        self.assertEqual(len(to_del), part_len)
+def get_start_date(event):
+    event_start = event['start']
+    start_date = None
+    is_date = False
+    if 'date' in event_start:
+        start_date = event_start['date']
+        is_date = True
+    if 'dateTime' in event_start:
+        start_date = event_start['dateTime']

-        self.assertEqual(
-            sorted(to_ins, key=lambda x: x['iCalUID']), lst_src[:part_len])
-        self.assertEqual(
-            sorted(to_del, key=lambda x: x['iCalUID']), lst_dst[part_len:])
+    result = dateutil.parser.parse(start_date)
+    if is_date:
+        result = datetime.date(result.year, result.month, result.day)

-        to_upd_ok = list(zip(lst_src[part_len:], lst_dst[:part_len]))
-        self.assertEqual(len(to_upd), len(to_upd_ok))
-        for item in to_upd_ok:
-            self.assertIn(item, to_upd)
+    return result

-    def test_filter_events_by_date(self):
-        msk = timezone('Europe/Moscow')
-        now = utc.localize(datetime.datetime.utcnow())
-        msk_now = msk.normalize(now.astimezone(msk))

-        one_hour = datetime.datetime(
+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):
+    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 = msk_now + (one_hour * 5)

-        events = TestCalendarSync.gen_events(1, 11, msk_now)
-        shuffle(events)
+    date_cmp = msk_now + (duration * part_len)

-        events_pending = CalendarSync._filter_events_by_date(
-            events, date_cmp, operator.ge)
-        events_past = CalendarSync._filter_events_by_date(
-            events, date_cmp, operator.lt)
+    if no_time:
+        date_cmp = datetime.date(
+            date_cmp.year, date_cmp.month, date_cmp.day)

-        self.assertEqual(len(events_pending), 6)
-        self.assertEqual(len(events_past), 4)
+    events = gen_events(
+        1, 1 + (part_len * 2), msk_now, no_time)
+    shuffle(events)

-        for event in events_pending:
-            self.assertGreaterEqual(
-                TestCalendarSync.get_start_date(event), date_cmp)
+    events_pending = CalendarSync._filter_events_by_date(
+        events, date_cmp, operator.ge)
+    events_past = CalendarSync._filter_events_by_date(
+        events, date_cmp, operator.lt)

-        for event in events_past:
-            self.assertLess(TestCalendarSync.get_start_date(event), date_cmp)
+    assert len(events_pending) == 1 + part_len
+    assert len(events_past) == part_len - 1

-    def test_filter_events_to_update(self):
-        msk = timezone('Europe/Moscow')
-        now = utc.localize(datetime.datetime.utcnow())
-        msk_now = msk.normalize(now.astimezone(msk))
+    for event in events_pending:
+        assert get_start_date(event) >= date_cmp

-        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 = TestCalendarSync.gen_events(1, 1 + count, msk_now)
-        events_new = TestCalendarSync.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()
-
-        self.assertEqual(len(sync1.to_update), count)
-        self.assertEqual(len(sync2.to_update), 0)
+    for event in events_past:
+        assert get_start_date(event) < date_cmp


-if __name__ == '__main__':
-    unittest.main()
+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 == []