b4tman
/
sync_ics2gcal
mirror of https://github.com/b4tman/sync_ics2gcal
1
0
Fork 0
Code Issues Releases Activity

Merge pull request #98 from b4tman/mypy 

type checking with mypy
This commit is contained in:
Dmitry Belyaev 2022-06-04 22:57:22 +03:00 committed by GitHub
parent 6ba8106f3e c9b38cd29e
commit a18be3d079
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 1367 additions and 1010 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
.github/workflows/pythonpackage.yml vendored
View File

@ -17,7 +17,7 @@ jobs:
strategy:
max-parallel: 4
matrix:
python-version: ['3.7', '3.8', '3.9', '3.10']
python-version: ['3.8', '3.9', '3.10']
steps:
- uses: actions/checkout@v3
@ -39,3 +39,18 @@ jobs:
poetry run flake8 sync_ics2gcal --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest
run: poetry run pytest -v
- name: Check type annotations with mypy
run: |
mkdir mypy_report
poetry run mypy --pretty --html-report mypy_report/ .
- name: Check type annotations with mypy strict mode (not failing)
run: |
poetry run mypy --strict --pretty . || true
- name: Upload mypy report
uses: actions/upload-artifact@v2
with:
name: mypy_report
path: mypy_report/

41
.github/workflows/reviewdog.yml vendored Normal file
View File

@ -0,0 +1,41 @@
name: reviewdog
on:
pull_request:
branches:
- master
- develop
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: 3.x
- name: Upgrade pip
run: python -m pip install --upgrade pip
- name: Install Poetry
uses: snok/install-poetry@v1
- name: Install deps
run: poetry install
- name: setup mypy
run: |
mkdir tmp_bin/
echo "#!/bin/sh" > tmp_bin/mypy
echo "poetry run mypy \$@" >> tmp_bin/mypy
chmod +x tmp_bin/mypy
echo "$(pwd)/tmp_bin" >> $GITHUB_PATH
- uses: tsuyoshicho/action-mypy@v3
with:
reporter: github-pr-review
level: warning

5
.gitignore vendored
View File

@ -4,8 +4,13 @@ service-account.json
my-test*.ics
.vscode/
.idea/
.venv/
.pytest_cache/
.mypy_cache/
/dist/
/*.egg-info/
/build/
/.eggs/
venv/
mypy_report/
tmp_bin/

1837
poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

17
pyproject.toml
View File

@ -11,14 +11,13 @@ keywords = ["icalendar", "sync", "google", "calendar"]
classifiers = [
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
'Programming Language :: Python :: 3.10',
]
[tool.poetry.dependencies]
python = "^3.7"
python = "^3.8"
google-auth = "2.6.6"
google-api-python-client = "2.49.0"
icalendar = "4.0.9"
@ -30,6 +29,11 @@ fire = "0.4.0"
pytest = "^7.1.2"
flake8 = "^4.0.1"
black = "^22.3.0"
mypy = ">=0.960"
types-python-dateutil = "^2.8.17"
types-pytz = ">=2021.3.8"
types-PyYAML = "^6.0.8"
lxml = "^4.9.0"
[tool.poetry.scripts]
sync-ics2gcal = "sync_ics2gcal.sync_calendar:main"
@ -38,3 +42,12 @@ manage-ics2gcal = "sync_ics2gcal.manage_calendars:main"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[[tool.mypy.overrides]]
module = [
'icalendar',
'google.*',
'googleapiclient',
'fire'
]
ignore_missing_imports = true

35
sync_ics2gcal/__init__.py
View File

@ -6,6 +6,39 @@ from .gcal import (
EventData,
EventList,
EventTuple,
EventDataKey,
EventDateOrDateTime,
EventDate,
EventDateTime,
EventsSearchResults,
ACLRule,
ACLScope,
CalendarData,
BatchRequestCallback,
)
from .sync import CalendarSync
from .sync import CalendarSync, ComparedEvents
__all__ = [
"ical",
"gcal",
"sync",
"CalendarConverter",
"EventConverter",
"DateDateTime",
"GoogleCalendarService",
"GoogleCalendar",
"EventData",
"EventList",
"EventTuple",
"EventDataKey",
"EventDateOrDateTime",
"EventDate",
"EventDateTime",
"EventsSearchResults",
"ACLRule",
"ACLScope",
"CalendarData",
"CalendarSync",
"ComparedEvents",
]

213
sync_ics2gcal/gcal.py
View File

@ -1,17 +1,97 @@
import logging
from datetime import datetime
from typing import List, Dict, Any, Callable, Tuple, Optional, Union
from typing import (
List,
Dict,
Any,
Callable,
Tuple,
Optional,
Union,
TypedDict,
Literal,
NamedTuple,
)
import google.auth
from google.oauth2 import service_account
from googleapiclient import discovery
from pytz import utc
EventData = Dict[str, Union[str, "EventData", None]]
class EventDate(TypedDict, total=False):
date: str
timeZone: str
class EventDateTime(TypedDict, total=False):
dateTime: str
timeZone: str
EventDateOrDateTime = Union[EventDate, EventDateTime]
class ACLScope(TypedDict, total=False):
type: str
value: str
class ACLRule(TypedDict, total=False):
scope: ACLScope
role: str
class CalendarData(TypedDict, total=False):
id: str
summary: str
description: str
timeZone: str
class EventData(TypedDict, total=False):
id: str
summary: str
description: str
start: EventDateOrDateTime
end: EventDateOrDateTime
iCalUID: str
location: str
status: str
created: str
updated: str
sequence: int
transparency: str
visibility: str
EventDataKey = Union[
Literal["id"],
Literal["summary"],
Literal["description"],
Literal["start"],
Literal["end"],
Literal["iCalUID"],
Literal["location"],
Literal["status"],
Literal["created"],
Literal["updated"],
Literal["sequence"],
Literal["transparency"],
Literal["visibility"],
]
EventList = List[EventData]
EventTuple = Tuple[EventData, EventData]
class EventsSearchResults(NamedTuple):
exists: List[EventTuple]
new: List[EventData]
BatchRequestCallback = Callable[[str, Any, Optional[Exception]], None]
class GoogleCalendarService:
"""class for make google calendar service Resource
@ -20,7 +100,7 @@ class GoogleCalendarService:
"""
@staticmethod
def default():
def default() -> discovery.Resource:
"""make service Resource from default credentials (authorize)
( https://developers.google.com/identity/protocols/application-default-credentials )
( https://googleapis.dev/python/google-auth/latest/reference/google.auth.html#google.auth.default )
@ -34,7 +114,7 @@ class GoogleCalendarService:
return service
@staticmethod
def from_srv_acc_file(service_account_file: str):
def from_srv_acc_file(service_account_file: str) -> discovery.Resource:
"""make service Resource from service account filename (authorize)"""
scopes = ["https://www.googleapis.com/auth/calendar"]
@ -48,7 +128,7 @@ class GoogleCalendarService:
return service
@staticmethod
def from_config(config: Optional[Dict[str, Optional[str]]] = None):
def from_config(config: Optional[Dict[str, str]] = None) -> discovery.Resource:
"""make service Resource from config dict
Arguments:
@ -60,7 +140,8 @@ class GoogleCalendarService:
"""
if config is not None and "service_account" in config:
service = GoogleCalendarService.from_srv_acc_file(config["service_account"])
service_account_filename: str = config["service_account"]
service = GoogleCalendarService.from_srv_acc_file(service_account_filename)
else:
service = GoogleCalendarService.default()
return service
@ -76,7 +157,7 @@ def select_event_key(event: EventData) -> Optional[str]:
key name or None if no key found
"""
key = None
key: Optional[str] = None
if "iCalUID" in event:
key = "iCalUID"
elif "id" in event:
@ -91,9 +172,11 @@ class GoogleCalendar:
def __init__(self, service: discovery.Resource, calendar_id: Optional[str]):
self.service: discovery.Resource = service
self.calendar_id: str = calendar_id
self.calendar_id: str = str(calendar_id)
def _make_request_callback(self, action: str, events_by_req: EventList) -> Callable:
def _make_request_callback(
self, action: str, events_by_req: EventList
) -> BatchRequestCallback:
"""make callback for log result of batch request
Arguments:
@ -104,9 +187,12 @@ class GoogleCalendar:
callback function
"""
def callback(request_id, response, exception):
event = events_by_req[int(request_id)]
key = select_event_key(event)
def callback(
request_id: str, response: Any, exception: Optional[Exception]
) -> None:
event: EventData = events_by_req[int(request_id)]
event_key: Optional[str] = select_event_key(event)
key: str = event_key if event_key is not None else ""
if exception is not None:
self.logger.error(
@ -117,7 +203,7 @@ class GoogleCalendar:
str(exception),
)
else:
resp_key = select_event_key(response)
resp_key: Optional[str] = select_event_key(response)
if resp_key is not None:
event = response
key = resp_key
@ -127,10 +213,10 @@ class GoogleCalendar:
def list_events_from(self, start: datetime) -> EventList:
"""list events from calendar, where start date >= start"""
fields = "nextPageToken,items(id,iCalUID,updated)"
events = []
page_token = None
time_min = (
fields: str = "nextPageToken,items(id,iCalUID,updated)"
events: EventList = []
page_token: Optional[str] = None
time_min: str = (
utc.normalize(start.astimezone(utc)).replace(tzinfo=None).isoformat() + "Z"
)
while True:
@ -153,25 +239,27 @@ class GoogleCalendar:
self.logger.info("%d events listed", len(events))
return events
def find_exists(self, events: List) -> Tuple[List[EventTuple], EventList]:
def find_exists(self, events: EventList) -> EventsSearchResults:
"""find existing events from list, by 'iCalUID' field
Arguments:
events {list} -- list of events
Returns:
tuple -- (events_exist, events_not_found)
EventsSearchResults -- (events_exist, events_not_found)
events_exist - list of tuples: (new_event, exists_event)
"""
fields = "items(id,iCalUID,updated)"
events_by_req = []
exists = []
not_found = []
fields: str = "items(id,iCalUID,updated)"
events_by_req: EventList = []
exists: List[EventTuple] = []
not_found: EventList = []
def list_callback(request_id, response, exception):
found = False
cur_event = events_by_req[int(request_id)]
def list_callback(
request_id: str, response: Any, exception: Optional[Exception]
) -> None:
found: bool = False
cur_event: EventData = events_by_req[int(request_id)]
if exception is None:
found = [] != response["items"]
else:
@ -186,7 +274,7 @@ class GoogleCalendar:
not_found.append(events_by_req[int(request_id)])
batch = self.service.new_batch_http_request(callback=list_callback)
i = 0
i: int = 0
for event in events:
events_by_req.append(event)
batch.add(
@ -201,21 +289,21 @@ class GoogleCalendar:
i += 1
batch.execute()
self.logger.info("%d events exists, %d not found", len(exists), len(not_found))
return exists, not_found
return EventsSearchResults(exists, not_found)
def insert_events(self, events: EventList):
def insert_events(self, events: EventList) -> None:
"""insert list of events
Arguments:
events - events list
"""
fields = "id"
events_by_req = []
fields: str = "id"
events_by_req: EventList = []
insert_callback = self._make_request_callback("insert", events_by_req)
batch = self.service.new_batch_http_request(callback=insert_callback)
i = 0
i: int = 0
for event in events:
events_by_req.append(event)
batch.add(
@ -227,19 +315,19 @@ class GoogleCalendar:
i += 1
batch.execute()
def patch_events(self, event_tuples: List[EventTuple]):
def patch_events(self, event_tuples: List[EventTuple]) -> None:
"""patch (update) events
Arguments:
event_tuples -- list of tuples: (new_event, exists_event)
"""
fields = "id"
events_by_req = []
fields: str = "id"
events_by_req: EventList = []
patch_callback = self._make_request_callback("patch", events_by_req)
batch = self.service.new_batch_http_request(callback=patch_callback)
i = 0
i: int = 0
for event_new, event_old in event_tuples:
if "id" not in event_old:
continue
@ -254,19 +342,19 @@ class GoogleCalendar:
i += 1
batch.execute()
def update_events(self, event_tuples: List[EventTuple]):
def update_events(self, event_tuples: List[EventTuple]) -> None:
"""update events
Arguments:
event_tuples -- list of tuples: (new_event, exists_event)
"""
fields = "id"
events_by_req = []
fields: str = "id"
events_by_req: EventList = []
update_callback = self._make_request_callback("update", events_by_req)
batch = self.service.new_batch_http_request(callback=update_callback)
i = 0
i: int = 0
for event_new, event_old in event_tuples:
if "id" not in event_old:
continue
@ -283,18 +371,18 @@ class GoogleCalendar:
i += 1
batch.execute()
def delete_events(self, events: EventList):
def delete_events(self, events: EventList) -> None:
"""delete events
Arguments:
events -- list of events
"""
events_by_req = []
events_by_req: EventList = []
delete_callback = self._make_request_callback("delete", events_by_req)
batch = self.service.new_batch_http_request(callback=delete_callback)
i = 0
i: int = 0
for event in events:
events_by_req.append(event)
batch.add(
@ -319,7 +407,7 @@ class GoogleCalendar:
calendar Resource
"""
calendar = {"summary": summary}
calendar = CalendarData(summary=summary)
if time_zone is not None:
calendar["timeZone"] = time_zone
@ -327,42 +415,27 @@ class GoogleCalendar:
self.calendar_id = created_calendar["id"]
return created_calendar
def delete(self):
def delete(self) -> None:
"""delete calendar"""
self.service.calendars().delete(calendarId=self.calendar_id).execute()
def make_public(self):
def make_public(self) -> None:
"""make calendar public"""
rule_public = {
"scope": {
"type": "default",
},
"role": "reader",
}
return (
self.service.acl()
.insert(calendarId=self.calendar_id, body=rule_public)
.execute()
)
rule_public = ACLRule(scope=ACLScope(type="default"), role="reader")
self.service.acl().insert(
calendarId=self.calendar_id, body=rule_public
).execute()
def add_owner(self, email: str):
def add_owner(self, email: str) -> None:
"""add calendar owner by email
Arguments:
email -- email to add
"""
rule_owner = {
"scope": {
"type": "user",
"value": email,
},
"role": "owner",
}
return (
self.service.acl()
.insert(calendarId=self.calendar_id, body=rule_owner)
.execute()
)
rule_owner = ACLRule(scope=ACLScope(type="user", value=email), role="owner")
self.service.acl().insert(
calendarId=self.calendar_id, body=rule_owner
).execute()

54
sync_ics2gcal/ical.py
View File

@ -1,11 +1,18 @@
import datetime
import logging
from typing import Union, Dict, Callable, Optional
from typing import Union, Dict, Callable, Optional, Mapping, TypedDict
from icalendar import Calendar, Event
from pytz import utc
from .gcal import EventData, EventList
from .gcal import (
EventData,
EventList,
EventDateOrDateTime,
EventDateTime,
EventDate,
EventDataKey,
)
DateDateTime = Union[datetime.date, datetime.datetime]
@ -28,7 +35,7 @@ def format_datetime_utc(value: DateDateTime) -> str:
def gcal_date_or_datetime(
value: DateDateTime, check_value: Optional[DateDateTime] = None
) -> Dict[str, str]:
) -> EventDateOrDateTime:
"""date or datetime to gcal (start or end dict)
Arguments:
@ -42,18 +49,18 @@ def gcal_date_or_datetime(
if check_value is None:
check_value = value
result: Dict[str, str] = {}
result: EventDateOrDateTime
if isinstance(check_value, datetime.datetime):
result["dateTime"] = format_datetime_utc(value)
result = EventDateTime(dateTime=format_datetime_utc(value))
else:
if isinstance(check_value, datetime.date):
if isinstance(value, datetime.datetime):
value = datetime.date(value.year, value.month, value.day)
result["date"] = value.isoformat()
result = EventDate(date=value.isoformat())
return result
class EventConverter(Event):
class EventConverter(Event): # type: ignore
"""Convert icalendar event to google calendar resource
( https://developers.google.com/calendar/v3/reference/events#resource-representations )
"""
@ -68,7 +75,7 @@ class EventConverter(Event):
string value
"""
return self.decoded(prop).decode(encoding="utf-8")
return str(self.decoded(prop).decode(encoding="utf-8"))
def _datetime_str_prop(self, prop: str) -> str:
"""utc datetime as string from property
@ -82,7 +89,7 @@ class EventConverter(Event):
return format_datetime_utc(self.decoded(prop))
def _gcal_start(self) -> Dict[str, str]:
def _gcal_start(self) -> EventDateOrDateTime:
"""event start dict from icalendar event
Raises:
@ -95,7 +102,7 @@ class EventConverter(Event):
value = self.decoded("DTSTART")
return gcal_date_or_datetime(value)
def _gcal_end(self) -> Dict[str, str]:
def _gcal_end(self) -> EventDateOrDateTime:
"""event end dict from icalendar event
Raises:
@ -104,7 +111,7 @@ class EventConverter(Event):
dict
"""
result: Dict[str, str]
result: EventDateOrDateTime
if "DTEND" in self:
value = self.decoded("DTEND")
result = gcal_date_or_datetime(value)
@ -121,10 +128,10 @@ class EventConverter(Event):
def _put_to_gcal(
self,
gcal_event: EventData,
prop: str,
prop: EventDataKey,
func: Callable[[str], str],
ics_prop: Optional[str] = None,
):
) -> None:
"""get property from ical event if existed, and put to gcal event
Arguments:
@ -139,18 +146,18 @@ class EventConverter(Event):
if ics_prop in self:
gcal_event[prop] = func(ics_prop)
def to_gcal(self) -> EventData:
def convert(self) -> EventData:
"""Convert
Returns:
dict - google calendar#event resource
"""
event = {
"iCalUID": self._str_prop("UID"),
"start": self._gcal_start(),
"end": self._gcal_end(),
}
event: EventData = EventData(
iCalUID=self._str_prop("UID"),
start=self._gcal_start(),
end=self._gcal_end(),
)
self._put_to_gcal(event, "summary", self._str_prop)
self._put_to_gcal(event, "description", self._str_prop)
@ -172,22 +179,23 @@ class CalendarConverter:
def __init__(self, calendar: Optional[Calendar] = None):
self.calendar: Optional[Calendar] = calendar
def load(self, filename: str):
def load(self, filename: str) -> None:
"""load calendar from ics file"""
with open(filename, "r", encoding="utf-8") as f:
self.calendar = Calendar.from_ical(f.read())
self.logger.info("%s loaded", filename)
def loads(self, string: str):
def loads(self, string: str) -> None:
"""load calendar from ics string"""
self.calendar = Calendar.from_ical(string)
def events_to_gcal(self) -> EventList:
"""Convert events to google calendar resources"""
ics_events = self.calendar.walk(name="VEVENT")
calendar: Calendar = self.calendar
ics_events = calendar.walk(name="VEVENT")
self.logger.info("%d events read", len(ics_events))
result = list(map(lambda event: EventConverter(event).to_gcal(), ics_events))
result = list(map(lambda event: EventConverter(event).convert(), ics_events))
self.logger.info("%d events converted", len(result))
return result

6
sync_ics2gcal/manage_calendars.py
View File

@ -8,7 +8,7 @@ from . import GoogleCalendar, GoogleCalendarService
def load_config(filename: str) -> Optional[Dict[str, Any]]:
result = None
result: Optional[Dict[str, Any]] = None
try:
with open(filename, "r", encoding="utf-8") as f:
result = yaml.safe_load(f)
@ -21,7 +21,7 @@ def load_config(filename: str) -> Optional[Dict[str, Any]]:
class PropertyCommands:
"""get/set google calendar properties"""
def __init__(self, _service):
def __init__(self, _service: Any) -> None:
self._service = _service
def get(self, calendar_id: str, property_name: str) -> None:
@ -146,7 +146,7 @@ class Commands:
print("{}: {}".format(summary, calendar_id))
def main():
def main() -> None:
fire.Fire(Commands, name="manage-ics2gcal")

54
sync_ics2gcal/sync.py
View File

@ -1,15 +1,31 @@
import datetime
import logging
import operator
from typing import List, Dict, Set, Tuple, Union, Callable
from typing import List, Dict, Set, Tuple, Union, Callable, NamedTuple
import dateutil.parser
from pytz import utc
from .gcal import GoogleCalendar, EventData, EventList, EventTuple
from .gcal import (
GoogleCalendar,
EventData,
EventList,
EventTuple,
EventDataKey,
EventDateOrDateTime,
EventDate,
)
from .ical import CalendarConverter, DateDateTime
class ComparedEvents(NamedTuple):
"""Compared events"""
added: EventList
changed: List[EventTuple]
deleted: EventList
class CalendarSync:
"""class for synchronize calendar with Google"""
@ -24,8 +40,8 @@ class CalendarSync:
@staticmethod
def _events_list_compare(
items_src: EventList, items_dst: EventList, key: str = "iCalUID"
) -> Tuple[EventList, List[EventTuple], EventList]:
items_src: EventList, items_dst: EventList, key: EventDataKey = "iCalUID"
) -> ComparedEvents:
"""compare list of events by key
Arguments:
@ -34,13 +50,11 @@ class CalendarSync:
key {str} -- name of key to compare (default: {'iCalUID'})
Returns:
tuple -- (items_to_insert,
items_to_update,
items_to_delete)
ComparedEvents -- (added, changed, deleted)
"""
def get_key(item: EventData) -> str:
return item[key]
return str(item[key])
keys_src: Set[str] = set(map(get_key, items_src))
keys_dst: Set[str] = set(map(get_key, items_dst))
@ -49,21 +63,21 @@ class CalendarSync:
keys_to_update = keys_src & keys_dst
keys_to_delete = keys_dst - keys_src
def items_by_keys(items: EventList, key_name: str, keys: Set[str]) -> EventList:
return list(filter(lambda item: item[key_name] in keys, items))
def items_by_keys(items: EventList, keys: Set[str]) -> EventList:
return list(filter(lambda item: get_key(item) in keys, items))
items_to_insert = items_by_keys(items_src, key, keys_to_insert)
items_to_delete = items_by_keys(items_dst, key, keys_to_delete)
items_to_insert = items_by_keys(items_src, keys_to_insert)
items_to_delete = items_by_keys(items_dst, keys_to_delete)
to_upd_src = items_by_keys(items_src, key, keys_to_update)
to_upd_dst = items_by_keys(items_dst, key, keys_to_update)
to_upd_src = items_by_keys(items_src, keys_to_update)
to_upd_dst = items_by_keys(items_dst, keys_to_update)
to_upd_src.sort(key=get_key)
to_upd_dst.sort(key=get_key)
items_to_update = list(zip(to_upd_src, to_upd_dst))
return items_to_insert, items_to_update, items_to_delete
return ComparedEvents(items_to_insert, items_to_update, items_to_delete)
def _filter_events_to_update(self):
def _filter_events_to_update(self) -> None:
"""filter 'to_update' events by 'updated' datetime"""
def filter_updated(event_tuple: EventTuple) -> bool:
@ -95,17 +109,17 @@ class CalendarSync:
def filter_by_date(event: EventData) -> bool:
date_cmp = date
event_start: Dict[str, str] = event["start"]
event_start: EventDateOrDateTime = event["start"]
event_date: Union[DateDateTime, str, None] = None
compare_dates = False
if "date" in event_start:
event_date = event_start["date"]
event_date = event_start["date"] # type: ignore
compare_dates = True
elif "dateTime" in event_start:
event_date = event_start["dateTime"]
event_date = event_start["dateTime"] # type: ignore
event_date = dateutil.parser.parse(event_date)
event_date = dateutil.parser.parse(str(event_date))
if compare_dates:
date_cmp = datetime.date(date.year, date.month, date.day)
event_date = datetime.date(

4
sync_ics2gcal/sync_calendar.py
View File

@ -14,7 +14,7 @@ ConfigDate = Union[str, datetime.datetime]
def load_config() -> Dict[str, Any]:
with open("config.yml", "r", encoding="utf-8") as f:
result = yaml.safe_load(f)
return result
return result # type: ignore
def get_start_date(date: ConfigDate) -> datetime.datetime:
@ -27,7 +27,7 @@ def get_start_date(date: ConfigDate) -> datetime.datetime:
return result
def main():
def main() -> None:
config = load_config()
if "logging" in config:

16
tests/test_converter.py
View File

@ -1,5 +1,5 @@
import datetime
from typing import Tuple
from typing import Tuple, Any
import pytest
from pytz import timezone, utc
@ -57,21 +57,21 @@ def ics_test_event(content: str) -> str:
return ics_test_cal("BEGIN:VEVENT\r\n{}END:VEVENT\r\n".format(content))
def test_empty_calendar():
def test_empty_calendar() -> None:
converter = CalendarConverter()
converter.loads(ics_test_cal(""))
evnts = converter.events_to_gcal()
assert evnts == []
def test_empty_event():
def test_empty_event() -> None:
converter = CalendarConverter()
converter.loads(ics_test_event(""))
with pytest.raises(KeyError):
converter.events_to_gcal()
def test_event_no_end():
def test_event_no_end() -> None:
converter = CalendarConverter()
converter.loads(ics_test_event(only_start_date))
with pytest.raises(ValueError):
@ -102,11 +102,11 @@ def test_event_no_end():
"datetime utc duration",
],
)
def param_events_start_end(request):
def param_events_start_end(request: Any) -> Any:
return request.param
def test_event_start_end(param_events_start_end: Tuple[str, str, str, str]):
def test_event_start_end(param_events_start_end: Tuple[str, str, str, str]) -> None:
(date_type, ics_str, start, end) = param_events_start_end
converter = CalendarConverter()
converter.loads(ics_str)
@ -117,7 +117,7 @@ def test_event_start_end(param_events_start_end: Tuple[str, str, str, str]):
assert event["end"] == {date_type: end}
def test_event_created_updated():
def test_event_created_updated() -> None:
converter = CalendarConverter()
converter.loads(ics_test_event(created_updated))
events = converter.events_to_gcal()
@ -142,5 +142,5 @@ def test_event_created_updated():
],
ids=["utc", "with timezone", "date"],
)
def test_format_datetime_utc(value: datetime.datetime, expected_str: str):
def test_format_datetime_utc(value: datetime.datetime, expected_str: str) -> None:
assert format_datetime_utc(value) == expected_str

78
tests/test_sync.py
View File

@ -3,95 +3,93 @@ import hashlib
import operator
from copy import deepcopy
from random import shuffle
from typing import Union, List, Dict, Optional
from typing import Union, List, Dict, Optional, AnyStr
import dateutil.parser
import pytest
from pytz import timezone, utc
from sync_ics2gcal import CalendarSync
from sync_ics2gcal import CalendarSync, DateDateTime
from sync_ics2gcal.gcal import EventDateOrDateTime, EventData, EventList
def sha1(string: Union[str, bytes]) -> str:
if isinstance(string, str):
string = string.encode("utf8")
def sha1(s: AnyStr) -> str:
h = hashlib.sha1()
h.update(string)
h.update(str(s).encode("utf8") if isinstance(s, str) else s)
return h.hexdigest()
def gen_events(
start: int,
stop: int,
start_time: Union[datetime.datetime, datetime.date],
start_time: DateDateTime,
no_time: bool = False,
) -> List[Dict[str, Union[str, Dict[str, str]]]]:
) -> 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.timedelta = datetime.date(1, 1, 2) - datetime.date(1, 1, 1)
date_key: str = "date"
date_end: str = ""
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)
duration: datetime.timedelta = datetime.datetime(
1, 1, 1, 2
) - datetime.datetime(1, 1, 1, 1)
date_key: str = "dateTime"
date_end: str = "Z"
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: List[Dict[str, Union[str, Dict[str, str]]]] = []
result: EventList = []
for i in range(start, stop):
event_start = start_time + (duration * i)
event_end = event_start + duration
updated: Union[datetime.datetime, datetime.date] = event_start
updated: DateDateTime = event_start
if no_time:
updated = datetime.datetime(
updated.year, updated.month, updated.day, 0, 0, 0, 1, tzinfo=utc
)
event: Dict[str, Union[str, Dict[str, str]]] = {
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},
"end": {date_key: event_end.isoformat() + date_end},
"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]] = []
def gen_list_to_compare(start: int, stop: int) -> EventList:
result: EventList = []
for i in range(start, stop):
result.append({"iCalUID": "test{:06d}".format(i)})
return result
def get_start_date(
event: Dict[str, Union[str, Dict[str, str]]]
) -> Union[datetime.datetime, datetime.date]:
event_start: Dict[str, str] = event["start"]
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"]
start_date = event_start["date"] # type: ignore
is_date = True
if "dateTime" in event_start:
start_date = event_start["dateTime"]
start_date = event_start["dateTime"] # type: ignore
result = dateutil.parser.parse(start_date)
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
def test_compare() -> None:
part_len: int = 20
# [1..2n]
lst_src = gen_list_to_compare(1, 1 + part_len * 2)
# [n..3n]
@ -119,7 +117,7 @@ def test_compare():
@pytest.mark.parametrize("no_time", [True, False], ids=["date", "dateTime"])
def test_filter_events_by_date(no_time: bool):
def test_filter_events_by_date(no_time: bool) -> None:
msk = timezone("Europe/Moscow")
now = utc.localize(datetime.datetime.utcnow())
msk_now = msk.normalize(now.astimezone(msk))
@ -131,7 +129,7 @@ def test_filter_events_by_date(no_time: bool):
else:
duration = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
date_cmp = msk_now + (duration * part_len)
date_cmp: DateDateTime = msk_now + (duration * part_len)
if no_time:
date_cmp = datetime.date(date_cmp.year, date_cmp.month, date_cmp.day)
@ -152,7 +150,7 @@ def test_filter_events_by_date(no_time: bool):
assert get_start_date(event) < date_cmp
def test_filter_events_to_update():
def test_filter_events_to_update() -> None:
msk = timezone("Europe/Moscow")
now = utc.localize(datetime.datetime.utcnow())
msk_now = msk.normalize(now.astimezone(msk))
@ -164,11 +162,11 @@ def test_filter_events_to_update():
events_old = gen_events(1, 1 + count, msk_now)
events_new = gen_events(1, 1 + count, date_upd)
sync1 = CalendarSync(None, None)
sync1 = CalendarSync(None, None) # type: ignore
sync1.to_update = list(zip(events_new, events_old))
sync1._filter_events_to_update()
sync2 = CalendarSync(None, None)
sync2 = CalendarSync(None, None) # type: ignore
sync2.to_update = list(zip(events_old, events_new))
sync2._filter_events_to_update()
@ -176,7 +174,7 @@ def test_filter_events_to_update():
assert sync2.to_update == []
def test_filter_events_no_updated():
def test_filter_events_no_updated() -> None:
"""
test filtering events that not have 'updated' field
such events should always pass the filter
@ -197,7 +195,7 @@ def test_filter_events_no_updated():
del event["updated"]
i += 1
sync = CalendarSync(None, None)
sync = CalendarSync(None, None) # type: ignore
sync.to_update = list(zip(events_old, events_new))
sync._filter_events_to_update()
assert len(sync.to_update) == count // 2