general: make time.tz.via_location user config lazy, move tests to my.tests package
also gets rid of the problematic reset_modules thingie
This commit is contained in:
parent
270080bd56
commit
a5643206a0
15 changed files with 269 additions and 233 deletions
|
@ -19,7 +19,7 @@ def _calendar():
|
||||||
# todo switch to using time.tz.main once _get_tz stabilizes?
|
# todo switch to using time.tz.main once _get_tz stabilizes?
|
||||||
from ..time.tz import via_location as LTZ
|
from ..time.tz import via_location as LTZ
|
||||||
# TODO would be nice to do it dynamically depending on the past timezones...
|
# TODO would be nice to do it dynamically depending on the past timezones...
|
||||||
tz = LTZ._get_tz(datetime.now())
|
tz = LTZ.get_tz(datetime.now())
|
||||||
assert tz is not None
|
assert tz is not None
|
||||||
zone = tz.zone; assert zone is not None
|
zone = tz.zone; assert zone is not None
|
||||||
code = zone_to_countrycode(zone)
|
code = zone_to_countrycode(zone)
|
||||||
|
|
|
@ -125,8 +125,10 @@ def test_fromisoformat() -> None:
|
||||||
|
|
||||||
if sys.version_info[:2] >= (3, 10):
|
if sys.version_info[:2] >= (3, 10):
|
||||||
from types import NoneType
|
from types import NoneType
|
||||||
|
from typing import TypeAlias
|
||||||
else:
|
else:
|
||||||
NoneType = type(None)
|
NoneType = type(None)
|
||||||
|
from typing_extensions import TypeAlias
|
||||||
|
|
||||||
|
|
||||||
if sys.version_info[:2] >= (3, 11):
|
if sys.version_info[:2] >= (3, 11):
|
||||||
|
|
|
@ -29,7 +29,6 @@ from typing import Iterator, List
|
||||||
|
|
||||||
from my.core import make_logger
|
from my.core import make_logger
|
||||||
from my.core.compat import bisect_left
|
from my.core.compat import bisect_left
|
||||||
from my.ip.all import ips
|
|
||||||
from my.location.common import Location
|
from my.location.common import Location
|
||||||
from my.location.fallback.common import FallbackLocation, DateExact, _datetime_timestamp
|
from my.location.fallback.common import FallbackLocation, DateExact, _datetime_timestamp
|
||||||
|
|
||||||
|
@ -37,6 +36,9 @@ logger = make_logger(__name__, level="warning")
|
||||||
|
|
||||||
|
|
||||||
def fallback_locations() -> Iterator[FallbackLocation]:
|
def fallback_locations() -> Iterator[FallbackLocation]:
|
||||||
|
# prefer late import since ips get overridden in tests
|
||||||
|
from my.ip.all import ips
|
||||||
|
|
||||||
dur = config.for_duration.total_seconds()
|
dur = config.for_duration.total_seconds()
|
||||||
for ip in ips():
|
for ip in ips():
|
||||||
lat, lon = ip.latlon
|
lat, lon = ip.latlon
|
||||||
|
|
9
my/tests/calendar.py
Normal file
9
my/tests/calendar.py
Normal file
|
@ -0,0 +1,9 @@
|
||||||
|
from my.calendar.holidays import is_holiday
|
||||||
|
|
||||||
|
from .shared_tz_config import config # autoused fixture
|
||||||
|
|
||||||
|
|
||||||
|
def test_is_holiday() -> None:
|
||||||
|
assert is_holiday('20190101')
|
||||||
|
assert not is_holiday('20180601')
|
||||||
|
assert is_holiday('20200906') # national holiday in Bulgaria
|
|
@ -1,7 +1,5 @@
|
||||||
import os
|
import os
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
import re
|
|
||||||
import sys
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
@ -13,20 +11,6 @@ skip_if_not_karlicoss = pytest.mark.skipif(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def reset_modules() -> None:
|
|
||||||
'''
|
|
||||||
A hack to 'unload' HPI modules, otherwise some modules might cache the config
|
|
||||||
TODO: a bit crap, need a better way..
|
|
||||||
'''
|
|
||||||
to_unload = [m for m in sys.modules if re.match(r'my[.]?', m)]
|
|
||||||
for m in to_unload:
|
|
||||||
if 'my.pdfs' in m:
|
|
||||||
# temporary hack -- since my.pdfs migrated to a 'lazy' config, this isn't necessary anymore
|
|
||||||
# but if we reset module anyway, it confuses the ProcessPool inside my.pdfs
|
|
||||||
continue
|
|
||||||
del sys.modules[m]
|
|
||||||
|
|
||||||
|
|
||||||
def testdata() -> Path:
|
def testdata() -> Path:
|
||||||
d = Path(__file__).absolute().parent.parent.parent / 'testdata'
|
d = Path(__file__).absolute().parent.parent.parent / 'testdata'
|
||||||
assert d.exists(), d
|
assert d.exists(), d
|
||||||
|
|
|
@ -1,8 +1,10 @@
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
# I guess makes sense by default
|
# I guess makes sense by default
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture(autouse=True)
|
||||||
def without_cachew():
|
def without_cachew():
|
||||||
from my.core.cachew import disabled_cachew
|
from my.core.cachew import disabled_cachew
|
||||||
|
|
||||||
with disabled_cachew():
|
with disabled_cachew():
|
||||||
yield
|
yield
|
||||||
|
|
|
@ -2,32 +2,23 @@
|
||||||
To test my.location.fallback_location.all
|
To test my.location.fallback_location.all
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
from typing import Iterator
|
from typing import Iterator
|
||||||
from datetime import datetime, timezone, timedelta
|
|
||||||
|
|
||||||
|
import pytest
|
||||||
from more_itertools import ilen
|
from more_itertools import ilen
|
||||||
|
|
||||||
from my.ip.common import IP
|
|
||||||
|
|
||||||
def data() -> Iterator[IP]:
|
|
||||||
# random IP addresses
|
|
||||||
yield IP(addr="67.98.113.0", dt=datetime(2020, 1, 1, 12, 0, 0, tzinfo=timezone.utc))
|
|
||||||
yield IP(addr="67.98.112.0", dt=datetime(2020, 1, 15, 12, 0, 0, tzinfo=timezone.utc))
|
|
||||||
yield IP(addr="59.40.113.87", dt=datetime(2020, 2, 1, 12, 0, 0, tzinfo=timezone.utc))
|
|
||||||
yield IP(addr="59.40.139.87", dt=datetime(2020, 2, 1, 16, 0, 0, tzinfo=timezone.utc))
|
|
||||||
yield IP(addr="161.235.192.228", dt=datetime(2020, 3, 1, 12, 0, 0, tzinfo=timezone.utc))
|
|
||||||
|
|
||||||
# redefine the my.ip.all function using data for testing
|
|
||||||
import my.ip.all as ip_module
|
import my.ip.all as ip_module
|
||||||
ip_module.ips = data
|
from my.ip.common import IP
|
||||||
|
|
||||||
from my.location.fallback import via_ip
|
from my.location.fallback import via_ip
|
||||||
|
|
||||||
|
from ..shared_tz_config import config # autoused fixture
|
||||||
|
|
||||||
|
|
||||||
# these are all tests for the bisect algorithm defined in via_ip.py
|
# these are all tests for the bisect algorithm defined in via_ip.py
|
||||||
# to make sure we can correctly find IPs that are within the 'for_duration' of a given datetime
|
# to make sure we can correctly find IPs that are within the 'for_duration' of a given datetime
|
||||||
|
|
||||||
def test_ip_fallback() -> None:
|
def test_ip_fallback() -> None:
|
||||||
# make sure that the data override works
|
# precondition, make sure that the data override works
|
||||||
assert ilen(ip_module.ips()) == ilen(data())
|
assert ilen(ip_module.ips()) == ilen(data())
|
||||||
assert ilen(ip_module.ips()) == ilen(via_ip.fallback_locations())
|
assert ilen(ip_module.ips()) == ilen(via_ip.fallback_locations())
|
||||||
assert ilen(via_ip.fallback_locations()) == 5
|
assert ilen(via_ip.fallback_locations()) == 5
|
||||||
|
@ -47,7 +38,9 @@ def test_ip_fallback() -> None:
|
||||||
assert len(est) == 1
|
assert len(est) == 1
|
||||||
|
|
||||||
# right after the 'for_duration' for an IP
|
# right after the 'for_duration' for an IP
|
||||||
est = list(via_ip.estimate_location(datetime(2020, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + via_ip.config.for_duration + timedelta(seconds=1)))
|
est = list(
|
||||||
|
via_ip.estimate_location(datetime(2020, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + via_ip.config.for_duration + timedelta(seconds=1))
|
||||||
|
)
|
||||||
assert len(est) == 0
|
assert len(est) == 0
|
||||||
|
|
||||||
# on 2/1/2020, threes one IP if before 16:30
|
# on 2/1/2020, threes one IP if before 16:30
|
||||||
|
@ -75,8 +68,8 @@ def test_ip_fallback() -> None:
|
||||||
#
|
#
|
||||||
# redefine fallback_estimators to prevent possible namespace packages the user
|
# redefine fallback_estimators to prevent possible namespace packages the user
|
||||||
# may have installed from having side effects testing this
|
# may have installed from having side effects testing this
|
||||||
from my.location.fallback import all
|
from my.location.fallback import all, via_home
|
||||||
from my.location.fallback import via_home
|
|
||||||
def _fe() -> Iterator[all.LocationEstimator]:
|
def _fe() -> Iterator[all.LocationEstimator]:
|
||||||
yield via_ip.estimate_location
|
yield via_ip.estimate_location
|
||||||
yield via_home.estimate_location
|
yield via_home.estimate_location
|
||||||
|
@ -88,6 +81,7 @@ def test_ip_fallback() -> None:
|
||||||
#
|
#
|
||||||
# just passing via_ip should give one IP
|
# just passing via_ip should give one IP
|
||||||
from my.location.fallback.common import _iter_estimate_from
|
from my.location.fallback.common import _iter_estimate_from
|
||||||
|
|
||||||
raw_est = list(_iter_estimate_from(use_dt, (via_ip.estimate_location,)))
|
raw_est = list(_iter_estimate_from(use_dt, (via_ip.estimate_location,)))
|
||||||
assert len(raw_est) == 1
|
assert len(raw_est) == 1
|
||||||
assert raw_est[0].datasource == "via_ip"
|
assert raw_est[0].datasource == "via_ip"
|
||||||
|
@ -110,7 +104,7 @@ def test_ip_fallback() -> None:
|
||||||
# should have used the IP from via_ip since it was more accurate
|
# should have used the IP from via_ip since it was more accurate
|
||||||
assert all_est.datasource == "via_ip"
|
assert all_est.datasource == "via_ip"
|
||||||
|
|
||||||
# test that a home defined in shared_config.py is used if no IP is found
|
# test that a home defined in shared_tz_config.py is used if no IP is found
|
||||||
loc = all.estimate_location(datetime(2021, 1, 1, 12, 30, 0, tzinfo=timezone.utc))
|
loc = all.estimate_location(datetime(2021, 1, 1, 12, 30, 0, tzinfo=timezone.utc))
|
||||||
assert loc.datasource == "via_home"
|
assert loc.datasource == "via_home"
|
||||||
|
|
||||||
|
@ -121,5 +115,21 @@ def test_ip_fallback() -> None:
|
||||||
assert (loc.lat, loc.lon) != (bulgaria.lat, bulgaria.lon)
|
assert (loc.lat, loc.lon) != (bulgaria.lat, bulgaria.lon)
|
||||||
|
|
||||||
|
|
||||||
# re-use prepare fixture for overriding config from shared_config.py
|
def data() -> Iterator[IP]:
|
||||||
from .tz import prepare
|
# random IP addresses
|
||||||
|
yield IP(addr="67.98.113.0", dt=datetime(2020, 1, 1, 12, 0, 0, tzinfo=timezone.utc))
|
||||||
|
yield IP(addr="67.98.112.0", dt=datetime(2020, 1, 15, 12, 0, 0, tzinfo=timezone.utc))
|
||||||
|
yield IP(addr="59.40.113.87", dt=datetime(2020, 2, 1, 12, 0, 0, tzinfo=timezone.utc))
|
||||||
|
yield IP(addr="59.40.139.87", dt=datetime(2020, 2, 1, 16, 0, 0, tzinfo=timezone.utc))
|
||||||
|
yield IP(addr="161.235.192.228", dt=datetime(2020, 3, 1, 12, 0, 0, tzinfo=timezone.utc))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def prepare(config):
|
||||||
|
before = ip_module.ips
|
||||||
|
# redefine the my.ip.all function using data for testing
|
||||||
|
ip_module.ips = data
|
||||||
|
try:
|
||||||
|
yield
|
||||||
|
finally:
|
||||||
|
ip_module.ips = before
|
|
@ -1,47 +1,26 @@
|
||||||
# Defines some shared config for tests
|
"""
|
||||||
|
Helper to test various timezone/location dependent things
|
||||||
|
"""
|
||||||
|
|
||||||
from datetime import datetime, date, timezone
|
from datetime import date, datetime, timezone
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from typing import Any, NamedTuple
|
import pytest
|
||||||
import my.time.tz.via_location as LTZ
|
|
||||||
from more_itertools import one
|
from more_itertools import one
|
||||||
|
|
||||||
|
from my.core.cfg import tmp_config
|
||||||
class SharedConfig(NamedTuple):
|
|
||||||
google: Any
|
|
||||||
location: Any
|
|
||||||
time: Any
|
|
||||||
|
|
||||||
|
|
||||||
def _prepare_google_config(tmp_path: Path):
|
@pytest.fixture(autouse=True)
|
||||||
from my.tests.common import testdata
|
def config(tmp_path: Path):
|
||||||
try:
|
# TODO could just pick a part of shared config? not sure
|
||||||
track = one(testdata().rglob('italy-slovenia-2017-07-29.json'))
|
_takeout_path = _prepare_takeouts_dir(tmp_path)
|
||||||
except ValueError:
|
|
||||||
raise RuntimeError('testdata not found, setup git submodules?')
|
|
||||||
|
|
||||||
|
class google:
|
||||||
# todo ugh. unnecessary zipping, but at the moment takeout provider doesn't support plain dirs
|
takeout_path = _takeout_path
|
||||||
import zipfile
|
|
||||||
with zipfile.ZipFile(tmp_path / 'takeout.zip', 'w') as zf:
|
|
||||||
zf.writestr('Takeout/Location History/Location History.json', track.read_bytes())
|
|
||||||
|
|
||||||
class google_config:
|
|
||||||
takeout_path = tmp_path
|
|
||||||
return google_config
|
|
||||||
|
|
||||||
|
|
||||||
# pass tmp_path from pytest to this helper function
|
|
||||||
# see tests/tz.py as an example
|
|
||||||
def temp_config(temp_path: Path) -> Any:
|
|
||||||
from my.tests.common import reset_modules
|
|
||||||
reset_modules()
|
|
||||||
|
|
||||||
LTZ.config.fast = True
|
|
||||||
|
|
||||||
class location:
|
class location:
|
||||||
home_accuracy = 30_000
|
# fmt: off
|
||||||
home = (
|
home = (
|
||||||
# supports ISO strings
|
# supports ISO strings
|
||||||
('2005-12-04' , (42.697842, 23.325973)), # Bulgaria, Sofia
|
('2005-12-04' , (42.697842, 23.325973)), # Bulgaria, Sofia
|
||||||
|
@ -50,16 +29,32 @@ def temp_config(temp_path: Path) -> Any:
|
||||||
# check tz handling..
|
# check tz handling..
|
||||||
(datetime.fromtimestamp(1600000000, tz=timezone.utc), (55.7558 , 37.6173 )), # Moscow, Russia
|
(datetime.fromtimestamp(1600000000, tz=timezone.utc), (55.7558 , 37.6173 )), # Moscow, Russia
|
||||||
)
|
)
|
||||||
|
# fmt: on
|
||||||
# note: order doesn't matter, will be sorted in the data provider
|
# note: order doesn't matter, will be sorted in the data provider
|
||||||
class via_ip:
|
|
||||||
accuracy = 15_000
|
|
||||||
class gpslogger:
|
|
||||||
pass
|
|
||||||
|
|
||||||
class time:
|
class time:
|
||||||
class tz:
|
class tz:
|
||||||
class via_location:
|
class via_location:
|
||||||
pass # just rely on the defaults...
|
fast = True # some tests rely on it
|
||||||
|
|
||||||
|
with tmp_config() as cfg:
|
||||||
|
cfg.google = google
|
||||||
|
cfg.location = location
|
||||||
|
cfg.time = time
|
||||||
|
yield cfg
|
||||||
|
|
||||||
|
|
||||||
return SharedConfig(google=_prepare_google_config(temp_path), location=location, time=time)
|
def _prepare_takeouts_dir(tmp_path: Path) -> Path:
|
||||||
|
from .common import testdata
|
||||||
|
|
||||||
|
try:
|
||||||
|
track = one(testdata().rglob('italy-slovenia-2017-07-29.json'))
|
||||||
|
except ValueError:
|
||||||
|
raise RuntimeError('testdata not found, setup git submodules?')
|
||||||
|
|
||||||
|
# todo ugh. unnecessary zipping, but at the moment takeout provider doesn't support plain dirs
|
||||||
|
import zipfile
|
||||||
|
|
||||||
|
with zipfile.ZipFile(tmp_path / 'takeout.zip', 'w') as zf:
|
||||||
|
zf.writestr('Takeout/Location History/Location History.json', track.read_bytes())
|
||||||
|
return tmp_path
|
107
my/tests/tz.py
Normal file
107
my/tests/tz.py
Normal file
|
@ -0,0 +1,107 @@
|
||||||
|
import sys
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import pytz
|
||||||
|
|
||||||
|
import my.time.tz.main as tz_main
|
||||||
|
import my.time.tz.via_location as tz_via_location
|
||||||
|
from my.core import notnone
|
||||||
|
from my.core.compat import fromisoformat
|
||||||
|
|
||||||
|
from .shared_tz_config import config # autoused fixture
|
||||||
|
|
||||||
|
|
||||||
|
def getzone(dt: datetime) -> str:
|
||||||
|
tz = notnone(dt.tzinfo)
|
||||||
|
return getattr(tz, 'zone')
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize('fast', [False, True])
|
||||||
|
def test_iter_tzs(fast: bool, config) -> None:
|
||||||
|
# TODO hmm.. maybe need to make sure we start with empty config?
|
||||||
|
config.time.tz.via_location.fast = fast
|
||||||
|
|
||||||
|
ll = list(tz_via_location._iter_tzs())
|
||||||
|
zones = [x.zone for x in ll]
|
||||||
|
|
||||||
|
if fast:
|
||||||
|
assert zones == [
|
||||||
|
'Europe/Rome',
|
||||||
|
'Europe/Rome',
|
||||||
|
'Europe/Vienna',
|
||||||
|
'Europe/Vienna',
|
||||||
|
'Europe/Vienna',
|
||||||
|
]
|
||||||
|
else:
|
||||||
|
assert zones == [
|
||||||
|
'Europe/Rome',
|
||||||
|
'Europe/Rome',
|
||||||
|
'Europe/Ljubljana',
|
||||||
|
'Europe/Ljubljana',
|
||||||
|
'Europe/Ljubljana',
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_past() -> None:
|
||||||
|
"""
|
||||||
|
Should fallback to the 'home' location provider
|
||||||
|
"""
|
||||||
|
dt = fromisoformat('2000-01-01 12:34:45')
|
||||||
|
dt = tz_main.localize(dt)
|
||||||
|
assert getzone(dt) == 'America/New_York'
|
||||||
|
|
||||||
|
|
||||||
|
def test_future() -> None:
|
||||||
|
"""
|
||||||
|
For locations in the future should rely on 'home' location
|
||||||
|
"""
|
||||||
|
fut = datetime.now() + timedelta(days=100)
|
||||||
|
fut = tz_main.localize(fut)
|
||||||
|
assert getzone(fut) == 'Europe/Moscow'
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_tz(config) -> None:
|
||||||
|
# todo hmm, the way it's implemented at the moment, never returns None?
|
||||||
|
get_tz = tz_via_location.get_tz
|
||||||
|
|
||||||
|
# not present in the test data
|
||||||
|
tz = get_tz(fromisoformat('2020-01-01 10:00:00'))
|
||||||
|
assert notnone(tz).zone == 'Europe/Sofia'
|
||||||
|
|
||||||
|
tz = get_tz(fromisoformat('2017-08-01 11:00:00'))
|
||||||
|
assert notnone(tz).zone == 'Europe/Vienna'
|
||||||
|
|
||||||
|
tz = get_tz(fromisoformat('2017-07-30 10:00:00'))
|
||||||
|
assert notnone(tz).zone == 'Europe/Rome'
|
||||||
|
|
||||||
|
tz = get_tz(fromisoformat('2020-10-01 14:15:16'))
|
||||||
|
assert tz is not None
|
||||||
|
|
||||||
|
on_windows = sys.platform == 'win32'
|
||||||
|
if not on_windows:
|
||||||
|
tz = get_tz(datetime.min)
|
||||||
|
assert tz is not None
|
||||||
|
else:
|
||||||
|
# seems this fails because windows doesn't support same date ranges
|
||||||
|
# https://stackoverflow.com/a/41400321/
|
||||||
|
with pytest.raises(OSError):
|
||||||
|
get_tz(datetime.min)
|
||||||
|
|
||||||
|
|
||||||
|
def test_policies() -> None:
|
||||||
|
naive = fromisoformat('2017-07-30 10:00:00')
|
||||||
|
assert naive.tzinfo is None # just in case
|
||||||
|
|
||||||
|
# actual timezone at the time
|
||||||
|
assert getzone(tz_main.localize(naive)) == 'Europe/Rome'
|
||||||
|
|
||||||
|
z = pytz.timezone('America/New_York')
|
||||||
|
aware = z.localize(naive)
|
||||||
|
|
||||||
|
assert getzone(tz_main.localize(aware)) == 'America/New_York'
|
||||||
|
|
||||||
|
assert getzone(tz_main.localize(aware, policy='convert')) == 'Europe/Rome'
|
||||||
|
|
||||||
|
with pytest.raises(RuntimeError):
|
||||||
|
assert tz_main.localize(aware, policy='throw')
|
|
@ -1,52 +1,43 @@
|
||||||
'''
|
'''
|
||||||
Timezone data provider, guesses timezone based on location data (e.g. GPS)
|
Timezone data provider, guesses timezone based on location data (e.g. GPS)
|
||||||
'''
|
'''
|
||||||
|
|
||||||
REQUIRES = [
|
REQUIRES = [
|
||||||
# for determining timezone by coordinate
|
# for determining timezone by coordinate
|
||||||
'timezonefinder',
|
'timezonefinder',
|
||||||
]
|
]
|
||||||
|
|
||||||
|
import heapq
|
||||||
|
import os
|
||||||
from collections import Counter
|
from collections import Counter
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from datetime import date, datetime
|
from datetime import date, datetime
|
||||||
from functools import lru_cache
|
from functools import lru_cache
|
||||||
import heapq
|
|
||||||
from itertools import groupby
|
from itertools import groupby
|
||||||
import os
|
from typing import (
|
||||||
from typing import Iterator, Optional, Tuple, Any, List, Iterable, Set, Dict
|
TYPE_CHECKING,
|
||||||
|
Any,
|
||||||
|
Dict,
|
||||||
|
Iterable,
|
||||||
|
Iterator,
|
||||||
|
List,
|
||||||
|
Optional,
|
||||||
|
Protocol,
|
||||||
|
Set,
|
||||||
|
Tuple,
|
||||||
|
)
|
||||||
|
|
||||||
import pytz
|
import pytz
|
||||||
|
|
||||||
|
from my.core import Stats, datetime_aware, make_logger, stat
|
||||||
from my.core.cachew import mcachew
|
from my.core.cachew import mcachew
|
||||||
from my.core import make_logger, stat, Stats, datetime_aware
|
from my.core.compat import TypeAlias
|
||||||
from my.core.source import import_source
|
from my.core.source import import_source
|
||||||
from my.core.warnings import high
|
from my.core.warnings import high
|
||||||
|
|
||||||
from my.location.common import LatLon
|
from my.location.common import LatLon
|
||||||
|
|
||||||
|
|
||||||
## user might not have tz config section, so makes sense to be more defensive about it
|
class config(Protocol):
|
||||||
# todo might be useful to extract a helper for this
|
|
||||||
try:
|
|
||||||
from my.config import time
|
|
||||||
except ImportError as ie:
|
|
||||||
if ie.name != 'time':
|
|
||||||
raise ie
|
|
||||||
else:
|
|
||||||
try:
|
|
||||||
user_config = time.tz.via_location
|
|
||||||
except AttributeError as ae:
|
|
||||||
if not ("'tz'" in str(ae) or "'via_location'"):
|
|
||||||
raise ae
|
|
||||||
|
|
||||||
# deliberately dynamic to prevent confusing mypy
|
|
||||||
if 'user_config' not in globals():
|
|
||||||
globals()['user_config'] = object
|
|
||||||
##
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class config(user_config):
|
|
||||||
# less precise, but faster
|
# less precise, but faster
|
||||||
fast: bool = True
|
fast: bool = True
|
||||||
|
|
||||||
|
@ -62,6 +53,43 @@ class config(user_config):
|
||||||
_iter_tz_refresh_time: int = 6
|
_iter_tz_refresh_time: int = 6
|
||||||
|
|
||||||
|
|
||||||
|
def _get_user_config():
|
||||||
|
## user might not have tz config section, so makes sense to be more defensive about it
|
||||||
|
|
||||||
|
class empty_config: ...
|
||||||
|
|
||||||
|
try:
|
||||||
|
from my.config import time
|
||||||
|
except ImportError as ie:
|
||||||
|
if "'time'" not in str(ie):
|
||||||
|
raise ie
|
||||||
|
else:
|
||||||
|
return empty_config
|
||||||
|
|
||||||
|
try:
|
||||||
|
user_config = time.tz.via_location
|
||||||
|
except AttributeError as ae:
|
||||||
|
if not ("'tz'" in str(ae) or "'via_location'" in str(ae)):
|
||||||
|
raise ae
|
||||||
|
else:
|
||||||
|
return empty_config
|
||||||
|
|
||||||
|
return user_config
|
||||||
|
|
||||||
|
|
||||||
|
def make_config() -> config:
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
import my.config
|
||||||
|
|
||||||
|
user_config: TypeAlias = my.config.time.tz.via_location
|
||||||
|
else:
|
||||||
|
user_config = _get_user_config()
|
||||||
|
|
||||||
|
class combined_config(user_config, config): ...
|
||||||
|
|
||||||
|
return combined_config()
|
||||||
|
|
||||||
|
|
||||||
logger = make_logger(__name__)
|
logger = make_logger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@ -78,6 +106,7 @@ def _timezone_finder(fast: bool) -> Any:
|
||||||
# for backwards compatibility
|
# for backwards compatibility
|
||||||
def _locations() -> Iterator[Tuple[LatLon, datetime_aware]]:
|
def _locations() -> Iterator[Tuple[LatLon, datetime_aware]]:
|
||||||
try:
|
try:
|
||||||
|
raise RuntimeError
|
||||||
import my.location.all
|
import my.location.all
|
||||||
|
|
||||||
for loc in my.location.all.locations():
|
for loc in my.location.all.locations():
|
||||||
|
@ -140,13 +169,14 @@ def _find_tz_for_locs(finder: Any, locs: Iterable[Tuple[LatLon, datetime]]) -> I
|
||||||
# Note: this takes a while, as the upstream since _locations isn't sorted, so this
|
# Note: this takes a while, as the upstream since _locations isn't sorted, so this
|
||||||
# has to do an iterative sort of the entire my.locations.all list
|
# has to do an iterative sort of the entire my.locations.all list
|
||||||
def _iter_local_dates() -> Iterator[DayWithZone]:
|
def _iter_local_dates() -> Iterator[DayWithZone]:
|
||||||
finder = _timezone_finder(fast=config.fast) # rely on the default
|
cfg = make_config()
|
||||||
|
finder = _timezone_finder(fast=cfg.fast) # rely on the default
|
||||||
# pdt = None
|
# pdt = None
|
||||||
# TODO: warnings doesn't actually warn?
|
# TODO: warnings doesn't actually warn?
|
||||||
# warnings = []
|
# warnings = []
|
||||||
|
|
||||||
locs: Iterable[Tuple[LatLon, datetime]]
|
locs: Iterable[Tuple[LatLon, datetime]]
|
||||||
locs = _sorted_locations() if config.sort_locations else _locations()
|
locs = _sorted_locations() if cfg.sort_locations else _locations()
|
||||||
|
|
||||||
yield from _find_tz_for_locs(finder, locs)
|
yield from _find_tz_for_locs(finder, locs)
|
||||||
|
|
||||||
|
@ -158,11 +188,13 @@ def _iter_local_dates() -> Iterator[DayWithZone]:
|
||||||
def _iter_local_dates_fallback() -> Iterator[DayWithZone]:
|
def _iter_local_dates_fallback() -> Iterator[DayWithZone]:
|
||||||
from my.location.fallback.all import fallback_locations as flocs
|
from my.location.fallback.all import fallback_locations as flocs
|
||||||
|
|
||||||
|
cfg = make_config()
|
||||||
|
|
||||||
def _fallback_locations() -> Iterator[Tuple[LatLon, datetime]]:
|
def _fallback_locations() -> Iterator[Tuple[LatLon, datetime]]:
|
||||||
for loc in sorted(flocs(), key=lambda x: x.dt):
|
for loc in sorted(flocs(), key=lambda x: x.dt):
|
||||||
yield ((loc.lat, loc.lon), loc.dt)
|
yield ((loc.lat, loc.lon), loc.dt)
|
||||||
|
|
||||||
yield from _find_tz_for_locs(_timezone_finder(fast=config.fast), _fallback_locations())
|
yield from _find_tz_for_locs(_timezone_finder(fast=cfg.fast), _fallback_locations())
|
||||||
|
|
||||||
|
|
||||||
def most_common(lst: Iterator[DayWithZone]) -> DayWithZone:
|
def most_common(lst: Iterator[DayWithZone]) -> DayWithZone:
|
||||||
|
@ -180,7 +212,8 @@ def _iter_tz_depends_on() -> str:
|
||||||
2022-04-26_12
|
2022-04-26_12
|
||||||
2022-04-26_18
|
2022-04-26_18
|
||||||
"""
|
"""
|
||||||
mod = config._iter_tz_refresh_time
|
cfg = make_config()
|
||||||
|
mod = cfg._iter_tz_refresh_time
|
||||||
assert mod >= 1
|
assert mod >= 1
|
||||||
day = str(date.today())
|
day = str(date.today())
|
||||||
hr = datetime.now().hour
|
hr = datetime.now().hour
|
||||||
|
@ -293,5 +326,13 @@ def stats(quick: bool = False) -> Stats:
|
||||||
return stat(localized_years)
|
return stat(localized_years)
|
||||||
|
|
||||||
|
|
||||||
# deprecated -- still used in some other modules so need to keep
|
## deprecated -- keeping for now as might be used in other modules?
|
||||||
_get_tz = get_tz
|
if not TYPE_CHECKING:
|
||||||
|
from my.core.compat import deprecated
|
||||||
|
|
||||||
|
@deprecated('use get_tz function instead')
|
||||||
|
def _get_tz(*args, **kwargs):
|
||||||
|
return get_tz(*args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
|
##
|
||||||
|
|
|
@ -1,19 +0,0 @@
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from my.calendar.holidays import is_holiday
|
|
||||||
|
|
||||||
|
|
||||||
def test() -> None:
|
|
||||||
assert is_holiday('20190101')
|
|
||||||
assert not is_holiday('20180601')
|
|
||||||
assert is_holiday('20200906') # national holiday in Bulgaria
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
|
||||||
def prepare(tmp_path: Path):
|
|
||||||
from . import tz
|
|
||||||
# todo meh. fixtures can't be called directly?
|
|
||||||
orig = tz.prepare.__wrapped__ # type: ignore
|
|
||||||
yield from orig(tmp_path)
|
|
|
@ -1,10 +1,9 @@
|
||||||
from my.tests.common import skip_if_not_karlicoss as pytestmark
|
from my.tests.common import skip_if_not_karlicoss as pytestmark
|
||||||
|
|
||||||
|
def test() -> None:
|
||||||
from my import orgmode
|
from my import orgmode
|
||||||
from my.core.orgmode import collect
|
from my.core.orgmode import collect
|
||||||
|
|
||||||
|
|
||||||
def test() -> None:
|
|
||||||
# meh
|
# meh
|
||||||
results = list(orgmode.query().collect_all(lambda n: [n] if 'python' in n.tags else []))
|
results = list(orgmode.query().collect_all(lambda n: [n] if 'python' in n.tags else []))
|
||||||
assert len(results) > 5
|
assert len(results) > 5
|
||||||
|
|
|
@ -1,4 +1,4 @@
|
||||||
#!/usr/bin/env python3
|
from my.tests.common import skip_if_not_karlicoss as pytestmark
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from itertools import islice
|
from itertools import islice
|
||||||
import pytz
|
import pytz
|
||||||
|
|
95
tests/tz.py
95
tests/tz.py
|
@ -1,95 +0,0 @@
|
||||||
import sys
|
|
||||||
from datetime import datetime, timedelta
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
import pytz
|
|
||||||
|
|
||||||
from my.core.error import notnone
|
|
||||||
|
|
||||||
import my.time.tz.main as TZ
|
|
||||||
import my.time.tz.via_location as LTZ
|
|
||||||
|
|
||||||
|
|
||||||
def test_iter_tzs() -> None:
|
|
||||||
ll = list(LTZ._iter_tzs())
|
|
||||||
assert len(ll) > 3
|
|
||||||
|
|
||||||
|
|
||||||
def test_past() -> None:
|
|
||||||
# should fallback to the home location provider
|
|
||||||
dt = D('20000101 12:34:45')
|
|
||||||
dt = TZ.localize(dt)
|
|
||||||
tz = dt.tzinfo
|
|
||||||
assert tz is not None
|
|
||||||
assert getattr(tz, 'zone') == 'America/New_York'
|
|
||||||
|
|
||||||
|
|
||||||
def test_future() -> None:
|
|
||||||
fut = datetime.now() + timedelta(days=100)
|
|
||||||
# shouldn't crash at least
|
|
||||||
assert TZ.localize(fut) is not None
|
|
||||||
|
|
||||||
|
|
||||||
def test_tz() -> None:
|
|
||||||
# todo hmm, the way it's implemented at the moment, never returns None?
|
|
||||||
|
|
||||||
# not present in the test data
|
|
||||||
tz = LTZ._get_tz(D('20200101 10:00:00'))
|
|
||||||
assert notnone(tz).zone == 'Europe/Sofia'
|
|
||||||
|
|
||||||
tz = LTZ._get_tz(D('20170801 11:00:00'))
|
|
||||||
assert notnone(tz).zone == 'Europe/Vienna'
|
|
||||||
|
|
||||||
tz = LTZ._get_tz(D('20170730 10:00:00'))
|
|
||||||
assert notnone(tz).zone == 'Europe/Rome'
|
|
||||||
|
|
||||||
tz = LTZ._get_tz(D('20201001 14:15:16'))
|
|
||||||
assert tz is not None
|
|
||||||
|
|
||||||
on_windows = sys.platform == 'win32'
|
|
||||||
if not on_windows:
|
|
||||||
tz = LTZ._get_tz(datetime.min)
|
|
||||||
assert tz is not None
|
|
||||||
else:
|
|
||||||
# seems this fails because windows doesn't support same date ranges
|
|
||||||
# https://stackoverflow.com/a/41400321/
|
|
||||||
with pytest.raises(OSError):
|
|
||||||
LTZ._get_tz(datetime.min)
|
|
||||||
|
|
||||||
|
|
||||||
def test_policies() -> None:
|
|
||||||
getzone = lambda dt: getattr(dt.tzinfo, 'zone')
|
|
||||||
|
|
||||||
naive = D('20170730 10:00:00')
|
|
||||||
# actual timezone at the time
|
|
||||||
assert getzone(TZ.localize(naive)) == 'Europe/Rome'
|
|
||||||
|
|
||||||
z = pytz.timezone('America/New_York')
|
|
||||||
aware = z.localize(naive)
|
|
||||||
|
|
||||||
assert getzone(TZ.localize(aware)) == 'America/New_York'
|
|
||||||
|
|
||||||
assert getzone(TZ.localize(aware, policy='convert')) == 'Europe/Rome'
|
|
||||||
|
|
||||||
|
|
||||||
with pytest.raises(RuntimeError):
|
|
||||||
assert TZ.localize(aware, policy='throw')
|
|
||||||
|
|
||||||
|
|
||||||
def D(dstr: str) -> datetime:
|
|
||||||
return datetime.strptime(dstr, '%Y%m%d %H:%M:%S')
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
|
||||||
def prepare(tmp_path: Path):
|
|
||||||
from .shared_config import temp_config
|
|
||||||
conf = temp_config(tmp_path)
|
|
||||||
|
|
||||||
import my.core.cfg as C
|
|
||||||
with C.tmp_config() as config:
|
|
||||||
config.google = conf.google
|
|
||||||
config.time = conf.time
|
|
||||||
config.location = conf.location
|
|
||||||
yield
|
|
1
tox.ini
1
tox.ini
|
@ -88,7 +88,6 @@ commands =
|
||||||
|
|
||||||
{envpython} -m pytest tests \
|
{envpython} -m pytest tests \
|
||||||
# ignore some tests which might take a while to run on ci..
|
# ignore some tests which might take a while to run on ci..
|
||||||
--ignore tests/takeout.py \
|
|
||||||
--ignore tests/extra/polar.py
|
--ignore tests/extra/polar.py
|
||||||
{posargs}
|
{posargs}
|
||||||
|
|
||||||
|
|
Loading…
Add table
Reference in a new issue