109 lines
3.0 KiB
Python
109 lines
3.0 KiB
Python
import os
|
|
import shutil
|
|
import sqlite3
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from meshchatx.src.backend.map_manager import MapManager
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
dir_path = tempfile.mkdtemp()
|
|
yield dir_path
|
|
shutil.rmtree(dir_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_config():
|
|
config = MagicMock()
|
|
config.map_offline_path.get.return_value = None
|
|
config.map_mbtiles_dir.get.return_value = None
|
|
return config
|
|
|
|
|
|
def test_map_manager_init(mock_config, temp_dir):
|
|
mm = MapManager(mock_config, temp_dir)
|
|
assert mm.storage_dir == temp_dir
|
|
|
|
|
|
def test_get_offline_path_default(mock_config, temp_dir):
|
|
mm = MapManager(mock_config, temp_dir)
|
|
default_path = os.path.join(temp_dir, "offline_map.mbtiles")
|
|
|
|
# Not exists
|
|
assert mm.get_offline_path() is None
|
|
|
|
# Exists
|
|
with open(default_path, "w") as f:
|
|
f.write("data")
|
|
assert mm.get_offline_path() == default_path
|
|
|
|
|
|
def test_list_mbtiles(mock_config, temp_dir):
|
|
mm = MapManager(mock_config, temp_dir)
|
|
|
|
# Create some dummy .mbtiles files
|
|
f1 = os.path.join(temp_dir, "map1.mbtiles")
|
|
f2 = os.path.join(temp_dir, "map2.mbtiles")
|
|
with open(f1, "w") as f:
|
|
f.write("1")
|
|
with open(f2, "w") as f:
|
|
f.write("22")
|
|
|
|
files = mm.list_mbtiles()
|
|
assert len(files) == 2
|
|
assert any(f["name"] == "map1.mbtiles" for f in files)
|
|
assert any(f["size"] == 2 for f in files if f["name"] == "map2.mbtiles")
|
|
|
|
|
|
def test_get_metadata(mock_config, temp_dir):
|
|
mm = MapManager(mock_config, temp_dir)
|
|
db_path = os.path.join(temp_dir, "test.mbtiles")
|
|
mock_config.map_offline_path.get.return_value = db_path
|
|
|
|
# Create valid sqlite mbtiles
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("CREATE TABLE metadata (name text, value text)")
|
|
conn.execute("INSERT INTO metadata VALUES ('name', 'Test Map')")
|
|
conn.execute("INSERT INTO metadata VALUES ('format', 'jpg')")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
metadata = mm.get_metadata()
|
|
assert metadata["name"] == "Test Map"
|
|
assert metadata["format"] == "jpg"
|
|
|
|
|
|
def test_get_tile(mock_config, temp_dir):
|
|
mm = MapManager(mock_config, temp_dir)
|
|
db_path = os.path.join(temp_dir, "test.mbtiles")
|
|
mock_config.map_offline_path.get.return_value = db_path
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute(
|
|
"CREATE TABLE tiles (zoom_level integer, tile_column integer, tile_row integer, tile_data blob)",
|
|
)
|
|
# Zoom 0, Tile 0,0. TMS y for 0/0/0 is (1<<0)-1-0 = 0
|
|
conn.execute(
|
|
"INSERT INTO tiles VALUES (0, 0, 0, ?)",
|
|
(sqlite3.Binary(b"tile_data"),),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
tile = mm.get_tile(0, 0, 0)
|
|
assert tile == b"tile_data"
|
|
|
|
|
|
def test_start_export_status(mock_config, temp_dir):
|
|
mm = MapManager(mock_config, temp_dir)
|
|
|
|
with patch.object(mm, "_run_export"):
|
|
export_id = mm.start_export("test_id", [0, 0, 1, 1], 0, 1)
|
|
assert export_id == "test_id"
|
|
status = mm.get_export_status(export_id)
|
|
assert status["status"] == "starting"
|