"""Tests for mcp.tools.scraping_tools.sound_discovery helpers.""" from __future__ import annotations import asyncio import json from typing import Any from unittest.mock import ANY, 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_discovery import ( _check_video_for_sound, find_original_sound_id, ) 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 _html_with_blob(blob: dict[str, Any]) -> str: return ( f'' ) def _make_video_blob( *, title: str = "hateou", desc: str = "check out #hateou", challenges: list[str] | None = None, music_id: str | None = "999", ) -> dict[str, Any]: return { "__DEFAULT_SCOPE__": { "webapp.video-detail": { "itemInfo": { "itemStruct": { "desc": desc, "challenges": [{"title": c} for c in (challenges or [])], "music": { "id": music_id, "title": title, }, } } } } } def _make_page(html: str = "") -> MagicMock: page = MagicMock() page.goto = AsyncMock(return_value=None) page.content = AsyncMock(return_value=html) return page def _make_find_page(hrefs: list[str] | None = None) -> MagicMock: """Page mock for find_original_sound_id: goto + wait_for_selector + 3 scroll evaluates + hrefs evaluate.""" page = MagicMock() page.goto = AsyncMock(return_value=None) page.wait_for_selector = AsyncMock(return_value=None) page.evaluate = AsyncMock( side_effect=[None, None, None, hrefs if hrefs is not None else []] ) return page # ── _check_video_for_sound ──────────────────────────────────────────────────── class TestCheckVideoForSound: def test_match_via_desc_returns_sound_id(self) -> None: blob = _make_video_blob(desc="check out #hateou great song") page = _make_page(_html_with_blob(blob)) result = _run(_check_video_for_sound(page, "https://t.tt/@a/video/1", "hateou")) assert result == "999" def test_match_via_challenges_returns_sound_id(self) -> None: blob = _make_video_blob(desc="some other desc", challenges=["hateou"]) page = _make_page(_html_with_blob(blob)) result = _run(_check_video_for_sound(page, "https://t.tt/@a/video/2", "hateou")) assert result == "999" def test_match_via_music_title_returns_sound_id(self) -> None: blob = _make_video_blob(title="hateou2024", desc="unrelated desc") page = _make_page(_html_with_blob(blob)) result = _run(_check_video_for_sound(page, "https://t.tt/@a/video/3", "hateou")) assert result == "999" def test_no_match_returns_none(self) -> None: blob = _make_video_blob( title="completely different", desc="unrelated", challenges=[] ) page = _make_page(_html_with_blob(blob)) result = _run(_check_video_for_sound(page, "https://t.tt/@a/video/4", "hateou")) assert result is None def test_no_blob_in_html_returns_none(self) -> None: page = _make_page("no blob here") result = _run(_check_video_for_sound(page, "https://t.tt/@a/video/5", "hateou")) assert result is None def test_blob_missing_item_struct_returns_none(self) -> None: blob: dict[str, Any] = {"__DEFAULT_SCOPE__": {"webapp.video-detail": {}}} page = _make_page(_html_with_blob(blob)) result = _run(_check_video_for_sound(page, "https://t.tt/@a/video/6", "hateou")) assert result is None def test_music_id_none_returns_none(self) -> None: blob = _make_video_blob(desc="#hateou", music_id=None) page = _make_page(_html_with_blob(blob)) result = _run(_check_video_for_sound(page, "https://t.tt/@a/video/7", "hateou")) assert result is None def test_goto_raises_propagates(self) -> None: page = _make_page() page.goto = AsyncMock(side_effect=PlaywrightError("timeout")) with pytest.raises(PlaywrightError): _run(_check_video_for_sound(page, "https://t.tt/@a/video/8", "hateou")) def test_music_title_contains_tag_returns_match(self) -> None: # tag_lower is substring of music_title_norm blob = _make_video_blob(title="hateouremix", desc="other content") page = _make_page(_html_with_blob(blob)) result = _run(_check_video_for_sound(page, "https://t.tt/@a/video/9", "hateou")) assert result == "999" def test_tag_contains_music_title_returns_match(self) -> None: # music_title_norm is substring of tag_lower blob = _make_video_blob(title="hate", desc="other content") page = _make_page(_html_with_blob(blob)) result = _run( _check_video_for_sound(page, "https://t.tt/@a/video/10", "hateouspecial") ) assert result == "999" # ── find_original_sound_id ──────────────────────────────────────────────────── _SD_MOD = "marketing_intelligence.mcp.tools.scraping_tools.sound_discovery" class TestFindOriginalSoundId: def _pool_patch(self, page: MagicMock) -> tuple[MagicMock, MagicMock]: mock_pool = MagicMock() mock_pool.get_page = AsyncMock(return_value=page) mock_pool.close = AsyncMock() return mock_pool, mock_pool.close def test_finds_sound_on_first_video(self) -> None: page = _make_find_page( hrefs=["https://t.tt/@a/video/1", "https://t.tt/@a/video/2"] ) with ( patch(f"{_SD_MOD}._pool") as mp, patch("asyncio.sleep", new_callable=AsyncMock), patch( f"{_SD_MOD}._check_video_for_sound", new=AsyncMock(return_value="sound123"), ), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() result = _run( find_original_sound_id(artist_handle="testartist", track_tag="hateou") ) assert result.success is True assert result.sound_id == "sound123" assert result.artist_handle == "testartist" def test_not_found_returns_failure(self) -> None: page = _make_find_page(hrefs=["https://t.tt/@a/video/1"]) with ( patch(f"{_SD_MOD}._pool") as mp, patch("asyncio.sleep", new_callable=AsyncMock), patch( f"{_SD_MOD}._check_video_for_sound", new=AsyncMock(return_value=None) ), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() result = _run( find_original_sound_id(artist_handle="testartist", track_tag="hateou") ) assert result.success is False assert result.sound_id is None assert "original sound not found" in result.error def test_empty_hrefs_returns_failure(self) -> None: page = _make_find_page(hrefs=[]) with ( patch(f"{_SD_MOD}._pool") as mp, patch("asyncio.sleep", new_callable=AsyncMock), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() result = _run( find_original_sound_id(artist_handle="testartist", track_tag="hateou") ) assert result.success is False assert "0 videos" in result.error def test_profile_load_playwright_error(self) -> None: page = MagicMock() page.goto = AsyncMock(side_effect=PlaywrightError("net::ERR_NAME_NOT_RESOLVED")) with ( patch(f"{_SD_MOD}._pool") as mp, patch("asyncio.sleep", new_callable=AsyncMock), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() result = _run( find_original_sound_id(artist_handle="testartist", track_tag="hateou") ) assert result.success is False assert "profile video scan failed" in result.error def test_video_exception_skipped_next_found(self) -> None: page = _make_find_page( hrefs=["https://t.tt/@a/video/1", "https://t.tt/@a/video/2"] ) check_mock = AsyncMock(side_effect=[PlaywrightError("timeout"), "sound456"]) with ( patch(f"{_SD_MOD}._pool") as mp, patch("asyncio.sleep", new_callable=AsyncMock), patch(f"{_SD_MOD}._check_video_for_sound", new=check_mock), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() result = _run( find_original_sound_id(artist_handle="testartist", track_tag="hateou") ) assert result.success is True assert result.sound_id == "sound456" def test_no_session_closes_pool_in_finally(self) -> None: page = _make_find_page(hrefs=[]) with ( patch(f"{_SD_MOD}._pool") as mp, patch("asyncio.sleep", new_callable=AsyncMock), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() _run(find_original_sound_id(artist_handle="testartist", track_tag="hateou")) mp.close.assert_awaited_once_with("__no_session__") def test_with_session_id_uses_proxy_skips_close(self) -> None: page = _make_find_page(hrefs=[]) mock_sess = MagicMock() mock_sess.proxy = "http://px:3128" with ( patch(f"{_SD_MOD}._pool") as mp, patch(f"{_SD_MOD}.get_session", return_value=mock_sess), patch("asyncio.sleep", new_callable=AsyncMock), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() _run( find_original_sound_id( artist_handle="testartist", track_tag="hateou", session_id="sess1" ) ) mp.get_page.assert_awaited_once_with("sess1", "http://px:3128", headless=ANY) mp.close.assert_not_called() def test_at_prefix_stripped_from_handle(self) -> None: page = _make_find_page(hrefs=["https://t.tt/@a/video/1"]) with ( patch(f"{_SD_MOD}._pool") as mp, patch("asyncio.sleep", new_callable=AsyncMock), patch( f"{_SD_MOD}._check_video_for_sound", new=AsyncMock(return_value="s999") ), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() result = _run( find_original_sound_id(artist_handle="@myartist", track_tag="hateou") ) assert result.artist_handle == "myartist" def test_hash_prefix_stripped_from_tag(self) -> None: page = _make_find_page(hrefs=["https://t.tt/@a/video/1"]) check_mock = AsyncMock(return_value="s888") with ( patch(f"{_SD_MOD}._pool") as mp, patch("asyncio.sleep", new_callable=AsyncMock), patch(f"{_SD_MOD}._check_video_for_sound", new=check_mock), ): mp.get_page = AsyncMock(return_value=page) mp.close = AsyncMock() _run( find_original_sound_id(artist_handle="testartist", track_tag="#hateou") ) check_mock.assert_awaited_once_with(page, "https://t.tt/@a/video/1", "hateou")