rf4-spotter/tests/test_community_cli.py

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:
    """A02: check_and_reserve is atomic — one reservation per interval per source."""
    from rf4_research.community_cli import check_and_reserve

    state_file = tmp_path / "fetch-state.json"
    check_and_reserve("rf4map-point", state_file=state_file, now=1_000)

    # Second call for same source within interval should fail
    with pytest.raises(RuntimeError, match="retry in 1800 seconds"):
        check_and_reserve("rf4map-point", state_file=state_file, now=1_000)
    # Different source should succeed
    check_and_reserve("rf4posts-spot", state_file=state_file, now=1_000)
    # After interval expires, should succeed again
    check_and_reserve("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()


def test_check_and_reserve_allows_only_one_per_interval(tmp_path: Path) -> None:
    """A02: Atomic check-and-reserve — only one success per interval."""
    from rf4_research.community_cli import check_and_reserve

    state_file = tmp_path / "state.json"
    base_time = 1000.0

    # First call should succeed
    check_and_reserve("test-source", state_file=state_file, now=base_time)
    state = json.loads(state_file.read_text(encoding="utf-8"))
    assert state["test-source"] == base_time

    # Second call within interval should fail
    with pytest.raises(RuntimeError, match="retry in 1799"):
        check_and_reserve("test-source", state_file=state_file, now=base_time + 1)

    # After interval expires, should succeed again
    check_and_reserve("test-source", state_file=state_file, now=base_time + 1800)


def _try_reserve_for_test(args: tuple) -> tuple:
    """Helper for multiprocessing — must be at module level."""
    pid, state_file_str, results_list = args
    from rf4_research.community_cli import check_and_reserve
    from pathlib import Path
    try:
        check_and_reserve("shared-source", state_file=Path(state_file_str), now=1000.0)
        results_list.append((pid, "ok"))
    except RuntimeError as e:
        results_list.append((pid, str(e)))
    return (pid, "ok")


def test_check_and_reserve_atomic_under_concurrent_access(tmp_path: Path) -> None:
    """A02: Real multi-process test — concurrent processes get at most one reservation."""
    import multiprocessing

    state_file = tmp_path / "concurrent.json"
    results = multiprocessing.Manager().list()

    # Launch 3 processes simultaneously
    processes = []
    for i in range(3):
        args = (i, str(state_file), results)
        p = multiprocessing.Process(target=_try_reserve_for_test, args=(args,))
        processes.append(p)
    for p in processes:
        p.start()
    for p in processes:
        p.join(timeout=10)

    # At most one should succeed
    ok_count = sum(1 for _, r in results if r == "ok")
    assert ok_count == 1, f"Expected exactly 1 ok, got {ok_count}: {results}"
    denied_count = sum(1 for _, r in results if "cooldown" in r)
    assert denied_count == 2, f"Expected 2 denied, got {denied_count}: {results}"