MediaCrawler/tests/test_monitor_api.py

# -*- coding: utf-8 -*-
# Copyright (c) 2025 [email protected]
#
# This file is part of MediaCrawler project.
# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/tests/test_monitor_api.py
# GitHub: https://github.com/NanmiCoder
# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
#
# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
# 1. 不得用于任何商业用途。
# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
# 3. 不得进行大规模爬取或对平台造成运营干扰。
# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
# 5. 不得用于任何非法或不当的用途。
#
# 详细许可条款请参阅项目根目录下的LICENSE文件。
# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。

"""API-level tests for the monitoring endpoints.

Run against an ASGI transport with a temporary database, so no server, network
or login is required. Lifespan is deliberately not exercised: it would start the
scheduler, and these tests only cover routing, validation and persistence.
"""

import httpx
import pytest
import pytest_asyncio

from api.main import app
from api.monitor import db as monitor_db
from api.monitor.service import TargetParseError, parse_target_input

CREATOR_URL = (
    "https://www.xiaohongshu.com/user/profile/5f58bd990000000001003753"
    "?xsec_token=ABYVg1evluJZZzpMX-VWzchxQ1qSNVW3r-jOEnKqMcgZw=&xsec_source=pc_search"
)
NOTE_URL = "https://www.xiaohongshu.com/explore/6aa3d827000000002802c5c8?xsec_token=TOKEN&xsec_source=pc_search"


@pytest_asyncio.fixture
async def client(tmp_path):
    monitor_db.set_sqlite_path(tmp_path / "monitor.db")
    await monitor_db.init_db()

    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client:
        yield http_client

    await monitor_db.dispose_engine()


class TestParseTargetInput:
    def test_full_url_splits_id_from_token(self):
        """The id is the stable key; the token is a refreshable credential."""
        parsed = parse_target_input(CREATOR_URL, "creator")
        assert parsed["external_id"] == "5f58bd990000000001003753"
        assert parsed["xsec_token"].startswith("ABYVg1evluJZZzpMX")
        assert parsed["xsec_source"] == "pc_search"

    def test_bare_id_is_accepted(self):
        parsed = parse_target_input("5f58bd990000000001003753", "creator")
        assert parsed["external_id"] == "5f58bd990000000001003753"
        assert parsed["xsec_token"] == ""

    def test_note_url_without_token_still_parses(self):
        parsed = parse_target_input(
            "https://www.xiaohongshu.com/explore/6aa3d827000000002802c5c8", "note"
        )
        assert parsed["external_id"] == "6aa3d827000000002802c5c8"
        assert parsed["xsec_token"] == ""

    def test_creator_url_rejected_in_note_mode(self):
        with pytest.raises(TargetParseError):
            parse_target_input(CREATOR_URL, "note")

    def test_garbage_is_rejected(self):
        with pytest.raises(TargetParseError):
            parse_target_input("not a url at all !!", "creator")

    # --- 抖音 -------------------------------------------------------------
    # 链接形态由平台决定，所以每一个都要显式带上 "dy"。

    def test_douyin_creator_url(self):
        parsed = parse_target_input(
            "https://www.douyin.com/user/MS4wLjABAAAATJPY7LAlaa5X-c8uNdWkvz0jUGgpw4eeXIwu_8BhvqE"
            "?from_tab_name=main",
            "creator",
            "dy",
        )
        assert (
            parsed["external_id"]
            == "MS4wLjABAAAATJPY7LAlaa5X-c8uNdWkvz0jUGgpw4eeXIwu_8BhvqE"
        )

    def test_douyin_video_url(self):
        parsed = parse_target_input(
            "https://www.douyin.com/video/7525082444551310602", "note", "dy"
        )
        assert parsed["external_id"] == "7525082444551310602"

    def test_douyin_modal_id_url(self):
        """在别人主页或搜索结果里点开视频，拿到的就是带 modal_id 的链接。"""
        parsed = parse_target_input(
            "https://www.douyin.com/root/search/python?aid=b733a3b0&modal_id=7471165520058862848",
            "note",
            "dy",
        )
        assert parsed["external_id"] == "7471165520058862848"

    def test_douyin_bare_sec_uid_is_accepted(self):
        sec_uid = "MS4wLjABAAAATJPY7LAlaa5X-c8uNdWkvz0jUGgpw4eeXIwu_8BhvqE"

        parsed = parse_target_input(sec_uid, "creator", "dy")

        assert parsed["external_id"] == sec_uid

    def test_douyin_bare_sec_uid_beyond_the_xhs_length_cap(self):
        """裸 id 的长度上限必须按平台分开。

        小红书那条规则封顶 64 字符，而 sec_user_id 长过 64 是常态（实测样本 55，
        但字段本身是变长的）。共用一条规则的话，长一点的 sec_uid 会被直接拒掉 ——
        对用户来说就是「粘贴了一个完全正确的链接却报无法识别」。
        """
        sec_uid = "MS4wLjABAAAA" + "aB3dEf6hIj9lMn2pQr5tUv8xYz1" * 3
        assert len(sec_uid) > 64

        parsed = parse_target_input(sec_uid, "creator", "dy")
        assert parsed["external_id"] == sec_uid

        # 同一条 id 拿小红书规则来解析会被拒 —— 这正是两条规则必须分开的原因。
        with pytest.raises(TargetParseError):
            parse_target_input(sec_uid, "creator", "xhs")

    def test_douyin_bare_video_id(self):
        parsed = parse_target_input("7525082444551310602", "note", "dy")
        assert parsed["external_id"] == "7525082444551310602"
        # 抖音不需要 xsec_token —— 和小红书不同，裸链接就能用。
        assert parsed["xsec_token"] == ""

    def test_douyin_short_link_is_rejected_with_a_reason(self):
        """短链要联网跳一次才知道指向谁。明确拒绝好过存一个永远抓不到东西的目标。"""
        with pytest.raises(TargetParseError) as excinfo:
            parse_target_input("https://v.douyin.com/drIPtQ_WPWY/", "note", "dy")
        assert "短链" in str(excinfo.value)

    def test_a_douyin_link_is_not_parsed_with_xhs_rules(self):
        with pytest.raises(TargetParseError):
            parse_target_input(
                "https://www.douyin.com/video/7525082444551310602", "note", "xhs"
            )

    def test_an_xhs_link_is_not_parsed_for_douyin(self):
        with pytest.raises(TargetParseError):
            parse_target_input(NOTE_URL, "note", "dy")

    def test_a_platform_without_an_adapter_is_rejected(self):
        with pytest.raises(TargetParseError):
            parse_target_input("whatever", "creator", "bili")


class TestTargetReplacement:
    @pytest.mark.asyncio
    async def test_replacing_targets_uses_the_tasks_own_platform(self, client):
        """改目标必须按任务**自己**的平台解析。

        ``update_task`` 原先漏传了 platform，解析回落到默认的小红书。只有小红书时
        行为恰好正确，接上抖音就会拿小红书的正则去解析抖音链接 —— 建任务时对、
        改任务时错，是最难注意到的那种不一致。
        """
        sec_uid = "MS4wLjABAAAATJPY7LAlaa5X-c8uNdWkvz0jUGgpw4eeXIwu_8BhvqE"
        created = await client.post(
            "/api/monitor/tasks",
            json={"name": "抖音", "mode": "creator", "platform": "dy", "targets": [sec_uid]},
        )
        assert created.status_code == 201
        task_id = created.json()["id"]

        updated = await client.patch(
            f"/api/monitor/tasks/{task_id}",
            json={"targets": [f"https://www.douyin.com/user/{sec_uid}"]},
        )
        assert updated.status_code == 200

        tasks = (
            await client.get("/api/monitor/tasks", params={"platform": "dy"})
        ).json()["tasks"]
        target = tasks[0]["targets"][0]
        assert target["external_id"] == sec_uid
        assert target["raw_value"].startswith("https://www.douyin.com/user/")


class TestTaskCrud:
    @pytest.mark.asyncio
    async def test_create_and_list_task(self, client):
        response = await client.post(
            "/api/monitor/tasks",
            json={
                "name": "网文作者监控",
                "mode": "creator",
                "interval_minutes": 120,
                "targets": [CREATOR_URL, "5f58bd990000000001003754"],
            },
        )
        assert response.status_code == 201
        task_id = response.json()["id"]

        listing = await client.get("/api/monitor/tasks")
        assert listing.status_code == 200
        tasks = listing.json()["tasks"]
        assert len(tasks) == 1
        assert tasks[0]["id"] == task_id
        assert tasks[0]["target_count"] == 2
        # next_run_at is persisted so the schedule survives a restart.
        assert tasks[0]["next_run_at"] is not None

    @pytest.mark.asyncio
    async def test_duplicate_targets_are_deduplicated(self, client):
        response = await client.post(
            "/api/monitor/tasks",
            json={
                "name": "dedup",
                "mode": "creator",
                "targets": [CREATOR_URL, CREATOR_URL],
            },
        )
        assert response.status_code == 201

        listing = await client.get("/api/monitor/tasks")
        assert listing.json()["tasks"][0]["target_count"] == 1

    @pytest.mark.asyncio
    async def test_invalid_target_returns_400(self, client):
        response = await client.post(
            "/api/monitor/tasks",
            json={"name": "bad", "mode": "creator", "targets": ["!!! nonsense !!!"]},
        )
        assert response.status_code == 400

    @pytest.mark.asyncio
    async def test_interval_floor_is_enforced(self, client):
        """A tight poll loop is the pattern that triggers platform rate limits."""
        response = await client.post(
            "/api/monitor/tasks",
            json={"name": "too fast", "mode": "creator", "interval_minutes": 1, "targets": [CREATOR_URL]},
        )
        assert response.status_code == 422

    @pytest.mark.asyncio
    async def test_update_and_delete(self, client):
        created = await client.post(
            "/api/monitor/tasks",
            json={"name": "t", "mode": "note", "targets": [NOTE_URL]},
        )
        task_id = created.json()["id"]

        patched = await client.patch(f"/api/monitor/tasks/{task_id}", json={"enabled": False})
        assert patched.status_code == 200
        listing = await client.get("/api/monitor/tasks")
        assert listing.json()["tasks"][0]["enabled"] is False

        deleted = await client.delete(f"/api/monitor/tasks/{task_id}")
        assert deleted.status_code == 200
        assert (await client.get("/api/monitor/tasks")).json()["tasks"] == []

    @pytest.mark.asyncio
    async def test_run_now_on_missing_task_is_404(self, client):
        response = await client.post("/api/monitor/tasks/9999/run")
        assert response.status_code == 404

    @pytest.mark.asyncio
    async def test_run_history_starts_empty(self, client):
        created = await client.post(
            "/api/monitor/tasks",
            json={"name": "t", "mode": "creator", "targets": [CREATOR_URL]},
        )
        task_id = created.json()["id"]
        runs = await client.get(f"/api/monitor/tasks/{task_id}/runs")
        assert runs.status_code == 200
        assert runs.json()["runs"] == []


class TestCookieEndpoints:
    @pytest.mark.asyncio
    async def test_cookie_value_is_never_returned(self, client):
        """The GET must expose health only, never the credential."""
        secret = "web_session=SUPERSECRETVALUE; a1=abc123"
        saved = await client.post("/api/monitor/cookie", json={"cookie": secret})
        assert saved.status_code == 200

        status_response = await client.get("/api/monitor/cookie")
        assert status_response.status_code == 200
        body = status_response.json()

        assert body["present"] is True
        assert body["length"] == len(secret)
        assert "SUPERSECRETVALUE" not in status_response.text

    @pytest.mark.asyncio
    async def test_cookie_initially_absent_and_clearable(self, client):
        assert (await client.get("/api/monitor/cookie")).json()["present"] is False

        await client.post("/api/monitor/cookie", json={"cookie": "web_session=x"})
        assert (await client.get("/api/monitor/cookie")).json()["present"] is True

        await client.delete("/api/monitor/cookie")
        assert (await client.get("/api/monitor/cookie")).json()["present"] is False


class TestDashboardQueries:
    @pytest.mark.asyncio
    async def test_empty_dashboard_shapes(self, client):
        assert (await client.get("/api/monitor/notes")).json()["notes"] == []
        assert (await client.get("/api/monitor/comments")).json()["comments"] == []
        assert (await client.get("/api/monitor/events")).json()["events"] == []

        overview = (await client.get("/api/monitor/overview")).json()
        assert overview["tasks"] == 0
        assert overview["notes"] == 0