mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-07-04 13:23:18 +02:00
Krzysztof kuhy Rudnicki
b8bd8459e4
- Extract count_comp from detail page in _apply_detail_to_extras so the all-playstyles completion count is populated even when the search API returns 0 (Mini Ghost: 0 → 69, now passes confidence thresholds) - Fix _refresh_candidate_confidence to trigger re-fetch when count_comp==0 even if comp_100_count>0 (was silently skipping stale partial entries) - Filter colon-stripped fallback candidates (e.g. "Vox Populi" from "Vox Populi: Poland 2023") to full-edition or exact matches only, preventing cross-franchise false positives - Demote "All N ProtonDB ratings found in cache" log to DEBUG to remove per-game noise from the scan output Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
349 lines
11 KiB
Python
349 lines
11 KiB
Python
"""Tests for HLTB search, batch-fetch, and page parsing — part 2."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import TYPE_CHECKING, Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import aiohttp
|
|
from typing_extensions import Self
|
|
|
|
from steam_backlog_enforcer._hltb_search import (
|
|
_fetch_batch,
|
|
_search_one,
|
|
_SearchCtx,
|
|
)
|
|
from steam_backlog_enforcer._hltb_types import (
|
|
_SAVE_INTERVAL,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable
|
|
|
|
|
|
class _FakeResponse:
|
|
"""Async context manager mimicking aiohttp response."""
|
|
|
|
def __init__(self, status: int, json_data: dict[str, Any] | None = None) -> None:
|
|
self.status = status
|
|
self._json_data = json_data or {}
|
|
|
|
async def __aenter__(self) -> Self:
|
|
return self
|
|
|
|
async def __aexit__(self, *args: object) -> None:
|
|
pass
|
|
|
|
async def json(self) -> dict[str, Any]:
|
|
return self._json_data
|
|
|
|
|
|
def _make_session(resp: _FakeResponse) -> MagicMock:
|
|
session = MagicMock()
|
|
session.post.return_value = resp
|
|
return session
|
|
|
|
|
|
def _make_ctx(
|
|
session: MagicMock,
|
|
*,
|
|
cache: dict[int, float] | None = None,
|
|
progress_cb: Callable[..., object] | None = None,
|
|
) -> _SearchCtx:
|
|
return _SearchCtx(
|
|
session=session,
|
|
search_url="https://example.com/search",
|
|
headers={},
|
|
cache=cache if cache is not None else {},
|
|
counter={"done": 0, "found": 0},
|
|
total=1,
|
|
progress_cb=progress_cb,
|
|
)
|
|
|
|
|
|
class TestSearchOne:
|
|
"""Tests for _search_one."""
|
|
|
|
def test_found(self) -> None:
|
|
resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "TF2",
|
|
"game_alias": "",
|
|
"comp_100": 180000,
|
|
"game_id": 12345,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is not None
|
|
assert result.app_id == 440
|
|
|
|
def test_found_without_game_id(self) -> None:
|
|
"""Found result with hltb_game_id=0 does not populate ctx.hltb_game_id."""
|
|
resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "TF2",
|
|
"game_alias": "",
|
|
"comp_100": 180000,
|
|
"game_id": 0,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is not None
|
|
assert 440 not in ctx.hltb_game_id
|
|
|
|
def test_not_found(self) -> None:
|
|
resp = _FakeResponse(200, {"data": []})
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is None
|
|
assert ctx.cache[440] == -1
|
|
|
|
def test_error(self) -> None:
|
|
session = MagicMock()
|
|
session.post.side_effect = aiohttp.ClientError("fail")
|
|
ctx = _make_ctx(session)
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is None
|
|
|
|
def test_non_200(self) -> None:
|
|
resp = _FakeResponse(500)
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is None
|
|
|
|
def test_fallback_name_without_year_suffix(self) -> None:
|
|
session = MagicMock()
|
|
session.post.side_effect = [
|
|
_FakeResponse(200, {"data": []}),
|
|
_FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "Final Fantasy VII",
|
|
"game_alias": "",
|
|
"game_type": "game",
|
|
"comp_100": 141120,
|
|
"game_id": 435,
|
|
"comp_100_count": 746,
|
|
"count_comp": 10450,
|
|
}
|
|
]
|
|
},
|
|
),
|
|
]
|
|
ctx = _make_ctx(session)
|
|
result = asyncio.run(
|
|
_search_one(asyncio.Semaphore(1), ctx, 39140, "Final Fantasy VII (2013)")
|
|
)
|
|
assert result is not None
|
|
assert result.app_id == 39140
|
|
assert result.comp_100_count == 746
|
|
assert result.count_comp == 10450
|
|
assert session.post.call_count == 2
|
|
|
|
def test_with_progress_cb(self) -> None:
|
|
resp = _FakeResponse(200, {"data": []})
|
|
cb = MagicMock()
|
|
ctx = _make_ctx(_make_session(resp), progress_cb=cb)
|
|
asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
cb.assert_called_once()
|
|
|
|
def test_low_similarity_skipped(self) -> None:
|
|
resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "Completely Different Name",
|
|
"game_alias": "",
|
|
"comp_100": 3600,
|
|
"game_id": 1,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is None
|
|
|
|
def test_zero_comp_100_skipped(self) -> None:
|
|
resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "TF2",
|
|
"game_alias": "",
|
|
"comp_100": 0,
|
|
"game_id": 1,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is None
|
|
|
|
def test_alias_match(self) -> None:
|
|
resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "Team Fortress 2",
|
|
"game_alias": "TF2",
|
|
"comp_100": 180000,
|
|
"game_id": 12345,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is not None
|
|
|
|
def test_full_edition_colon(self) -> None:
|
|
resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "TF2: Complete",
|
|
"game_alias": "",
|
|
"comp_100": 180000,
|
|
"game_id": 99,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is not None
|
|
|
|
def test_full_edition_dash(self) -> None:
|
|
resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "TF2 - Complete",
|
|
"game_alias": "",
|
|
"comp_100": 180000,
|
|
"game_id": 99,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
ctx = _make_ctx(_make_session(resp))
|
|
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
assert result is not None
|
|
|
|
def test_save_interval(self) -> None:
|
|
"""Trigger the _SAVE_INTERVAL branch."""
|
|
resp = _FakeResponse(200, {"data": []})
|
|
ctx = _make_ctx(_make_session(resp))
|
|
# Set done to one less than _SAVE_INTERVAL so it triggers save
|
|
|
|
ctx.counter["done"] = _SAVE_INTERVAL - 1
|
|
with patch("steam_backlog_enforcer._hltb_search.save_hltb_cache") as mock_save:
|
|
asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
|
|
mock_save.assert_called_once()
|
|
|
|
def test_colon_strip_fallback_rejects_cross_franchise_match(self) -> None:
|
|
"""Colon-stripped fallback must not match a different franchise loosely.
|
|
|
|
"Vox Populi: Poland 2023" stripped to "Vox Populi" should NOT match
|
|
"Vox Populi Vox Dei 2" (different game, low-similarity entry).
|
|
"""
|
|
empty_resp = _FakeResponse(200, {"data": []})
|
|
loose_resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "Vox Populi Vox Dei 2",
|
|
"game_alias": "",
|
|
"game_type": "game",
|
|
"comp_100": 14400,
|
|
"comp_100_count": 9,
|
|
"count_comp": 57,
|
|
"game_id": 99999,
|
|
}
|
|
]
|
|
},
|
|
)
|
|
session = MagicMock()
|
|
session.post.side_effect = [empty_resp, loose_resp]
|
|
ctx = _make_ctx(session)
|
|
result = asyncio.run(
|
|
_search_one(asyncio.Semaphore(1), ctx, 2590810, "Vox Populi: Poland 2023")
|
|
)
|
|
assert result is None
|
|
|
|
def test_colon_strip_fallback_accepts_full_edition(self) -> None:
|
|
"""Colon-stripped fallback must still match when the HLTB entry is a
|
|
full edition of the stripped name (name starts with stripped + ':').
|
|
"""
|
|
empty_resp = _FakeResponse(200, {"data": []})
|
|
full_edition_resp = _FakeResponse(
|
|
200,
|
|
{
|
|
"data": [
|
|
{
|
|
"game_name": "Batman: Arkham Asylum",
|
|
"game_alias": "",
|
|
"game_type": "game",
|
|
"comp_100": 144000,
|
|
"comp_100_count": 300,
|
|
"count_comp": 5000,
|
|
"game_id": 11111,
|
|
}
|
|
]
|
|
},
|
|
)
|
|
session = MagicMock()
|
|
session.post.side_effect = [empty_resp, full_edition_resp]
|
|
ctx = _make_ctx(session)
|
|
result = asyncio.run(
|
|
_search_one(asyncio.Semaphore(1), ctx, 35140, "Batman: Arkham Asylum")
|
|
)
|
|
assert result is not None
|
|
assert result.game_name == "Batman: Arkham Asylum"
|
|
|
|
|
|
class TestFetchBatchHltb:
|
|
"""Tests for _fetch_batch (the hltb version)."""
|
|
|
|
def test_no_auth(self) -> None:
|
|
with (
|
|
patch(
|
|
"steam_backlog_enforcer._hltb_search._get_hltb_search_url",
|
|
return_value="https://example.com",
|
|
),
|
|
patch(
|
|
"steam_backlog_enforcer._hltb_search._get_auth_info",
|
|
new_callable=AsyncMock,
|
|
return_value=None,
|
|
),
|
|
):
|
|
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, {}, None))
|
|
assert results == []
|
|
|
|
|
|
class TestPickBestEntry:
|
|
"""Tests for exact-vs-extended entry choice logic."""
|