shared config for tests,add fallback location test

my/location/all.py

@@ -32,6 +32,7 @@ def _gpslogger_locations() -> Iterator[Location]:
     yield from gpslogger.locations()
 
 
+# TODO: remove, user should use fallback.estimate_location or fallback.fallback_locations instead
 @import_source(module_name="my.location.via_ip")
 def _ip_locations() -> Iterator[Location]:
     from . import via_ip

my/location/fallback/all.py

@@ -1,7 +1,7 @@
 # TODO: add config here which passes kwargs to estimate_from (under_accuracy)
 # overwritable by passing the kwarg name here to the top-level estimate_location
 
-from typing import Iterator
+from typing import Iterator, Optional
 
 from my.core.source import import_source
 from my.location.fallback.common import (
@@ -12,8 +12,8 @@ from my.location.fallback.common import (
 )
 
 
-# can comment/uncomment sources here to enable/disable them
 def fallback_locations() -> Iterator[FallbackLocation]:
+    # can comment/uncomment sources here to enable/disable them
     yield from _ip_fallback_locations()
 
 
@@ -24,8 +24,8 @@ def fallback_estimators() -> Iterator[LocationEstimator]:
     yield _home_estimate
 
 
-def estimate_location(dt: DateExact) -> FallbackLocation:
+def estimate_location(dt: DateExact, first_match: bool=False, under_accuracy: Optional[int] = None) -> FallbackLocation:
-    loc = estimate_from(dt, estimators=list(fallback_estimators()))
+    loc = estimate_from(dt, estimators=list(fallback_estimators()), first_match=first_match, under_accuracy=under_accuracy)
     # should never happen if the user has home configured
     if loc is None:
         raise ValueError("Could not estimate location")

my/location/fallback/via_ip.py

@@ -86,8 +86,9 @@ def estimate_location(dt: DateExact) -> Iterator[FallbackLocation]:
         if start_time <= dt_ts <= end_time:
             # logger.debug(f"Found location for {dt}: {loc}")
             yield loc
-        if end_time > dt_ts:
+        # no more locations could possibly contain dt
-            # logger.debug(f"Passed end time: {end_time} > {dt_ts} ({datetime.fromtimestamp(end_time)} > {datetime.fromtimestamp(dt_ts)})")
+        if start_time > dt_ts:
+            # logger.debug(f"Passed start time: {end_time} > {dt_ts} ({datetime.fromtimestamp(end_time)} > {datetime.fromtimestamp(dt_ts)})")
             break
         idx += 1
 

tests/location.py

@@ -1,7 +1,5 @@
 from pathlib import Path
 
-from more_itertools import one
-
 import pytest # type: ignore
 
 
@@ -20,26 +18,11 @@ def test() -> None:
 
 @pytest.fixture(autouse=True)
 def prepare(tmp_path: Path):
-    from .common import reset_modules
+    from .shared_config import temp_config
-    reset_modules()
+    user_config = temp_config(tmp_path)
-
-    user_config = _prepare_google_config(tmp_path)
 
     import my.core.cfg as C
     with C.tmp_config() as config:
-        config.google = user_config # type: ignore
+        config.google = user_config.google
         yield
 
-
-def _prepare_google_config(tmp_path: Path):
-    from .common import testdata
-    track = one(testdata().rglob('italy-slovenia-2017-07-29.json'))
-
-    # 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

tests/location_fallback.py

@@ -0,0 +1,124 @@
+"""
+To test my.location.fallback_location.all
+"""
+
+from typing import Iterator
+from datetime import datetime, timezone, timedelta
+
+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
+ip_module.ips = data
+
+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
+    assert ilen(ip_module.ips()) == ilen(data())
+    assert ilen(ip_module.ips()) == ilen(via_ip.fallback_locations())
+    assert ilen(via_ip.fallback_locations()) == 5
+    assert ilen(via_ip._sorted_fallback_locations()) == 5
+
+    # confirm duration from via_ip since that is used for bisect
+    assert via_ip.config.for_duration == timedelta(hours=24)
+
+    # basic tests
+
+    # try estimating slightlight before the first IP
+    est = list(via_ip.estimate_location(datetime(2020, 1, 1, 11, 59, 59, tzinfo=timezone.utc)))
+    assert len(est) == 0
+
+    # during the duration for the first IP
+    est = list(via_ip.estimate_location(datetime(2020, 1, 1, 12, 30, 0, tzinfo=timezone.utc)))
+    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)))
+    assert len(est) == 0
+
+    # on 2/1/2020, theres one IP if before 16:30
+    est = list(via_ip.estimate_location(datetime(2020, 2, 1, 12, 30, 0, tzinfo=timezone.utc)))
+    assert len(est) == 1
+
+    # and two if after 16:30
+    est = list(via_ip.estimate_location(datetime(2020, 2, 1, 17, 00, 0, tzinfo=timezone.utc)))
+    assert len(est) == 2
+
+    # the 12:30 IP should 'expire' before the 16:30 IP, use 3:30PM on the next day
+    est = list(via_ip.estimate_location(datetime(2020, 2, 2, 15, 30, 0, tzinfo=timezone.utc)))
+    assert len(est) == 1
+
+    use_dt = datetime(2020, 3, 1, 12, 15, 0, tzinfo=timezone.utc)
+
+    # test last IP
+    est = list(via_ip.estimate_location(use_dt))
+    assert len(est) == 1
+
+    # datetime should be the IPs, not the passed IP (if via_home, it uses the passed dt)
+    assert est[0].dt != use_dt
+
+    # test interop with other fallback estimators/all.py
+    #
+    # 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
+    def _fe() -> Iterator[all.LocationEstimator]:
+        yield via_ip.estimate_location
+        yield via_home.estimate_location
+
+    all.fallback_estimators = _fe
+    assert ilen(all.fallback_estimators()) == 2
+
+    # test that all.estimate_location has access to both IPs
+    #
+    # 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].accuracy == 10_000
+
+    # passing home should give one
+    home_est = list(_iter_estimate_from(use_dt, (via_home.estimate_location,)))
+    assert len(home_est) == 1
+    assert home_est[0].accuracy == 30_000
+
+    # make sure ip accuracy is more accurate
+    assert raw_est[0].accuracy < home_est[0].accuracy
+
+    # passing both should give two
+    raw_est = list(_iter_estimate_from(use_dt, (via_ip.estimate_location, via_home.estimate_location)))
+    assert len(raw_est) == 2
+
+    # shouldnt raise value error
+    all_est = all.estimate_location(use_dt)
+    # 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
+    loc = all.estimate_location(datetime(2021, 1, 1, 12, 30, 0, tzinfo=timezone.utc))
+    assert loc.datasource == "via_home"
+
+    # test a different home using location.fallback.all
+    bulgaria = all.estimate_location(datetime(2006, 1, 1, 12, 30, 0, tzinfo=timezone.utc))
+    assert bulgaria.datasource == "via_home"
+    assert (bulgaria.lat, bulgaria.lon) == (42.697842, 23.325973)
+    assert (loc.lat, loc.lon) != (bulgaria.lat, bulgaria.lon)
+
+
+# re-use prepare fixture for overriding config from shared_config.py
+from .tz import prepare

tests/shared_config.py

@@ -0,0 +1,64 @@
+# Defines some shared config for tests
+
+from datetime import datetime, date, timezone
+from pathlib import Path
+
+from typing import Any, NamedTuple
+import my.time.tz.via_location as LTZ
+from more_itertools import one
+
+
+class SharedConfig(NamedTuple):
+    google: Any
+    location: Any
+    time: Any
+
+
+def _prepare_google_config(tmp_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())
+
+    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 .common import reset_modules
+    reset_modules()
+
+    LTZ.config.fast = True
+
+    class location:
+        home = (
+            # supports ISO strings
+            ('2005-12-04'                                       , (42.697842, 23.325973)), # Bulgaria, Sofia
+            # supports date/datetime objects
+            (date(year=1980, month=2, day=15)                   , (40.7128  , -74.0060 )), # NY
+            # check tz handling..
+            (datetime.fromtimestamp(1600000000, tz=timezone.utc), (55.7558  , 37.6173  )), # Moscow, Russia
+        )
+        # note: order doesn't matter, will be sorted in the data provider
+        class via_ip:
+            pass
+        class gpslogger:
+            pass
+
+    class time:
+        class tz:
+            class via_location:
+                pass # just rely on the defaults...
+
+
+    return SharedConfig(google=_prepare_google_config(temp_path), location=location, time=time)

tests/tz.py

@@ -1,5 +1,5 @@
 import sys
-from datetime import datetime, timedelta, date, timezone
+from datetime import datetime, timedelta
 from pathlib import Path
 
 import pytest # type: ignore
@@ -81,40 +81,15 @@ def D(dstr: str) -> datetime:
     return datetime.strptime(dstr, '%Y%m%d %H:%M:%S')
 
 
-# TODO copy pasted from location.py, need to extract some common provider
+
 @pytest.fixture(autouse=True)
 def prepare(tmp_path: Path):
-    from .common import reset_modules
+    from .shared_config import temp_config
-    reset_modules()
+    conf = temp_config(tmp_path)
-
-    LTZ.config.fast = True
-
-    from .location import _prepare_google_config
-    google = _prepare_google_config(tmp_path)
-
-    class location:
-        home = (
-            # supports ISO strings
-            ('2005-12-04'                                       , (42.697842, 23.325973)), # Bulgaria, Sofia
-            # supports date/datetime objects
-            (date(year=1980, month=2, day=15)                   , (40.7128  , -74.0060 )), # NY
-            # check tz handling..
-            (datetime.fromtimestamp(1600000000, tz=timezone.utc), (55.7558  , 37.6173  )), # Moscow, Russia
-        )
-        # note: order doesn't matter, will be sorted in the data provider
-        class via_ip:
-            pass
-        class gpslogger:
-            pass
-
-    class time:
-        class tz:
-            class via_location:
-                pass # just rely on the defaults...
 
     import my.core.cfg as C
     with C.tmp_config() as config:
-        config.google   = google
+        config.google   = conf.google
-        config.time     = time
+        config.time     = conf.time
-        config.location = location
+        config.location = conf.location
         yield