mirror of
https://gitlvb.teallvbs.xyz/IQ.Lvbs/IQ.Pilot.git
synced 2026-09-11 08:13:43 +08:00
IQ.Lvbs CI [bot]
98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
# Copyright (c) 2026 IQ.Lvbs. All rights reserved.
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from iqpilot.ui.onroad import nav_map_panel as nav
|
|
|
|
|
|
class ImmediateThread:
|
|
def __init__(self, target, args=(), **kwargs):
|
|
self.target = target
|
|
self.args = args
|
|
|
|
def start(self):
|
|
self.target(*self.args)
|
|
|
|
|
|
@pytest.fixture
|
|
def provider(tmp_path, monkeypatch):
|
|
obj = nav.MapboxTileProvider.__new__(nav.MapboxTileProvider)
|
|
obj._textures = {}
|
|
obj._pending_tiles = {}
|
|
obj._inflight = set()
|
|
obj._lock = threading.Lock()
|
|
obj._cache_disabled = False
|
|
obj._provider_disabled = False
|
|
obj._cache_root = tmp_path
|
|
obj._style = 'test'
|
|
obj._keep_margin = 0
|
|
obj._status = 'idle'
|
|
obj._token = lambda: 'test'
|
|
obj._prune_cache = lambda keep: None
|
|
monkeypatch.setattr(nav.threading, 'Thread', ImmediateThread)
|
|
monkeypatch.setattr(nav, '_decode_tile_image', lambda payload: object())
|
|
monkeypatch.setattr(nav.rl, 'unload_image', lambda image: None)
|
|
return obj
|
|
|
|
|
|
def cache_tiles(provider, count):
|
|
tiles = [(15, x, 10) for x in range(count)]
|
|
for tile in tiles:
|
|
path = provider._cache_path(tile)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b'tile')
|
|
return tiles
|
|
|
|
|
|
def test_fast_cache_workers_keep_decoded_images_bounded(provider):
|
|
tiles = cache_tiles(provider, 64)
|
|
for tile in tiles:
|
|
assert provider._queue_cached_tile(tile)
|
|
assert len(provider._pending_tiles) == nav.MAX_INFLIGHT_TILES
|
|
assert not provider._inflight
|
|
|
|
|
|
def test_cache_capacity_includes_running_and_decoded_tiles(provider):
|
|
tiles = cache_tiles(provider, 20)
|
|
provider._inflight.update(tiles[:3])
|
|
provider._pending_tiles.update({tile: object() for tile in tiles[3:8]})
|
|
for tile in tiles[8:]:
|
|
assert provider._queue_cached_tile(tile)
|
|
assert len(provider._inflight) + len(provider._pending_tiles) == nav.MAX_INFLIGHT_TILES
|
|
|
|
|
|
def test_cached_pending_tiles_are_not_fetched_again(provider):
|
|
tiles = cache_tiles(provider, 4)
|
|
fetched = []
|
|
provider._fetch_worker = lambda tile, token: fetched.append(tile)
|
|
provider._visible_tile_keys = lambda *args: (15, 1, 0, 0, tiles)
|
|
provider.update(0, 0, 15, 100, 100)
|
|
assert set(provider._pending_tiles) == set(tiles)
|
|
assert fetched == []
|
|
|
|
|
|
@pytest.mark.parametrize('provider_class', [nav.MapboxTileProvider, nav.OsmOfflineProvider])
|
|
def test_worker_completion_during_viewport_calculation_stays_bounded(provider, provider_class):
|
|
tiles = [(15, x, 10) for x in range(20)]
|
|
obj = provider if provider_class is nav.MapboxTileProvider else provider_class.__new__(provider_class)
|
|
if obj is not provider:
|
|
obj.__dict__.update(provider.__dict__)
|
|
obj._refresh_source = lambda *args: None
|
|
obj._status = 'offline_ready'
|
|
obj._source_tile_for_request = lambda tile: (tile, 0)
|
|
launched = []
|
|
obj._consume_pending = lambda: None
|
|
obj._queue_cached_tile = lambda tile: False
|
|
obj._fetch_worker = lambda tile, token: launched.append(tile)
|
|
obj._load_decode_worker = lambda tile: launched.append(tile)
|
|
|
|
def visible(*args):
|
|
obj._pending_tiles.update({tile: object() for tile in tiles[:5]})
|
|
obj._inflight.update(tiles[5:8])
|
|
return 15, 1, 0, 0, tiles
|
|
|
|
obj._visible_tile_keys = visible
|
|
obj.update(0, 0, 15, 100, 100)
|
|
assert launched == []
|