638 lines
24 KiB
Python
638 lines
24 KiB
Python
import hashlib
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import Mock
|
|
from xml.etree import ElementTree as ET
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import rasterio
|
|
from rasterio.transform import from_bounds
|
|
|
|
from flight_map.cli import main, parser, timestamp
|
|
from flight_map.geo import DEFAULT_BOUNDS, Bounds, Projection
|
|
from flight_map.lettering import GLYPHS, lettering
|
|
from flight_map.maps import map_data, map_lines
|
|
from flight_map.network import request_json
|
|
from flight_map.plot import SVG_NS, PlotOptions, render, utc
|
|
from flight_map.plot import label as draw_label
|
|
from flight_map.record import OpenSky, collect
|
|
from flight_map.storage import (
|
|
Observation,
|
|
Session,
|
|
checkpoint,
|
|
create_session,
|
|
database,
|
|
get_session,
|
|
initialize,
|
|
load_observations,
|
|
save_observations,
|
|
)
|
|
from flight_map.terrain import Elevation
|
|
from flight_map.tracks import make_tracks, split_tracks
|
|
|
|
T0 = 1_792_497_600.0
|
|
|
|
|
|
def observation(**changes: Any) -> Observation:
|
|
return replace(Observation("abc123", "KLM123", T0 + 10, 4.80, 52.24, 900, 1043), **changes)
|
|
|
|
|
|
def raster(path: Path, value: float, *, bounds: Bounds = DEFAULT_BOUNDS) -> Path:
|
|
with rasterio.open(
|
|
path,
|
|
"w",
|
|
driver="GTiff",
|
|
height=10,
|
|
width=10,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:4326",
|
|
transform=from_bounds(*bounds.values, 10, 10),
|
|
nodata=-9999,
|
|
) as dataset:
|
|
dataset.write(np.full((1, 10, 10), value, dtype="float32"))
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def recorded(tmp_path: Path) -> tuple[Path, Session]:
|
|
path = tmp_path / "flights.sqlite"
|
|
with database(path) as connection:
|
|
initialize(connection)
|
|
session = create_session(connection, DEFAULT_BOUNDS, 2000, 30, T0)
|
|
items = [
|
|
observation(longitude=4.77, timestamp=T0 + 10, geo_altitude=843),
|
|
observation(longitude=4.82, timestamp=T0 + 40, geo_altitude=1043),
|
|
observation(longitude=4.88, timestamp=T0 + 70, geo_altitude=1243),
|
|
]
|
|
save_observations(connection, session, items, T0 + 80)
|
|
checkpoint(connection, session.id, T0 + 90, "stopped")
|
|
return path, session
|
|
|
|
|
|
@pytest.fixture
|
|
def local_map(tmp_path: Path) -> Path:
|
|
path = tmp_path / "map.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"elements": [
|
|
{
|
|
"type": "way",
|
|
"id": 1,
|
|
"geometry": [
|
|
{"lon": 4.77, "lat": 52.235},
|
|
{"lon": 4.90, "lat": 52.245},
|
|
],
|
|
},
|
|
{
|
|
"type": "way",
|
|
"id": 2,
|
|
"geometry": [
|
|
{"lon": 4.80, "lat": 52.22},
|
|
{"lon": 4.83, "lat": 52.26},
|
|
],
|
|
},
|
|
]
|
|
}
|
|
)
|
|
)
|
|
return path
|
|
|
|
|
|
def test_default_and_reversed_bounds() -> None:
|
|
assert DEFAULT_BOUNDS.values == (4.784, 52.232, 4.864, 52.258)
|
|
assert Bounds.corners(4.864, 52.258, 4.784, 52.232) == DEFAULT_BOUNDS
|
|
assert DEFAULT_BOUNDS.contains(4.83, 52.24)
|
|
expanded = DEFAULT_BOUNDS.expand(2000)
|
|
assert expanded.west < DEFAULT_BOUNDS.west and expanded.north > DEFAULT_BOUNDS.north
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"values",
|
|
[
|
|
(1, 1, 1, 2),
|
|
(180, 1, 181, 2),
|
|
(-179, 1, 179, 2),
|
|
(1, -90, 2, 90),
|
|
(1, 2, float("nan"), 3),
|
|
],
|
|
)
|
|
def test_invalid_bounds(values: tuple[float, float, float, float]) -> None:
|
|
with pytest.raises(ValueError):
|
|
Bounds(*values)
|
|
|
|
|
|
def test_projection_corners() -> None:
|
|
projection = Projection(DEFAULT_BOUNDS)
|
|
southwest = projection.point(DEFAULT_BOUNDS.west, DEFAULT_BOUNDS.south)
|
|
northeast = projection.point(DEFAULT_BOUNDS.east, DEFAULT_BOUNDS.north)
|
|
assert projection.rectangle.bounds == (*southwest, *northeast)
|
|
assert northeast[0] > southwest[0] and northeast[1] > southwest[1]
|
|
|
|
|
|
def test_persistence_and_deduplication(recorded: tuple[Path, Session]) -> None:
|
|
path, original = recorded
|
|
with database(path) as connection:
|
|
session = get_session(connection, original.id)
|
|
assert session.bounds == DEFAULT_BOUNDS
|
|
assert session.collection_bounds == original.collection_bounds
|
|
assert session.status == "stopped"
|
|
items = load_observations(connection, session.id, T0, T0 + 100)
|
|
assert len(items) == 3
|
|
save_observations(connection, session, items, T0 + 100)
|
|
assert len(load_observations(connection, session.id, T0, T0 + 100)) == 3
|
|
assert items[0].geo_altitude == 843
|
|
with pytest.raises(ValueError, match="not found"):
|
|
get_session(connection, "missing")
|
|
|
|
|
|
def test_observations_outside_session_are_not_saved(recorded: tuple[Path, Session]) -> None:
|
|
path, session = recorded
|
|
with database(path) as connection:
|
|
save_observations(
|
|
connection,
|
|
session,
|
|
[
|
|
observation(timestamp=T0 - 1),
|
|
observation(timestamp=T0 + 500),
|
|
observation(longitude=5.5),
|
|
],
|
|
T0 + 100,
|
|
)
|
|
assert len(load_observations(connection, session.id, T0, T0 + 100)) == 3
|
|
|
|
|
|
def test_database_readonly(recorded: tuple[Path, Session]) -> None:
|
|
path, _ = recorded
|
|
with database(path, readonly=True) as connection:
|
|
with pytest.raises(sqlite3.OperationalError, match="readonly"):
|
|
connection.execute("DELETE FROM observations")
|
|
|
|
|
|
def test_local_terrain_summary(tmp_path: Path) -> None:
|
|
terrain = raster(tmp_path / "terrain.tif", -2)
|
|
with Elevation(terrain, "orthometric") as elevation:
|
|
assert elevation.mean_ground(DEFAULT_BOUNDS, 4) == -2
|
|
|
|
|
|
def test_terrain_missing_data(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
|
terrain = raster(tmp_path / "terrain.tif", -9999)
|
|
with Elevation(terrain, "wgs84-ellipsoid") as elevation:
|
|
assert elevation.mean_ground(DEFAULT_BOUNDS, 4) is None
|
|
small = Bounds(4.80, 52.24, 4.81, 52.25)
|
|
raster(terrain, 0, bounds=small)
|
|
with Elevation(terrain, "wgs84-ellipsoid") as elevation:
|
|
assert elevation.mean_ground(DEFAULT_BOUNDS, 16) == 0
|
|
assert 0 < elevation.land_coverage < 1
|
|
assert "averaging valid land only" in caplog.text
|
|
|
|
|
|
def test_equal_area_ground_average(tmp_path: Path) -> None:
|
|
terrain = raster(tmp_path / "terrain.tif", 0)
|
|
with rasterio.open(terrain, "r+") as dataset:
|
|
values = np.zeros((10, 10), dtype="float32")
|
|
values[:, 5:] = 10
|
|
dataset.write(values, 1)
|
|
with Elevation(terrain, "wgs84-ellipsoid") as elevation:
|
|
assert elevation.mean_ground(DEFAULT_BOUNDS, 10) == pytest.approx(5)
|
|
|
|
|
|
def test_reported_altitude_is_time_weighted_and_excludes_extensions() -> None:
|
|
bounds = Bounds(0, 0, 2, 1)
|
|
observations = [
|
|
observation(longitude=-1, latitude=0.5, timestamp=T0, baro_altitude=0),
|
|
observation(longitude=1, latitude=0.5, timestamp=T0 + 20, baro_altitude=200),
|
|
observation(longitude=3, latitude=0.5, timestamp=T0 + 60, baro_altitude=600),
|
|
]
|
|
tracks = make_tracks(observations, Projection(bounds), 120)
|
|
# Inside: 10s at mean altitude 150 + 20s at mean altitude 300.
|
|
assert len(tracks) == 1
|
|
assert tracks[0].average_altitude == pytest.approx(250)
|
|
assert tracks[0].start == pytest.approx(T0 + 10)
|
|
assert tracks[0].end == pytest.approx(T0 + 40)
|
|
|
|
|
|
def test_stationary_positions_and_missing_altitude() -> None:
|
|
items = [
|
|
observation(),
|
|
observation(timestamp=T0 + 25),
|
|
observation(timestamp=T0 + 40, longitude=4.82, baro_altitude=None),
|
|
]
|
|
tracks = make_tracks(items, Projection(DEFAULT_BOUNDS), 120)
|
|
assert len(tracks) == 1 and tracks[0].average_altitude is None
|
|
assert make_tracks(items[:2], Projection(DEFAULT_BOUNDS), 120) == []
|
|
|
|
|
|
def test_reported_altitude_does_not_require_geometric_altitude() -> None:
|
|
items = [
|
|
observation(geo_altitude=None),
|
|
observation(timestamp=T0 + 40, longitude=4.82, geo_altitude=None),
|
|
]
|
|
tracks = make_tracks(items, Projection(DEFAULT_BOUNDS), 120)
|
|
assert len(tracks) == 1
|
|
assert tracks[0].average_altitude == 900
|
|
|
|
|
|
@pytest.mark.parametrize("altitude", [0, -50, 1234])
|
|
def test_zero_negative_and_positive_reported_altitudes_are_preserved(altitude: float) -> None:
|
|
items = [
|
|
observation(baro_altitude=altitude, geo_altitude=9999),
|
|
observation(timestamp=T0 + 40, longitude=4.82, baro_altitude=altitude, geo_altitude=9999),
|
|
]
|
|
tracks = make_tracks(items, Projection(DEFAULT_BOUNDS), 120)
|
|
assert tracks[0].average_altitude == pytest.approx(altitude)
|
|
|
|
|
|
def test_track_splitting_and_ground_filter() -> None:
|
|
items = [observation(timestamp=T0 + i * 30) for i in range(6)]
|
|
items[2] = replace(items[2], on_ground=True)
|
|
items[4] = replace(items[4], callsign="OTHER")
|
|
items[5] = replace(items[5], callsign="OTHER")
|
|
assert [len(track) for track in split_tracks(items, 120)] == [2, 2]
|
|
assert split_tracks([observation(), observation(timestamp=T0 + 500)], 120) == []
|
|
|
|
|
|
def test_stroke_font_has_no_fills_or_missing_common_characters() -> None:
|
|
for character in GLYPHS:
|
|
assert all(line.length > 0 for line in lettering(character, 0, 0, 2))
|
|
with pytest.raises(ValueError, match="Unsupported"):
|
|
lettering("@", 0, 0, 2)
|
|
|
|
|
|
def test_end_to_end_offline_repeatable(
|
|
recorded: tuple[Path, Session],
|
|
local_map: Path,
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
path, session = recorded
|
|
monkeypatch.setattr("flight_map.record.OpenSky.observations", Mock(side_effect=AssertionError))
|
|
monkeypatch.setattr("flight_map.maps.request_json", Mock(side_effect=AssertionError))
|
|
terrain = raster(tmp_path / "terrain.tif", 41, bounds=session.collection_bounds)
|
|
before = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
output = tmp_path / "map.svg"
|
|
for name in ("map.svg", "second.svg"):
|
|
assert (
|
|
render(
|
|
path,
|
|
session.id,
|
|
tmp_path / name,
|
|
local_map=local_map,
|
|
terrain=terrain,
|
|
terrain_reference="wgs84-ellipsoid",
|
|
)
|
|
== 1
|
|
)
|
|
assert output.read_bytes() == (tmp_path / "second.svg").read_bytes()
|
|
assert hashlib.sha256(path.read_bytes()).hexdigest() == before
|
|
root = ET.parse(output).getroot()
|
|
assert root.attrib["width"] == "297mm" and root.attrib["height"] == "210mm"
|
|
assert root.attrib["fill"] == "none"
|
|
assert root.attrib["viewBox"] == "0 0 297 210"
|
|
forbidden = {"image", "text", "clipPath", "mask", "use"}
|
|
assert not any(element.tag.split("}")[-1] in forbidden for element in root.iter())
|
|
assert {g.attrib.get("id") for g in root.findall(f"{{{SVG_NS}}}g")} == {
|
|
"map",
|
|
"flights",
|
|
"lettering",
|
|
}
|
|
flight_paths = root.findall(f".//{{{SVG_NS}}}g[@id='flights']//{{{SVG_NS}}}path")
|
|
assert flight_paths and all("C " in element.attrib["d"] for element in flight_paths)
|
|
flight = root.find(f".//{{{SVG_NS}}}g[@id='flight-1']")
|
|
assert flight is not None and flight.attrib["data-callsign"] == "KLM123"
|
|
assert flight.attrib["data-average-barometric-altitude-metres"] == "900.000"
|
|
assert flight.attrib["data-altitude-source"] == "barometric"
|
|
for element in root.iter(f"{{{SVG_NS}}}path"):
|
|
assert "clip-path" not in element.attrib
|
|
for x, y in re.findall(r"(-?\d+\.\d+),(-?\d+\.\d+)", element.attrib["d"]):
|
|
assert 0 <= float(x) <= 297 and 0 <= float(y) <= 210
|
|
assert "BARO ALT" not in output.read_text() # Lettering is geometry, not SVG text.
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
render(path, session.id, output, local_map=local_map)
|
|
|
|
|
|
def test_invalid_plot_time_and_empty_data(recorded: tuple[Path, Session], tmp_path: Path) -> None:
|
|
path, session = recorded
|
|
with pytest.raises(ValueError, match="within"):
|
|
render(path, session.id, tmp_path / "bad.svg", start=T0 - 10)
|
|
with pytest.raises(ValueError, match="No recorded"):
|
|
render(path, session.id, tmp_path / "bad.svg", start=T0 + 81, end=T0 + 85)
|
|
with pytest.raises(ValueError, match="positive"):
|
|
PlotOptions(width=float("nan")).validate()
|
|
with pytest.raises(ValueError, match="no room"):
|
|
PlotOptions(width=10).validate()
|
|
with pytest.raises(ValueError, match="must not overwrite"):
|
|
render(path, session.id, path, force=True)
|
|
with pytest.raises(ValueError, match="must not overwrite"):
|
|
render(
|
|
path, session.id, tmp_path / "terrain.tif", terrain=tmp_path / "terrain.tif", force=True
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("count", [1, 2, 3, 4])
|
|
def test_printed_flight_legend_midpoint_and_two_columns(
|
|
recorded: tuple[Path, Session],
|
|
local_map: Path,
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
count: int,
|
|
) -> None:
|
|
path, session = recorded
|
|
with database(path) as connection:
|
|
original = load_observations(connection, session.id, T0, T0 + 90)
|
|
for index in range(1, count):
|
|
save_observations(
|
|
connection,
|
|
session,
|
|
[
|
|
replace(item, icao24=f"{index:06x}", callsign=f"TEST{index}")
|
|
for item in original
|
|
],
|
|
T0 + 90,
|
|
)
|
|
expected_tracks = make_tracks(
|
|
load_observations(connection, session.id, T0, T0 + 90),
|
|
Projection(session.bounds),
|
|
120,
|
|
)
|
|
captured = Mock(wraps=draw_label)
|
|
monkeypatch.setattr("flight_map.plot.label", captured)
|
|
output = tmp_path / "legend.svg"
|
|
options = PlotOptions()
|
|
assert (
|
|
render(path, session.id, output, local_map=local_map, no_terrain=True, options=options)
|
|
== count
|
|
)
|
|
entries = [call for call in captured.call_args_list if re.match(r"^\d+: ", call.args[1])]
|
|
assert len(entries) == count
|
|
root = ET.parse(output).getroot()
|
|
rows = (count + 1) // 2
|
|
legend_width = options.width - 2 * options.margin
|
|
column_gap = 2 * options.font_height
|
|
column_width = (legend_width - column_gap) / 2
|
|
for index, call in enumerate(entries):
|
|
text, x, y, height = call.args[1:]
|
|
flight = root.find(f".//{{{SVG_NS}}}g[@id='flight-{index + 1}']")
|
|
assert flight is not None
|
|
track = expected_tracks[index]
|
|
assert utc((track.start + track.end) / 2) in text
|
|
assert flight.attrib["data-start-utc"] == utc(track.start)
|
|
assert flight.attrib["data-end-utc"] == utc(track.end)
|
|
assert len(re.findall(r"\d{4}-\d{2}-\d{2}", text)) == 1
|
|
assert len(re.findall(r"\d{2}:\d{2}:\d{2}", text)) == 1
|
|
assert " Alt 900 M" in text and "BARO" not in text.upper() and " TO " not in text
|
|
column, row = divmod(index, rows)
|
|
assert x == pytest.approx(options.margin + column * (column_width + column_gap))
|
|
assert y == pytest.approx(entries[0].args[3] + row * height * 1.8)
|
|
assert len(text) * height <= column_width
|
|
assert x + len(text) * height <= options.width - options.margin
|
|
assert y + height <= options.height - options.margin
|
|
if count >= 2:
|
|
assert entries[rows].args[2] > entries[0].args[2]
|
|
|
|
|
|
def test_map_cache_and_errors(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
payload = {
|
|
"elements": [
|
|
{
|
|
"type": "way",
|
|
"geometry": [
|
|
{"lon": 4.8, "lat": 52.24},
|
|
{"lon": 4.82, "lat": 52.24},
|
|
],
|
|
}
|
|
]
|
|
}
|
|
fetch = Mock(return_value=payload)
|
|
monkeypatch.setattr("flight_map.maps.request_json", fetch)
|
|
assert map_data(DEFAULT_BOUNDS, tmp_path) == payload
|
|
assert map_data(DEFAULT_BOUNDS, tmp_path) == payload
|
|
fetch.assert_called_once()
|
|
with pytest.raises(ValueError, match="No drawable"):
|
|
map_lines({"elements": []}, Projection(DEFAULT_BOUNDS))
|
|
with pytest.raises(ValueError, match="out geom"):
|
|
map_lines({"elements": [{"type": "way"}]}, Projection(DEFAULT_BOUNDS))
|
|
fetch.return_value = {"elements": [], "remark": "runtime error"}
|
|
with pytest.raises(ValueError, match="could not complete"):
|
|
map_data(DEFAULT_BOUNDS, tmp_path / "other")
|
|
|
|
|
|
def test_cache_refresh_is_atomic_and_keys_include_area(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
old = {"elements": [], "version": 1}
|
|
new = {"elements": [], "version": 2}
|
|
fetch = Mock(return_value=old)
|
|
monkeypatch.setattr("flight_map.maps.request_json", fetch)
|
|
assert map_data(DEFAULT_BOUNDS, tmp_path) == old
|
|
entry = next(tmp_path.glob("osm-*.json"))
|
|
fetch.return_value = new
|
|
assert map_data(DEFAULT_BOUNDS, tmp_path) == old
|
|
assert fetch.call_count == 1
|
|
assert map_data(DEFAULT_BOUNDS, tmp_path, refresh=True) == new
|
|
assert json.loads(entry.read_text()) == new
|
|
assert not list(tmp_path.glob("*.tmp"))
|
|
fetch.side_effect = RuntimeError("offline")
|
|
with pytest.raises(RuntimeError, match="offline"):
|
|
map_data(DEFAULT_BOUNDS, tmp_path, refresh=True)
|
|
assert json.loads(entry.read_text()) == new
|
|
assert map_data(DEFAULT_BOUNDS, tmp_path) == new
|
|
fetch.side_effect = None
|
|
map_data(Bounds(4.8, 52.2, 4.9, 52.3), tmp_path)
|
|
assert len(list(tmp_path.glob("osm-*.json"))) == 2
|
|
entry.write_text("{")
|
|
with pytest.raises(ValueError, match="refresh-map"):
|
|
map_data(DEFAULT_BOUNDS, tmp_path)
|
|
assert map_data(DEFAULT_BOUNDS, tmp_path, refresh=True) == new
|
|
with pytest.raises(ValueError, match="cannot be combined"):
|
|
map_data(DEFAULT_BOUNDS, tmp_path, entry, refresh=True)
|
|
|
|
|
|
def test_interrupted_cache_write_preserves_existing_entry(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
payload = {"elements": []}
|
|
monkeypatch.setattr("flight_map.maps.request_json", Mock(return_value=payload))
|
|
map_data(DEFAULT_BOUNDS, tmp_path)
|
|
entry = next(tmp_path.glob("osm-*.json"))
|
|
before = entry.read_bytes()
|
|
monkeypatch.setattr("flight_map.maps.os.replace", Mock(side_effect=OSError("write failed")))
|
|
with pytest.raises(OSError, match="write failed"):
|
|
map_data(DEFAULT_BOUNDS, tmp_path, refresh=True)
|
|
assert entry.read_bytes() == before
|
|
assert not list(tmp_path.glob("*.tmp"))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"data",
|
|
[
|
|
{"elements": [None]},
|
|
{
|
|
"elements": [
|
|
{
|
|
"type": "way",
|
|
"geometry": [
|
|
{"lon": float("nan"), "lat": 52.2},
|
|
{"lon": 4.8, "lat": 52.3},
|
|
],
|
|
}
|
|
]
|
|
},
|
|
{"elements": [{"type": "way", "geometry": [None, None]}]},
|
|
],
|
|
)
|
|
def test_invalid_map_geometry_is_reported(data: dict[str, Any]) -> None:
|
|
with pytest.raises(ValueError, match="Map"):
|
|
map_lines(data, Projection(DEFAULT_BOUNDS))
|
|
|
|
|
|
def test_map_detail_is_selected_for_pen_width() -> None:
|
|
geometry = [
|
|
{"lon": 4.80, "lat": 52.24},
|
|
{"lon": 4.8001, "lat": 52.2401},
|
|
{"lon": 4.80, "lat": 52.24},
|
|
]
|
|
data = {
|
|
"elements": [
|
|
{"type": "way", "tags": {"building": "yes"}, "geometry": geometry},
|
|
{"type": "way", "tags": {"highway": "footway"}, "geometry": geometry},
|
|
{"type": "way", "tags": {"waterway": "stream"}, "geometry": geometry},
|
|
]
|
|
}
|
|
projection = Projection(DEFAULT_BOUNDS)
|
|
assert len(map_lines(data, projection, min_building_size=20)) == 1
|
|
assert len(map_lines(data, projection, min_building_size=0)) == 2
|
|
|
|
|
|
def test_opensky_parsing_and_auth(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
for key in ("OPENSKY_CLIENT_ID", "OPENSKY_CLIENT_SECRET", "OPENSKY_ACCESS_TOKEN"):
|
|
monkeypatch.delenv(key, raising=False)
|
|
state = ["abc123", " KLM123 ", "NL", T0, T0, 4.82, 52.24, 900, False, 100, 30, 0, None, 1043]
|
|
fetch = Mock(return_value={"states": [state]})
|
|
monkeypatch.setattr("flight_map.record.request_json", fetch)
|
|
client = OpenSky()
|
|
items = client.observations(DEFAULT_BOUNDS)
|
|
assert items[0].callsign == "KLM123" and items[0].geo_altitude == 1043
|
|
assert f"lamin={DEFAULT_BOUNDS.south}" in fetch.call_args.args[0]
|
|
state[5] = None
|
|
assert client.observations(DEFAULT_BOUNDS) == []
|
|
fetch.return_value = {"states": None}
|
|
assert client.observations(DEFAULT_BOUNDS) == []
|
|
fetch.return_value = {"wrong": []}
|
|
with pytest.raises(RuntimeError, match="missing"):
|
|
client.observations(DEFAULT_BOUNDS)
|
|
monkeypatch.setenv("OPENSKY_CLIENT_ID", "test-client")
|
|
with pytest.raises(ValueError, match="both"):
|
|
OpenSky()
|
|
monkeypatch.setenv("OPENSKY_CLIENT_SECRET", "test-secret")
|
|
fetch.side_effect = [{"access_token": "test-token", "expires_in": 300}, {"states": []}]
|
|
OpenSky().observations(DEFAULT_BOUNDS)
|
|
assert fetch.call_args.kwargs["headers"]["Authorization"] == "Bearer test-token"
|
|
|
|
|
|
def test_collection_checkpoints_without_map_services(
|
|
recorded: tuple[Path, Session],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
path, session = recorded
|
|
monkeypatch.setattr("flight_map.maps.request_json", Mock(side_effect=AssertionError))
|
|
client = Mock(spec=OpenSky)
|
|
client.observations.return_value = []
|
|
with database(path) as connection:
|
|
collect(connection, session, client, duration=0)
|
|
assert get_session(connection, session.id).status == "stopped"
|
|
client.observations.side_effect = RuntimeError("outage")
|
|
with pytest.raises(RuntimeError, match="outage"):
|
|
collect(connection, session, client, duration=0)
|
|
assert get_session(connection, session.id).status == "failed"
|
|
client.observations.side_effect = KeyboardInterrupt
|
|
collect(connection, session, client)
|
|
assert get_session(connection, session.id).status == "interrupted"
|
|
|
|
|
|
def test_cli_defaults_custom_bounds_and_resume(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
path = tmp_path / "flights.sqlite"
|
|
collector = Mock()
|
|
monkeypatch.setattr("flight_map.record.collect", collector)
|
|
monkeypatch.setattr("flight_map.record.OpenSky", Mock())
|
|
assert main(["collect", "--db", str(path), "--duration", "1"]) == 0
|
|
session = collector.call_args.args[1]
|
|
assert session.bounds == DEFAULT_BOUNDS
|
|
custom = ["--bounds", "4.90", "52.30", "4.80", "52.20"]
|
|
assert main(["collect", "--db", str(path), *custom, "--duration", "1"]) == 0
|
|
custom_session = collector.call_args.args[1]
|
|
assert custom_session.bounds == Bounds(4.8, 52.2, 4.9, 52.3)
|
|
assert main(["collect", "--db", str(path), "--resume", session.id, "--duration", "1"]) == 0
|
|
assert collector.call_args.args[1].bounds == DEFAULT_BOUNDS
|
|
assert main(["collect", "--db", str(path), "--resume", session.id, *custom]) == 1
|
|
assert main(["sessions", "--db", str(path)]) == 0
|
|
|
|
|
|
def test_cli_validation_and_plot(
|
|
recorded: tuple[Path, Session],
|
|
local_map: Path,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
path, session = recorded
|
|
assert (
|
|
main(
|
|
[
|
|
"plot",
|
|
"--db",
|
|
str(path),
|
|
"--session",
|
|
session.id,
|
|
"--map-data",
|
|
str(local_map),
|
|
"--no-terrain",
|
|
"--output",
|
|
str(tmp_path / "cli.svg"),
|
|
]
|
|
)
|
|
== 0
|
|
)
|
|
assert (
|
|
main(
|
|
[
|
|
"plot",
|
|
"--db",
|
|
str(path),
|
|
"--session",
|
|
"missing",
|
|
"--output",
|
|
str(tmp_path / "bad.svg"),
|
|
]
|
|
)
|
|
== 1
|
|
)
|
|
with pytest.raises(SystemExit):
|
|
parser().parse_args(["collect", "--db", str(path), "--interval", "nan"])
|
|
with pytest.raises(Exception, match="time zone"):
|
|
timestamp("2026-10-20T12:00:00")
|
|
assert timestamp("2026-10-20T12:00:00Z") == timestamp("2026-10-20T14:00:00+02:00")
|
|
|
|
|
|
def test_network_errors(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from email.message import Message
|
|
from urllib.error import HTTPError, URLError
|
|
|
|
headers = Message()
|
|
headers["Retry-After"] = "60"
|
|
fetch = Mock(side_effect=HTTPError("https://example.invalid", 429, "limited", headers, None))
|
|
monkeypatch.setattr("flight_map.network.urlopen", fetch)
|
|
with pytest.raises(RuntimeError, match="429.*60"):
|
|
request_json("https://example.invalid")
|
|
fetch.side_effect = URLError("offline")
|
|
with pytest.raises(RuntimeError, match="connection failed"):
|
|
request_json("https://example.invalid")
|