ik
4974f362ac
- _validate_url_before_io: check scheme (HTTPS only), port (80/443), host - fetch_html: recursive redirect validation with hop limit (MAX_REDIRECT_HOPS=5) - Reject non-HTTPS redirects and non-standard ports - All validation happens BEFORE urlopen() call - Updated tests for new validation messages - 109 Python tests pass
80 lines
3.3 KiB
Python
80 lines
3.3 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from rf4_research import community_cli
|
|
from rf4_research.community_cli import enforce_fetch_interval, fetch_site_key, mark_fetch
|
|
|
|
|
|
def test_fetch_cooldown_is_persistent_per_source(tmp_path: Path) -> None:
|
|
state_file = tmp_path / "fetch-state.json"
|
|
mark_fetch("rf4map-point", state_file=state_file, now=1_000)
|
|
|
|
with pytest.raises(RuntimeError, match="retry in 1800 seconds"):
|
|
enforce_fetch_interval("rf4map-point", state_file=state_file, now=1_000)
|
|
enforce_fetch_interval("rf4posts-spot", state_file=state_file, now=1_000)
|
|
enforce_fetch_interval("rf4map-point", state_file=state_file, now=2_800)
|
|
|
|
|
|
def test_fetch_site_key_groups_endpoints_and_normalizes_www() -> None:
|
|
assert fetch_site_key("https://rf4-stat.ru/fishing/") == "rf4-stat.ru"
|
|
assert fetch_site_key("https://www.rf4-stat.ru/posts/") == "rf4-stat.ru"
|
|
|
|
|
|
def test_failed_fetch_still_reserves_site_cooldown(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
state_file = tmp_path / "fetch-state.json"
|
|
|
|
def fail(_url: str) -> str:
|
|
raise OSError("offline")
|
|
|
|
monkeypatch.setattr(community_cli, "fetch_html", fail)
|
|
assert community_cli.main(["rf4db", "--state-file", str(state_file)]) == 1
|
|
assert "download.rf4db.com" in json.loads(state_file.read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_validate_url_host_rejects_disallowed_hosts() -> None:
|
|
from rf4_research.community_cli import _validate_url_host
|
|
|
|
# Allowed hosts pass
|
|
assert _validate_url_host("https://download.rf4db.com/ru/catches") == "download.rf4db.com"
|
|
assert _validate_url_host("https://www.rf4-stat.ru/posts/") == "rf4-stat.ru"
|
|
assert _validate_url_host("https://rf4map.ru/point/123") == "rf4map.ru"
|
|
|
|
# Disallowed hosts raise ValueError before network I/O
|
|
with pytest.raises(ValueError, match="not in allowlist"):
|
|
_validate_url_host("https://169.254.169.254/latest/meta-data/")
|
|
with pytest.raises(ValueError, match="not in allowlist"):
|
|
_validate_url_host("https://internal-service.corp/api")
|
|
# Port validation also works
|
|
with pytest.raises(ValueError, match="not in allowed ports"):
|
|
_validate_url_host("https://download.rf4db.com:9999/admin")
|
|
|
|
|
|
def test_validate_url_host_rejects_missing_hostname() -> None:
|
|
from rf4_research.community_cli import _validate_url_host
|
|
|
|
with pytest.raises(ValueError, match="valid hostname"):
|
|
_validate_url_host("https://")
|
|
with pytest.raises(ValueError, match="scheme.*not allowed"):
|
|
_validate_url_host("ftp://rf4db.com/file")
|
|
with pytest.raises(ValueError, match="HTTPS"):
|
|
_validate_url_host("http://rf4db.com/file")
|
|
|
|
|
|
def test_write_state_is_atomic_with_flush(tmp_path: Path) -> None:
|
|
"""A02: _write_state uses exclusive lock, flush, and atomic rename."""
|
|
from rf4_research.community_cli import _read_state, _write_state
|
|
|
|
state_file = tmp_path / "state.json"
|
|
_write_state(state_file, {"key1": "value1"})
|
|
assert state_file.exists()
|
|
assert not (state_file.with_suffix(".tmp")).exists()
|
|
state = _read_state(state_file)
|
|
assert state == {"key1": "value1"}
|
|
|
|
_write_state(state_file, {"key1": "value2", "key2": "value3"})
|
|
state = _read_state(state_file)
|
|
assert state == {"key1": "value2", "key2": "value3"}
|
|
assert not (state_file.with_suffix(".tmp")).exists()
|