"""Tests for mcp.tools.scraping_tools.sound_stream module-level helpers.""" from __future__ import annotations import asyncio from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import pytest from playwright.async_api import Error as PlaywrightError import marketing_intelligence.mcp.tools.scraping_tools.base as base_module from marketing_intelligence.mcp.tools.scraping_tools.sound_stream import ( _chain_next_sound_videos, _click_next_in_sound_overlay, _find_next_sound_href, _process_sound_video, _return_to_sound_page, _skip_sound_positions, ) def _run(coro: Any) -> Any: return asyncio.run(coro) @pytest.fixture(autouse=True) def _clear_claimed() -> Any: base_module._claimed_video_ids.clear() yield base_module._claimed_video_ids.clear() def _make_page( *, url: str = "https://www.tiktok.com/@u/video/111", keyboard_raises: Exception | None = None, goto_raises: Exception | None = None, go_back_raises: Exception | None = None, wait_fn_raises: Exception | None = None, wait_sel_raises: Exception | None = None, evaluate_side_effects: list[Any] | None = None, content_html: str = "", screenshot_bytes: bytes = b"\xff\xd8\xff", next_btn_count: int = 1, ) -> MagicMock: page = MagicMock() page.url = url page.goto = AsyncMock(side_effect=goto_raises) page.go_back = AsyncMock(side_effect=go_back_raises) page.content = AsyncMock(return_value=content_html) page.screenshot = AsyncMock(return_value=screenshot_bytes) # keyboard page.keyboard = MagicMock() page.keyboard.press = AsyncMock(side_effect=keyboard_raises) # wait_for_function: can raise on different calls via side_effect list if wait_fn_raises is not None: page.wait_for_function = AsyncMock(side_effect=wait_fn_raises) else: page.wait_for_function = AsyncMock(return_value=None) page.wait_for_selector = AsyncMock(side_effect=wait_sel_raises) # locator for next button mock_loc = MagicMock() mock_loc.count = AsyncMock(return_value=next_btn_count) mock_loc.first = MagicMock() mock_loc.first.click = AsyncMock(return_value=None) page.locator = MagicMock(return_value=mock_loc) if evaluate_side_effects is not None: page.evaluate = AsyncMock(side_effect=evaluate_side_effects) else: page.evaluate = AsyncMock(return_value=[]) return page # ── _return_to_sound_page ──────────────────────────────────────────────────── class TestReturnToSoundPage: def test_escape_works_returns_true(self) -> None: page = _make_page() result = _run(_return_to_sound_page(page, "https://entry")) assert result is True def test_escape_raises_go_back_succeeds(self) -> None: # keyboard.press raises; go_back succeeds; wait_for_function succeeds call_count = {"n": 0} async def _wait_fn(js: str, timeout: int = 5000) -> None: call_count["n"] += 1 if call_count["n"] == 1: raise PlaywrightError("url still /video/") # subsequent calls succeed page = _make_page(keyboard_raises=PlaywrightError("esc failed")) page.wait_for_function = AsyncMock(side_effect=_wait_fn) result = _run(_return_to_sound_page(page, "https://entry")) assert result is True def test_all_fallbacks_fail_returns_false(self) -> None: page = _make_page( keyboard_raises=PlaywrightError("esc"), go_back_raises=PlaywrightError("no history"), goto_raises=asyncio.TimeoutError(), ) page.wait_for_function = AsyncMock(side_effect=PlaywrightError("still video")) result = _run(_return_to_sound_page(page, "https://entry")) assert result is False def test_wait_for_selector_fails_returns_false(self) -> None: page = _make_page(wait_sel_raises=PlaywrightError("blank")) result = _run(_return_to_sound_page(page, "https://entry")) assert result is False # ── _find_next_sound_href ──────────────────────────────────────────────────── class TestFindNextSoundHref: def test_returns_unseen_href(self) -> None: page = _make_page() page.evaluate = AsyncMock(return_value=["https://www.tiktok.com/@a/video/123"]) result = _run(_find_next_sound_href(page, set())) assert result == "https://www.tiktok.com/@a/video/123" def test_skips_seen_hrefs(self) -> None: page = _make_page() page.evaluate = AsyncMock(return_value=["https://www.tiktok.com/@a/video/123"]) result = _run(_find_next_sound_href(page, {"123"})) # all hrefs seen → returns None after 6 scroll attempts assert result is None def test_returns_none_when_no_hrefs(self) -> None: page = _make_page() page.evaluate = AsyncMock(return_value=[]) result = _run(_find_next_sound_href(page, set())) assert result is None # ── _click_next_in_sound_overlay ───────────────────────────────────────────── class TestClickNextInSoundOverlay: def test_button_present_returns_true(self) -> None: page = _make_page() with ( patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.click_and_await_comments", new_callable=AsyncMock, ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.wait_for_comments_stable", new_callable=AsyncMock, ), ): result = _run(_click_next_in_sound_overlay(page)) assert result is True def test_button_absent_returns_false(self) -> None: page = _make_page(next_btn_count=0) result = _run(_click_next_in_sound_overlay(page)) assert result is False def test_click_exception_returns_false(self) -> None: page = _make_page() with patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.click_and_await_comments", new_callable=AsyncMock, side_effect=PlaywrightError("nav failed"), ): result = _run(_click_next_in_sound_overlay(page)) assert result is False # ── _process_sound_video ───────────────────────────────────────────────────── def _make_anthropic_mock(text: str = "A singer on stage") -> MagicMock: mock_cls = MagicMock() instance = MagicMock() mock_cls.return_value = instance resp = MagicMock() tb = MagicMock() tb.text = text resp.content = [tb] instance.messages.create.return_value = resp return mock_cls class TestProcessSoundVideo: def _backend(self) -> MagicMock: b = MagicMock() b.write_video_comments = AsyncMock(return_value=None) return b def test_happy_path_returns_1_0_vid(self) -> None: page = _make_page( evaluate_side_effects=[ ["love this", "amazing"], {"caption": "cap", "hashtags": ["#pop"]}, ], ) backend = self._backend() with patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._Anthropic", _make_anthropic_mock(), ): ds, de, vid_out = _run( _process_sound_video( page, "v1", "url", "snd1", "snd1", "c", "r", "track", "gb", backend ) ) assert (ds, de, vid_out) == (1, 0, "v1") backend.write_video_comments.assert_awaited_once() def test_is_original_when_sound_ids_match(self) -> None: page = _make_page( evaluate_side_effects=[ ["comment"], {"caption": None, "hashtags": []}, ], ) backend = self._backend() with patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._Anthropic", _make_anthropic_mock("not_loaded"), ): _run( _process_sound_video( page, "v1", "url", "snd1", "snd1", "c", "r", None, None, backend ) ) stored_data = backend.write_video_comments.call_args[0][3] assert stored_data["is_original_sound"] is True def test_vision_failure_still_stores(self) -> None: from anthropic import APIError page = _make_page( evaluate_side_effects=[ ["comment"], {"caption": None, "hashtags": []}, ], ) backend = self._backend() mock_cls = MagicMock() mock_inst = MagicMock() mock_cls.return_value = mock_inst mock_inst.messages.create.side_effect = APIError( message="quota", request=MagicMock(), body=None ) with patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._Anthropic", mock_cls, ): ds, de, vid_out = _run( _process_sound_video( page, "v1", "url", "snd1", None, "c", "r", None, None, backend ) ) assert ds == 1 assert vid_out == "v1" def test_outer_exception_returns_0_1_none(self) -> None: page = _make_page() page.evaluate = AsyncMock(side_effect=RuntimeError("crash")) backend = self._backend() with patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.release_video_id", new_callable=AsyncMock, ) as mock_rel: ds, de, vid_out = _run( _process_sound_video( page, "v1", "url", "snd1", None, "c", "r", "track", None, backend ) ) assert (ds, de, vid_out) == (0, 1, None) mock_rel.assert_awaited_once_with("c", "r", "track", "v1") def test_caption_failure_still_stores(self) -> None: page = _make_page( evaluate_side_effects=[ ["one comment"], PlaywrightError("cap eval failed"), ], ) backend = self._backend() with patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._Anthropic", _make_anthropic_mock("not_loaded"), ): ds, de, vid_out = _run( _process_sound_video( page, "v2", "url", "s", None, "c", "r", None, None, backend ) ) assert ds == 1 assert vid_out == "v2" # ── _skip_sound_positions ──────────────────────────────────────────────────── class TestSkipSoundPositions: def test_n_zero_no_op(self) -> None: page = _make_page() seen: set[str] = set() _run(_skip_sound_positions(page, 0, seen)) assert seen == set() def test_marks_n_videos_as_seen(self) -> None: page = _make_page() page.evaluate = AsyncMock( return_value=[ "https://www.tiktok.com/@a/video/1", "https://www.tiktok.com/@b/video/2", "https://www.tiktok.com/@c/video/3", ] ) seen: set[str] = set() _run(_skip_sound_positions(page, 2, seen)) assert len(seen) == 2 def test_stops_when_enough_skipped(self) -> None: page = _make_page() hrefs = [f"https://www.tiktok.com/@u/video/{i}" for i in range(10)] page.evaluate = AsyncMock(return_value=hrefs) seen: set[str] = set() _run(_skip_sound_positions(page, 3, seen)) assert len(seen) == 3 # ── _chain_next_sound_videos ───────────────────────────────────────────────── class TestChainNextSoundVideos: def _common_patches( self, *, advanced: bool = True, next_vid: str | None = "222", claim_returns: bool = True, process_returns: tuple[int, int, str | None] = (1, 0, "222"), ) -> dict[str, Any]: return { "_click_next_in_sound_overlay": AsyncMock(return_value=advanced), "claim_video_id": AsyncMock(return_value=claim_returns), "_process_sound_video": AsyncMock(return_value=process_returns), "author_and_video_id": (None, next_vid), } def test_chains_one_video(self) -> None: page = _make_page(url="https://www.tiktok.com/@u/video/222") seen: set[str] = set() video_ids: list[str] = [] backend = MagicMock() with ( patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._click_next_in_sound_overlay", AsyncMock(side_effect=[True, False]), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.claim_video_id", AsyncMock(return_value=True), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._process_sound_video", AsyncMock(return_value=(1, 0, "222")), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.TikTokExtractor.author_and_video_id", return_value=(None, "222"), ), ): stored, errors = _run( _chain_next_sound_videos( page, "snd1", seen, 0, 0, video_ids, 2, None, "c", "r", None, None, backend, ) ) assert stored == 1 assert "222" in video_ids def test_stops_when_next_button_unavailable(self) -> None: page = _make_page() video_ids: list[str] = [] backend = MagicMock() with patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._click_next_in_sound_overlay", AsyncMock(return_value=False), ): stored, errors = _run( _chain_next_sound_videos( page, "snd1", set(), 0, 0, video_ids, 5, None, "c", "r", None, None, backend, ) ) assert stored == 0 def test_skips_claimed_video(self) -> None: page = _make_page(url="https://www.tiktok.com/@u/video/333") seen: set[str] = set() video_ids: list[str] = [] backend = MagicMock() with ( patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._click_next_in_sound_overlay", AsyncMock(side_effect=[True, False]), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.claim_video_id", AsyncMock(return_value=False), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.TikTokExtractor.author_and_video_id", return_value=(None, "333"), ), ): stored, errors = _run( _chain_next_sound_videos( page, "snd1", seen, 0, 0, video_ids, 5, None, "c", "r", None, None, backend, ) ) assert stored == 0 assert "333" in seen def test_breaks_after_three_consecutive_repeats(self) -> None: page = _make_page(url="https://www.tiktok.com/@u/video/444") video_ids: list[str] = [] backend = MagicMock() with ( patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._click_next_in_sound_overlay", AsyncMock(return_value=True), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.TikTokExtractor.author_and_video_id", return_value=(None, "444"), ), ): # "444" is already in seen → repeats until break stored, errors = _run( _chain_next_sound_videos( page, "snd1", {"444"}, 0, 0, video_ids, 5, None, "c", "r", None, None, backend, ) ) assert stored == 0 def test_stops_when_sample_size_reached(self) -> None: page = _make_page(url="https://www.tiktok.com/@u/video/555") seen: set[str] = set() video_ids: list[str] = [] backend = MagicMock() with ( patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._click_next_in_sound_overlay", AsyncMock(return_value=True), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.claim_video_id", AsyncMock(return_value=True), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream._process_sound_video", AsyncMock(return_value=(1, 0, "555")), ), patch( "marketing_intelligence.mcp.tools.scraping_tools.sound_stream.TikTokExtractor.author_and_video_id", return_value=(None, "555"), ), ): stored, errors = _run( _chain_next_sound_videos( page, "snd1", seen, 2, 0, video_ids, 2, # already at sample_size None, "c", "r", None, None, backend, ) ) # sample_size already reached, so loop shouldn't run assert stored == 2