Skip to content

Commit

Permalink
general: make time.tz.via_location user config lazy, move tests to my…
Browse files Browse the repository at this point in the history
….tests package

also gets rid of the problematic reset_modules thingie
  • Loading branch information
karlicoss committed Aug 26, 2024
1 parent 9912555 commit 8c29c15
Show file tree
Hide file tree
Showing 15 changed files with 268 additions and 233 deletions.
2 changes: 1 addition & 1 deletion my/calendar/holidays.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ def _calendar():
# todo switch to using time.tz.main once _get_tz stabilizes?
from ..time.tz import via_location as LTZ
# 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
zone = tz.zone; assert zone is not None
code = zone_to_countrycode(zone)
Expand Down
3 changes: 2 additions & 1 deletion my/core/compat.py
Original file line number Diff line number Diff line change
Expand Up @@ -124,9 +124,10 @@ def test_fromisoformat() -> None:


if sys.version_info[:2] >= (3, 10):
from types import NoneType
from types import NoneType, TypeAlias
else:
NoneType = type(None)
from typing_extension import TypeAlias


if sys.version_info[:2] >= (3, 11):
Expand Down
4 changes: 3 additions & 1 deletion my/location/fallback/via_ip.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,14 +29,16 @@ class ip_config(location.via_ip):

from my.core import make_logger
from my.core.compat import bisect_left
from my.ip.all import ips
from my.location.common import Location
from my.location.fallback.common import FallbackLocation, DateExact, _datetime_timestamp

logger = make_logger(__name__, level="warning")


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()
for ip in ips():
lat, lon = ip.latlon
Expand Down
9 changes: 9 additions & 0 deletions my/tests/calendar.py
Original file line number Diff line number Diff line change
@@ -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
16 changes: 0 additions & 16 deletions my/tests/common.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
import os
from pathlib import Path
import re
import sys

import pytest

Expand All @@ -13,20 +11,6 @@
)


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:
d = Path(__file__).absolute().parent.parent.parent / 'testdata'
assert d.exists(), d
Expand Down
2 changes: 2 additions & 0 deletions my/tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
import pytest


# I guess makes sense by default
@pytest.fixture(autouse=True)
def without_cachew():
from my.core.cachew import disabled_cachew

with disabled_cachew():
yield
52 changes: 31 additions & 21 deletions tests/location_fallback.py → my/tests/location/fallback.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,32 +2,23 @@
To test my.location.fallback_location.all
"""

from datetime import datetime, timedelta, timezone
from typing import Iterator
from datetime import datetime, timezone, timedelta

import pytest
from more_itertools import ilen

import my.ip.all as ip_module
from my.ip.common import IP
from my.location.fallback import via_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
ip_module.ips = data
from ..shared_tz_config import config # autoused fixture

from my.location.fallback import via_ip

# 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

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(via_ip.fallback_locations())
assert ilen(via_ip.fallback_locations()) == 5
Expand All @@ -47,7 +38,9 @@ def test_ip_fallback() -> None:
assert len(est) == 1

# 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

# on 2/1/2020, threes one IP if before 16:30
Expand Down Expand Up @@ -75,8 +68,8 @@ def test_ip_fallback() -> None:
#
# redefine fallback_estimators to prevent possible namespace packages the user
# may have installed from having side effects testing this
from my.location.fallback import all
from my.location.fallback import via_home
from my.location.fallback import all, via_home

def _fe() -> Iterator[all.LocationEstimator]:
yield via_ip.estimate_location
yield via_home.estimate_location
Expand All @@ -88,6 +81,7 @@ def _fe() -> Iterator[all.LocationEstimator]:
#
# just passing via_ip should give one IP
from my.location.fallback.common import _iter_estimate_from

raw_est = list(_iter_estimate_from(use_dt, (via_ip.estimate_location,)))
assert len(raw_est) == 1
assert raw_est[0].datasource == "via_ip"
Expand All @@ -110,7 +104,7 @@ def _fe() -> Iterator[all.LocationEstimator]:
# should have used the IP from via_ip since it was more accurate
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))
assert loc.datasource == "via_home"

Expand All @@ -121,5 +115,21 @@ def _fe() -> Iterator[all.LocationEstimator]:
assert (loc.lat, loc.lon) != (bulgaria.lat, bulgaria.lon)


# re-use prepare fixture for overriding config from shared_config.py
from .tz import prepare
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))


@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
75 changes: 35 additions & 40 deletions tests/shared_config.py → my/tests/shared_tz_config.py
Original file line number Diff line number Diff line change
@@ -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 typing import Any, NamedTuple
import my.time.tz.via_location as LTZ
import pytest
from more_itertools import one

from my.core.cfg import tmp_config

class SharedConfig(NamedTuple):
google: Any
location: Any
time: Any

@pytest.fixture(autouse=True)
def config(tmp_path: Path):
# TODO could just pick a part of shared config? not sure
_takeout_path = _prepare_takeouts_dir(tmp_path)

def _prepare_google_config(tmp_path: Path):
from my.tests.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())

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 google:
takeout_path = _takeout_path

class location:
home_accuracy = 30_000
# fmt: off
home = (
# supports ISO strings
('2005-12-04' , (42.697842, 23.325973)), # Bulgaria, Sofia
Expand All @@ -50,16 +29,32 @@ class location:
# check tz handling..
(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
class via_ip:
accuracy = 15_000
class gpslogger:
pass

class time:
class tz:
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 changes: 107 additions & 0 deletions my/tests/tz.py
Original file line number Diff line number Diff line change
@@ -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')
Loading

0 comments on commit 8c29c15

Please sign in to comment.