77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[1] / "scripts" / "fetch-waterbodies.py"
|
|
SPEC = importlib.util.spec_from_file_location("fetch_waterbodies", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
fetch_waterbodies = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(fetch_waterbodies)
|
|
|
|
|
|
def test_fetch_snapshot_saves_catalog_atomically(tmp_path: Path, monkeypatch) -> None:
|
|
output = tmp_path / "nested" / "catalog.json"
|
|
calls = []
|
|
|
|
class Result:
|
|
returncode = 0
|
|
stdout = '[{"source_external_id": "level_019_american_pond"}]'
|
|
stderr = ""
|
|
|
|
def run(command, **kwargs):
|
|
calls.append((command, kwargs))
|
|
return Result()
|
|
|
|
monkeypatch.setattr(fetch_waterbodies.subprocess, "run", run)
|
|
assert fetch_waterbodies.fetch_snapshot(
|
|
"catalog", url=None, output=output, state_file=tmp_path / "state.json", limit=19,
|
|
) == 0
|
|
assert json.loads(output.read_text(encoding="utf-8"))[0]["source_external_id"] == "level_019_american_pond"
|
|
assert calls[0][0][0] == fetch_waterbodies.sys.executable
|
|
assert "--state-file" in calls[0][0]
|
|
assert "--limit" in calls[0][0]
|
|
|
|
|
|
def test_fetch_snapshot_parses_saved_catalog_without_network(tmp_path: Path, monkeypatch) -> None:
|
|
html = Path(__file__).parent / "fixtures" / "rf4db_waterbodies_sample.html"
|
|
output = tmp_path / "catalog.json"
|
|
parser = fetch_waterbodies.parse_rf4db_waterbodies
|
|
monkeypatch.setattr(fetch_waterbodies, "parse_rf4db_waterbodies", lambda document: parser(document, expected_count=2))
|
|
assert fetch_waterbodies.fetch_snapshot(
|
|
"catalog", url=None, html=html, output=output,
|
|
state_file=tmp_path / "state.json", limit=19,
|
|
) == 0
|
|
payload = json.loads(output.read_text(encoding="utf-8"))
|
|
assert len(payload) == 2
|
|
assert payload[0]["source_system"] == "rf4db"
|
|
|
|
|
|
def test_fetch_snapshot_reports_invalid_saved_detail_html(tmp_path: Path, capsys) -> None:
|
|
html = tmp_path / "challenge.html"
|
|
html.write_text("<html><title>Just a moment...</title></html>", encoding="utf-8")
|
|
output = tmp_path / "detail.json"
|
|
assert fetch_waterbodies.fetch_snapshot(
|
|
"detail", url="https://download.rf4db.com/ru/maps/level_019_american_pond",
|
|
html=html, output=output, state_file=tmp_path / "state.json", limit=100,
|
|
) == 1
|
|
assert not output.exists()
|
|
assert "local HTML parse failed" in capsys.readouterr().err
|
|
|
|
|
|
def test_fetch_snapshot_does_not_write_failed_fetch(tmp_path: Path, monkeypatch, capsys) -> None:
|
|
output = tmp_path / "detail.json"
|
|
|
|
class Result:
|
|
returncode = 1
|
|
stdout = ""
|
|
stderr = "community source failed: source cooldown is active\n"
|
|
|
|
monkeypatch.setattr(fetch_waterbodies.subprocess, "run", lambda *args, **kwargs: Result())
|
|
assert fetch_waterbodies.fetch_snapshot(
|
|
"detail", url="https://download.rf4db.com/ru/maps/level_019_american_pond",
|
|
output=output, state_file=tmp_path / "state.json", limit=100,
|
|
) == 1
|
|
assert not output.exists()
|
|
assert "cooldown" in capsys.readouterr().err
|