mirror of
https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git
synced 2026-07-09 12:41:05 +02:00
## ℹ️ Description - Link to the related issue(s): PR #1193 follow-up - Adds screen-aware humanization viewport selection so randomized desktop sizes fit the available browser-reported screen area while preserving permissive config validation. ## 📋 Changes Summary - Add larger common desktop viewport defaults: `2560x1440`, `1920x1200`, `1728x1117`, and `1512x982`. - Probe browser-reported CSS-pixel screen metrics at launch time using `window.screen.availWidth` / `window.screen.availHeight`. - Filter configured viewport base sizes against available screen dimensions, apply bounded jitter, and preserve explicit `--window-size` overrides. - Omit automatic `--window-size` when no configured size fits, screen metrics are unavailable, the launch is headless, or no real display/window manager is available; explicit user `--window-size` remains preserved. - Cache successful probe metrics and bound the probe with the `chrome_remote_probe` timeout. - Update generated default config and German translations. - Add unit coverage and real-browser integration coverage for viewport probing/filtering, headless/no-display skips, probe timeout/cache behavior, and `create_browser_session()`. ### ⚙️ Type of Change Select the type(s) of change(s) included in this pull request: - [ ] 🐞 Bug fix (non-breaking change which fixes an issue) - [x] ✨ New feature (adds new functionality without breaking existing usage) - [ ] 💥 Breaking change (changes that might break existing user setups, scripts, or configurations) ## ✅ Checklist Before requesting a review, confirm the following: - [x] I have reviewed my changes to ensure they meet the project's standards. - [x] I have tested my changes and ensured that all tests pass (`pdm run test`). - [x] I have formatted the code (`pdm run format`). - [x] I have verified that linting passes (`pdm run lint`). - [x] I have updated documentation where necessary. By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Browser sessions now support screen-aware viewport selection automatically when viewport randomization is enabled. * Expanded the default set of candidate viewport sizes used for randomized sizing. * **Bug Fixes** * Viewport sizing now filters out candidates that don’t fit within the available screen area. * If a window size is already specified, it’s preserved. * **Tests** * Added/expanded slow integration tests for screen probing and viewport fitting. * Expanded unit tests for viewport filtering, jittering, fallback behavior, and probe reuse/timeout. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
642 lines
27 KiB
Python
642 lines
27 KiB
Python
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
|
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
|
|
"""Tests for the human-like interaction behavior in WebScrapingMixin (bot-detection evasion)."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from kleinanzeigen_bot.model.config_model import Config, HumanizationConfig
|
|
from kleinanzeigen_bot.utils.web_scraping_mixin import (
|
|
By,
|
|
Element,
|
|
WebScrapingMixin,
|
|
_filter_viewport_sizes, # noqa: PLC2701 # type: ignore[attr-defined]
|
|
_jitter_viewport, # noqa: PLC2701 # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
def make_scraper(humanization:HumanizationConfig | None = None) -> WebScrapingMixin:
|
|
scraper = WebScrapingMixin()
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "user@example.com", "password": "secret"}, # noqa: S106
|
|
"humanization": (humanization or HumanizationConfig()).model_dump(),
|
|
})
|
|
return scraper
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# web_click: mouse-based clicking with fallback
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_click_uses_humanized_path_when_enabled() -> None:
|
|
scraper = make_scraper(HumanizationConfig(mouse_movement = True))
|
|
elem = AsyncMock(spec = Element)
|
|
with (
|
|
patch.object(scraper, "web_find", new_callable = AsyncMock, return_value = elem),
|
|
patch.object(scraper, "_humanized_click", new_callable = AsyncMock) as humanized,
|
|
patch.object(scraper, "web_sleep", new_callable = AsyncMock),
|
|
):
|
|
result = await scraper.web_click(By.ID, "x")
|
|
|
|
humanized.assert_awaited_once_with(elem)
|
|
elem.click.assert_not_awaited()
|
|
assert result is elem
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_click_falls_back_to_plain_click_on_failure() -> None:
|
|
scraper = make_scraper(HumanizationConfig(mouse_movement = True))
|
|
elem = AsyncMock(spec = Element)
|
|
with (
|
|
patch.object(scraper, "web_find", new_callable = AsyncMock, return_value = elem),
|
|
patch.object(scraper, "_humanized_click", new_callable = AsyncMock, side_effect = RuntimeError("no geometry")),
|
|
patch.object(scraper, "web_sleep", new_callable = AsyncMock),
|
|
):
|
|
await scraper.web_click(By.ID, "x")
|
|
|
|
elem.click.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_click_uses_plain_click_when_mouse_movement_disabled() -> None:
|
|
scraper = make_scraper(HumanizationConfig(mouse_movement = False))
|
|
elem = AsyncMock(spec = Element)
|
|
with (
|
|
patch.object(scraper, "web_find", new_callable = AsyncMock, return_value = elem),
|
|
patch.object(scraper, "_humanized_click", new_callable = AsyncMock) as humanized,
|
|
patch.object(scraper, "web_sleep", new_callable = AsyncMock),
|
|
):
|
|
await scraper.web_click(By.ID, "x")
|
|
|
|
humanized.assert_not_awaited()
|
|
elem.click.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_humanized_click_dispatches_mouse_events() -> None:
|
|
scraper = make_scraper()
|
|
tab = AsyncMock()
|
|
elem = AsyncMock(spec = Element)
|
|
elem.scroll_into_view = AsyncMock()
|
|
elem.get_position = AsyncMock(return_value = SimpleNamespace(center = (100.0, 200.0)))
|
|
elem._tab = tab
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.cdp_input.dispatch_mouse_event") as dispatch,
|
|
):
|
|
await scraper._humanized_click(elem)
|
|
|
|
# at least the moves plus a press and a release were dispatched
|
|
assert tab.send.await_count >= 3
|
|
dispatched_types = [call.args[0] for call in dispatch.call_args_list]
|
|
assert "mouseMoved" in dispatched_types
|
|
assert "mousePressed" in dispatched_types
|
|
assert "mouseReleased" in dispatched_types
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# web_input: typing jitter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_input_types_per_character_when_jitter_enabled() -> None:
|
|
scraper = make_scraper(HumanizationConfig(typing_delay_min_ms = 0, typing_delay_max_ms = 0))
|
|
field = AsyncMock(spec = Element)
|
|
with (
|
|
patch.object(scraper, "web_find", new_callable = AsyncMock, return_value = field),
|
|
patch.object(scraper, "_clear_input", new_callable = AsyncMock),
|
|
patch.object(scraper, "web_sleep", new_callable = AsyncMock),
|
|
):
|
|
await scraper.web_input(By.ID, "x", "abc")
|
|
|
|
assert field.send_keys.await_count == 3
|
|
assert [call.args[0] for call in field.send_keys.await_args_list] == ["a", "b", "c"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_input_single_burst_when_jitter_disabled() -> None:
|
|
scraper = make_scraper(HumanizationConfig(typing_jitter = False))
|
|
field = AsyncMock(spec = Element)
|
|
with (
|
|
patch.object(scraper, "web_find", new_callable = AsyncMock, return_value = field),
|
|
patch.object(scraper, "_clear_input", new_callable = AsyncMock),
|
|
patch.object(scraper, "web_sleep", new_callable = AsyncMock),
|
|
):
|
|
await scraper.web_input(By.ID, "x", "abc")
|
|
|
|
field.send_keys.assert_awaited_once_with("abc")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# idle actions + thinking pauses
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_random_human_actions_noop_when_disabled() -> None:
|
|
scraper = make_scraper(HumanizationConfig(enabled = False))
|
|
with (
|
|
patch.object(scraper, "_idle_scroll", new_callable = AsyncMock) as scroll,
|
|
patch.object(scraper, "_idle_mouse_wiggle", new_callable = AsyncMock) as wiggle,
|
|
patch.object(scraper, "web_think", new_callable = AsyncMock) as think,
|
|
):
|
|
await scraper.perform_random_human_actions()
|
|
|
|
scroll.assert_not_awaited()
|
|
wiggle.assert_not_awaited()
|
|
think.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_random_human_actions_runs_subset_when_gate_fires() -> None:
|
|
scraper = make_scraper(HumanizationConfig(idle_action_probability = 1.0))
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.random", return_value = 0.0),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.shuffle"),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.randint", return_value = 1),
|
|
patch.object(scraper, "_idle_scroll", new_callable = AsyncMock) as scroll,
|
|
patch.object(scraper, "_idle_mouse_wiggle", new_callable = AsyncMock) as wiggle,
|
|
patch.object(scraper, "web_think", new_callable = AsyncMock) as think,
|
|
):
|
|
await scraper.perform_random_human_actions()
|
|
|
|
# exactly one action runs (subset size forced to 1, shuffle disabled -> first action)
|
|
assert scroll.await_count + wiggle.await_count + think.await_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_random_human_actions_swallows_action_errors() -> None:
|
|
scraper = make_scraper(HumanizationConfig(idle_action_probability = 1.0))
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.random", return_value = 0.0),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.shuffle"),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.randint", return_value = 3),
|
|
patch.object(scraper, "_idle_scroll", new_callable = AsyncMock, side_effect = RuntimeError("boom")),
|
|
patch.object(scraper, "_idle_mouse_wiggle", new_callable = AsyncMock),
|
|
patch.object(scraper, "web_think", new_callable = AsyncMock),
|
|
):
|
|
# must not raise even though the first action fails
|
|
await scraper.perform_random_human_actions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_think_pauses_only_within_probability() -> None:
|
|
scraper = make_scraper(HumanizationConfig(long_pause_probability = 0.1))
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.random", return_value = 0.05),
|
|
patch.object(scraper, "web_sleep", new_callable = AsyncMock) as sleep,
|
|
):
|
|
await scraper.web_think()
|
|
sleep.assert_awaited_once()
|
|
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.random", return_value = 0.5),
|
|
patch.object(scraper, "web_sleep", new_callable = AsyncMock) as sleep2,
|
|
):
|
|
await scraper.web_think()
|
|
sleep2.assert_not_awaited()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# web_sleep configurable band
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_sleep_uses_configured_band() -> None:
|
|
scraper = make_scraper(HumanizationConfig(action_delay_min_ms = 10, action_delay_max_ms = 11))
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock) as sleep:
|
|
await scraper.web_sleep()
|
|
# 10 ms lower bound -> 0.010 s
|
|
assert sleep.await_args is not None
|
|
slept_seconds = sleep.await_args.args[0]
|
|
assert 0.010 <= slept_seconds <= 0.011
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_sleep_respects_explicit_bounds() -> None:
|
|
scraper = make_scraper()
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock) as sleep:
|
|
await scraper.web_sleep(50, 51)
|
|
assert sleep.await_args is not None
|
|
assert 0.050 <= sleep.await_args.args[0] <= 0.051
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_sleep_with_max_only_respects_hard_cap() -> None:
|
|
"""web_sleep(max_ms=500) must never sleep above 0.5 s."""
|
|
scraper = make_scraper()
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock) as sleep:
|
|
await scraper.web_sleep(max_ms = 500)
|
|
assert sleep.await_args is not None
|
|
assert 0 <= sleep.await_args.args[0] <= 0.5
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_sleep_with_min_only_is_fixed_delay() -> None:
|
|
"""web_sleep(min_ms=500) with omitted max resolves to exactly 0.5 s."""
|
|
scraper = make_scraper()
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock) as sleep:
|
|
await scraper.web_sleep(min_ms = 500)
|
|
assert sleep.await_args is not None
|
|
assert sleep.await_args.args[0] == 0.5
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# viewport randomization at launch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_viewport_arg_appended_when_selected() -> None:
|
|
"""selected_viewport_size parameter is honoured by _build_new_browser_launch_args."""
|
|
scraper = make_scraper(HumanizationConfig(randomize_viewport = True, viewport_sizes = ["1600x900"]))
|
|
args, _ = scraper._build_new_browser_launch_args(selected_viewport_size = "1600x900")
|
|
assert "--window-size=1600,900" in args
|
|
|
|
|
|
def test_viewport_arg_not_appended_when_unselected() -> None:
|
|
"""_build_new_browser_launch_args with no selected_viewport_size must not add --window-size."""
|
|
scraper = make_scraper(HumanizationConfig(randomize_viewport = True, viewport_sizes = ["1600x900"]))
|
|
args, _ = scraper._build_new_browser_launch_args()
|
|
assert not any(a.startswith("--window-size") for a in args)
|
|
|
|
|
|
def test_viewport_arg_not_appended_when_user_supplied() -> None:
|
|
scraper = make_scraper(HumanizationConfig(randomize_viewport = True, viewport_sizes = ["1600x900"]))
|
|
scraper.browser_config.arguments = ["--window-size=800,600"]
|
|
args, _ = scraper._build_new_browser_launch_args()
|
|
assert "--window-size=1600,900" not in args
|
|
assert "--window-size=800,600" in args
|
|
|
|
|
|
def test_viewport_arg_not_appended_when_disabled() -> None:
|
|
scraper = make_scraper(HumanizationConfig(randomize_viewport = False))
|
|
args, _ = scraper._build_new_browser_launch_args()
|
|
assert not any(arg.startswith("--window-size") for arg in args)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# web_sleep: zero-delay range does not raise
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_sleep_zero_delay() -> None:
|
|
"""web_sleep(0, 0) must not raise and must sleep 0 seconds."""
|
|
scraper = make_scraper()
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock) as sleep:
|
|
await scraper.web_sleep(0, 0)
|
|
assert sleep.await_count == 1
|
|
assert sleep.await_args is not None
|
|
assert sleep.await_args.args[0] == 0.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_sleep_default_zero_band() -> None:
|
|
"""web_sleep() without args with config band (0, 0) must not raise and sleep 0."""
|
|
scraper = make_scraper(HumanizationConfig(action_delay_min_ms = 0, action_delay_max_ms = 0))
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock) as sleep:
|
|
await scraper.web_sleep()
|
|
assert sleep.await_args is not None
|
|
assert sleep.await_args.args[0] == 0.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HumanizationConfig validators
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_invalid_viewport_format_raises_error() -> None:
|
|
with pytest.raises(ValueError, match = "Invalid viewport size"):
|
|
HumanizationConfig(viewport_sizes = ["invalid"])
|
|
|
|
with pytest.raises(ValueError, match = "Invalid viewport size"):
|
|
HumanizationConfig(viewport_sizes = ["1920x1080x720"])
|
|
|
|
with pytest.raises(ValueError, match = "Invalid viewport size"):
|
|
HumanizationConfig(viewport_sizes = ["abcxdef"])
|
|
|
|
with pytest.raises(ValueError, match = "Invalid viewport size"):
|
|
HumanizationConfig(viewport_sizes = ["0x0"])
|
|
|
|
with pytest.raises(ValueError, match = "Invalid viewport size"):
|
|
HumanizationConfig(viewport_sizes = ["0x720"])
|
|
|
|
with pytest.raises(ValueError, match = "Invalid viewport size"):
|
|
HumanizationConfig(viewport_sizes = ["1920x0"])
|
|
|
|
|
|
def test_reversed_min_max_raises_error() -> None:
|
|
with pytest.raises(ValueError, match = "must be >="):
|
|
HumanizationConfig(typing_delay_min_ms = 100, typing_delay_max_ms = 50)
|
|
|
|
with pytest.raises(ValueError, match = "must be >="):
|
|
HumanizationConfig(action_delay_min_ms = 3_000, action_delay_max_ms = 1_000)
|
|
|
|
with pytest.raises(ValueError, match = "must be >="):
|
|
HumanizationConfig(long_pause_min_ms = 5_000, long_pause_max_ms = 2_000)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _humanized_type fallback on per-character failure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_humanized_type_fallback_clears_and_sends_full_text() -> None:
|
|
"""When per-character typing fails, fallback clears and sends full text once."""
|
|
scraper = make_scraper(HumanizationConfig(typing_delay_min_ms = 0, typing_delay_max_ms = 0))
|
|
field = AsyncMock(spec = Element)
|
|
field.send_keys = AsyncMock(side_effect = [None, RuntimeError("cdp fail"), None])
|
|
|
|
with (
|
|
patch.object(scraper, "_clear_input", new_callable = AsyncMock) as clear_mock,
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock),
|
|
):
|
|
await scraper._humanized_type(field, "abc")
|
|
|
|
# _clear_input was called in the fallback path
|
|
clear_mock.assert_awaited_once_with(field)
|
|
# send_keys: "a" (ok), "b" (fails), then full "abc" after clear
|
|
assert field.send_keys.await_count == 3
|
|
assert field.send_keys.await_args_list[0].args[0] == "a"
|
|
assert field.send_keys.await_args_list[1].args[0] == "b"
|
|
assert field.send_keys.await_args_list[2].args[0] == "abc"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# viewport-size helper functions (_filter_viewport_sizes)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_filter_viewport_sizes_all_fit() -> None:
|
|
sizes = ["1920x1080", "1366x768", "2560x1440"]
|
|
assert _filter_viewport_sizes(sizes, 2560, 1440) == sizes
|
|
|
|
|
|
def test_filter_viewport_sizes_some_fit() -> None:
|
|
sizes = ["1920x1080", "2560x1440", "1366x768"]
|
|
result = _filter_viewport_sizes(sizes, 1920, 1080)
|
|
assert result == ["1920x1080", "1366x768"]
|
|
|
|
|
|
def test_filter_viewport_sizes_none_fit() -> None:
|
|
sizes = ["2560x1440", "3840x2160"]
|
|
assert _filter_viewport_sizes(sizes, 1920, 1080) == []
|
|
|
|
|
|
def test_filter_viewport_sizes_boundary_exact() -> None:
|
|
"""Sizes exactly equal to the available area must fit."""
|
|
assert _filter_viewport_sizes(["1920x1080"], 1920, 1080) == ["1920x1080"]
|
|
|
|
|
|
def test_filter_viewport_sizes_skips_parse_errors() -> None:
|
|
sizes = ["1920x1080", "not-valid", "1366x768"]
|
|
result = _filter_viewport_sizes(sizes, 1920, 1080)
|
|
assert result == ["1920x1080", "1366x768"]
|
|
|
|
|
|
def test_filter_viewport_sizes_empty_list() -> None:
|
|
assert _filter_viewport_sizes([], 1920, 1080) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _jitter_viewport
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_jitter_viewport_basic() -> None:
|
|
"""Jitter stays within ±24 width and ±16 height."""
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.randint", side_effect = [1024, 600]) as mock_rand:
|
|
jw, jh = _jitter_viewport(1024, 600, 2000, 1200)
|
|
assert jw == 1024
|
|
assert jh == 600
|
|
assert mock_rand.call_args_list[0].args == (max(1, 1024 - 24), min(2000, 1024 + 24))
|
|
assert mock_rand.call_args_list[1].args == (max(1, 600 - 16), min(1200, 600 + 16))
|
|
|
|
|
|
def test_jitter_viewport_caps_by_avail() -> None:
|
|
"""max_w / max_h are clamped to avail dimensions."""
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.randint", side_effect = [1900, 1050]) as mock_rand:
|
|
jw, jh = _jitter_viewport(1920, 1080, 1920, 1080)
|
|
assert jw == 1900
|
|
assert jh == 1050
|
|
# base_w + 24 = 1944 but avail_w caps at 1920
|
|
assert mock_rand.call_args_list[0].args == (max(1, 1920 - 24), 1920)
|
|
assert mock_rand.call_args_list[1].args == (max(1, 1080 - 16), 1080)
|
|
|
|
|
|
def test_jitter_viewport_floor_at_one() -> None:
|
|
"""min_w / min_h never drop below 1, even for tiny bases."""
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.randint", side_effect = [1, 1]) as mock_rand:
|
|
jw, jh = _jitter_viewport(10, 10, 1920, 1080)
|
|
assert jw == 1
|
|
assert jh == 1
|
|
# base_w - 24 = -14, clamped to 1
|
|
assert mock_rand.call_args_list[0].args == (1, 10 + 24)
|
|
assert mock_rand.call_args_list[1].args == (1, 10 + 16)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# integration-level: _select_viewport_size with jitter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_select_viewport_size_applies_jitter_when_metrics_known() -> None:
|
|
"""When metrics are available, a fitting base is jittered before returning."""
|
|
scraper = WebScrapingMixin()
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "u", "password": "p"}, # noqa: S106
|
|
"humanization": HumanizationConfig(randomize_viewport = True, viewport_sizes = ["1600x900", "1920x1080"]).model_dump(),
|
|
})
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._has_display_available", return_value = True),
|
|
patch.object(scraper, "_probe_screen_metrics", return_value = (1920, 1080)),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.choice", return_value = "1600x900"),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._rng.randint", side_effect = [1590, 890]),
|
|
):
|
|
result = await scraper._select_viewport_size()
|
|
# Jitter applied: base 1600x900 → ~1590x890
|
|
assert result == "1590x890"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_select_viewport_size_skips_headless_without_probe() -> None:
|
|
scraper = WebScrapingMixin()
|
|
scraper.browser_config.arguments = ["--headless=new"]
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "u", "password": "p"}, # noqa: S106
|
|
"humanization": HumanizationConfig(randomize_viewport = True, viewport_sizes = ["1600x900"]).model_dump(),
|
|
})
|
|
with patch.object(scraper, "_probe_screen_metrics", new_callable = AsyncMock) as probe:
|
|
result = await scraper._select_viewport_size()
|
|
probe.assert_not_awaited()
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_select_viewport_size_skips_no_display_on_linux_without_probe() -> None:
|
|
scraper = WebScrapingMixin()
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "u", "password": "p"}, # noqa: S106
|
|
"humanization": HumanizationConfig(randomize_viewport = True, viewport_sizes = ["1600x900"]).model_dump(),
|
|
})
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.platform.system", return_value = "Linux"),
|
|
patch.dict("kleinanzeigen_bot.utils.web_scraping_mixin.os.environ", {}, clear = True),
|
|
patch.object(scraper, "_probe_screen_metrics", new_callable = AsyncMock) as probe,
|
|
):
|
|
result = await scraper._select_viewport_size()
|
|
probe.assert_not_awaited()
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_select_viewport_size_skips_ci_without_display_without_probe() -> None:
|
|
scraper = WebScrapingMixin()
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "u", "password": "p"}, # noqa: S106
|
|
"humanization": HumanizationConfig(randomize_viewport = True, viewport_sizes = ["1600x900"]).model_dump(),
|
|
})
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.platform.system", return_value = "Darwin"),
|
|
patch.dict("kleinanzeigen_bot.utils.web_scraping_mixin.os.environ", {"CI": "true"}, clear = True),
|
|
patch.object(scraper, "_probe_screen_metrics", new_callable = AsyncMock) as probe,
|
|
):
|
|
result = await scraper._select_viewport_size()
|
|
probe.assert_not_awaited()
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_select_viewport_size_omits_when_none_fit() -> None:
|
|
"""When all sizes exceed avail, returns None regardless of jitter."""
|
|
scraper = WebScrapingMixin()
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "u", "password": "p"}, # noqa: S106
|
|
"humanization": HumanizationConfig(randomize_viewport = True, viewport_sizes = ["2560x1440"]).model_dump(),
|
|
})
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._has_display_available", return_value = True),
|
|
patch.object(scraper, "_probe_screen_metrics", return_value = (1920, 1080)),
|
|
):
|
|
result = await scraper._select_viewport_size()
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"metrics",
|
|
[
|
|
None,
|
|
(0, 1080),
|
|
(1920, 0),
|
|
(-1, 1080),
|
|
(1920, -1),
|
|
("1920", 1080),
|
|
(1920, "1080"),
|
|
(1920,),
|
|
(1920, 1080, 1),
|
|
({"availWidth": 1920},),
|
|
(True, 1080),
|
|
(float("nan"), 1080),
|
|
(float("inf"), 1080),
|
|
],
|
|
)
|
|
async def test_select_viewport_size_returns_none_for_invalid_metrics(metrics:object) -> None:
|
|
scraper = WebScrapingMixin()
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "u", "password": "p"}, # noqa: S106
|
|
"humanization": HumanizationConfig(randomize_viewport = True, viewport_sizes = ["2560x1440", "1366x768"]).model_dump(),
|
|
})
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin._has_display_available", return_value = True),
|
|
patch.object(scraper, "_probe_screen_metrics", return_value = metrics),
|
|
):
|
|
result = await scraper._select_viewport_size()
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_select_viewport_size_user_window_size_bypasses_selection() -> None:
|
|
"""User-supplied --window-size in browser_config.arguments prevents _select_viewport_size
|
|
from being called at all (tested via the guard in create_browser_session)."""
|
|
scraper = WebScrapingMixin()
|
|
scraper.browser_config.binary_location = "fake-browser"
|
|
scraper.browser_config.arguments = ["--window-size=800,600"]
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "u", "password": "p"}, # noqa: S106
|
|
"humanization": HumanizationConfig(randomize_viewport = True, viewport_sizes = ["1920x1080"]).model_dump(),
|
|
})
|
|
fake_browser = SimpleNamespace(websocket_url = "ws://test")
|
|
with (
|
|
patch.object(scraper, "_validate_chrome_version_configuration", new_callable = AsyncMock),
|
|
patch.object(scraper, "_select_viewport_size", new_callable = AsyncMock) as select_viewport,
|
|
patch.object(scraper, "_resolve_effective_user_data_dir", new_callable = AsyncMock, return_value = None),
|
|
patch.object(scraper, "_add_browser_extensions", new_callable = AsyncMock),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.files.exists", new_callable = AsyncMock, return_value = True),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.nodriver.start", new_callable = AsyncMock, return_value = fake_browser),
|
|
):
|
|
await scraper.create_browser_session()
|
|
select_viewport.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_screen_metrics_times_out() -> None:
|
|
scraper = WebScrapingMixin()
|
|
scraper.browser_config.binary_location = "fake-browser"
|
|
scraper.config = Config.model_validate({
|
|
"login": {"username": "u", "password": "p"}, # noqa: S106
|
|
"timeouts": {"chrome_remote_probe": 0.1},
|
|
})
|
|
|
|
async def slow_start(_cfg:object) -> object:
|
|
await asyncio.sleep(1)
|
|
return SimpleNamespace()
|
|
|
|
with patch("kleinanzeigen_bot.utils.web_scraping_mixin.nodriver.start", side_effect = slow_start):
|
|
result = await scraper._probe_screen_metrics()
|
|
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_screen_metrics_reuses_successful_probe() -> None:
|
|
scraper = WebScrapingMixin()
|
|
scraper.browser_config.binary_location = "fake-browser"
|
|
page = SimpleNamespace(evaluate = AsyncMock(return_value = {"availWidth": 111, "availHeight": 222}))
|
|
browser = SimpleNamespace(get = AsyncMock(return_value = page), stop = lambda: None)
|
|
|
|
with (
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.asyncio.sleep", new_callable = AsyncMock),
|
|
patch("kleinanzeigen_bot.utils.web_scraping_mixin.nodriver.start", new_callable = AsyncMock, return_value = browser) as start,
|
|
):
|
|
first = await scraper._probe_screen_metrics()
|
|
second = await scraper._probe_screen_metrics()
|
|
|
|
assert first == (111, 222)
|
|
assert second == (111, 222)
|
|
start.assert_awaited_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# default viewport sizes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_default_viewport_sizes_contain_new_entries() -> None:
|
|
cfg = HumanizationConfig()
|
|
expected_extras = {"2560x1440", "1920x1200", "1728x1117", "1512x982"}
|
|
for expected in expected_extras:
|
|
assert expected in cfg.viewport_sizes, f"Missing default viewport: {expected}"
|
|
|
|
|
|
def test_default_viewport_sizes_omits_4k() -> None:
|
|
cfg = HumanizationConfig()
|
|
assert "3840x2160" not in cfg.viewport_sizes
|
|
|
|
|
|
def test_default_viewport_sizes_count() -> None:
|
|
cfg = HumanizationConfig()
|
|
assert len(cfg.viewport_sizes) == 10
|