Xiaohongshu Cli

name: Bug Report
description: Report a bug or unexpected behavior
labels: ["bug"]
body:
  - type: input
    id: version
    attributes:
      label: Version
      description: "Run `xhs --version` or `pip show xiaohongshu-cli | grep Version`"
      placeholder: "e.g. 1.0.0"
    validations:
      required: true
  - type: dropdown
    id: os
    attributes:
      label: Operating System
      options:
        - macOS
        - Linux
        - Windows
        - Other
    validations:
      required: true
  - type: textarea
    id: description
    attributes:
      label: What happened?
      description: Describe the bug clearly.
    validations:
      required: true
  - type: textarea
    id: expected
    attributes:
      label: Expected behavior
      description: What did you expect to happen?
    validations:
      required: true
  - type: textarea
    id: reproduce
    attributes:
      label: Steps to reproduce
      description: "Commands or steps to reproduce the issue. Use `xhs -v <command>` for debug output."
      render: bash
  - type: textarea
    id: logs
    attributes:
      label: Error output / logs
      description: Paste any error messages or verbose output here.
      render: text

name: Feature Request
description: Suggest a new feature or improvement
labels: ["enhancement"]
body:
  - type: input
    id: version
    attributes:
      label: Current version
      description: "Run `xhs --version` or `pip show xiaohongshu-cli | grep Version`"
      placeholder: "e.g. 1.0.0"
    validations:
      required: false
  - type: textarea
    id: description
    attributes:
      label: Describe the feature
      description: What would you like to see added or changed?
    validations:
      required: true
  - type: textarea
    id: use_case
    attributes:
      label: Use case
      description: Why do you need this feature? How would you use it?

name: CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]
  workflow_call:

jobs:
  lint-and-test:
    name: Lint and test (Python ${{ matrix.python-version }})
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false
      matrix:
        python-version: ["3.10", "3.12", "3.13"]
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}

      - name: Install uv
        uses: astral-sh/setup-uv@v6

      - name: Install dependencies
        run: uv sync --extra dev

      - name: Run ruff
        run: uv run ruff check .

      - name: Run tests
        run: uv run python -m pytest -q --ignore=tests/test_integration.py

  build:
    name: Build package
    runs-on: ubuntu-latest
    needs: lint-and-test
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install uv
        uses: astral-sh/setup-uv@v6

      - name: Install dependencies
        run: uv sync --extra dev

      - name: Build distribution
        run: uv build

name: Publish to PyPI

on:
  push:
    tags:
      - "v*"
  workflow_dispatch:

jobs:
  verify:
    uses: ./.github/workflows/ci.yml

  publish:
    needs: verify
    runs-on: ubuntu-latest
    environment: pypi
    permissions:
      id-token: write

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Setup uv
        uses: astral-sh/setup-uv@v6

      - name: Build package
        run: uv build

      - name: Publish to PyPI
        uses: pypa/gh-action-pypi-publish@release/v1

__pycache__/
*.pyc
.venv/
dist/
*.egg-info/
.ruff_cache/
.mypy_cache/
.pytest_cache/
.playwright-mcp/
xhs_login_page.png

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "xiaohongshu-cli"
version = "0.6.4"
description = "A CLI for Xiaohongshu (Red) — search, read, interact via reverse-engineered API"
readme = "README.md"
license = "Apache-2.0"
requires-python = ">=3.10"
authors = [{ name = "jackwener", email = "jakevingoo@gmail.com" }]
keywords = ["xiaohongshu", "xhs", "redbook", "cli", "api"]
classifiers = [
    "Development Status :: 3 - Alpha",
    "Environment :: Console",
    "License :: OSI Approved :: Apache Software License",
    "Programming Language :: Python :: 3",
    "Topic :: Utilities",
]
dependencies = [
    "httpx>=0.27",
    "click>=8.0",
    "rich>=13.0",
    "browser-cookie3>=0.19",
    "pycryptodome>=3.20",
    "xhshow>=0.1.9",
    "PyYAML>=6.0",
    "qrcode>=7.0",
    "camoufox>=0.4.11",
]

[project.optional-dependencies]
dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "ruff>=0.11.0", "mypy>=1.15.0"]

[project.urls]
Homepage = "https://github.com/jackwener/xiaohongshu-cli"
Repository = "https://github.com/jackwener/xiaohongshu-cli"
Issues = "https://github.com/jackwener/xiaohongshu-cli/issues"

[project.scripts]
xhs = "xhs_cli.cli:cli"

[tool.hatch.build.targets.wheel]
packages = ["xhs_cli"]

[tool.ruff]
target-version = "py310"
line-length = 120

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP"]

[tool.mypy]
python_version = "3.10"
ignore_missing_imports = true
check_untyped_defs = true

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
addopts = "-m 'not smoke'"
markers = ["smoke: real-API integration tests (run with: pytest -m smoke)"]

from __future__ import annotations

import os

os.environ.setdefault("OUTPUT", "rich")

"""Tests for anti-detection measures: UA/platform alignment, header completeness, jitter."""


from xhs_cli.client import XhsClient
from xhs_cli.constants import CHROME_VERSION, PLATFORM, USER_AGENT
from xhs_cli.signing import sign_main_api


class TestUAPlatformConsistency:
    """UA, sec-ch-ua, and fingerprint must all agree on macOS Chrome."""

    def test_ua_is_macos_chrome(self):
        assert "Macintosh" in USER_AGENT
        assert "Chrome/" in USER_AGENT
        assert "Edg/" not in USER_AGENT
        assert "Windows" not in USER_AGENT

    def test_platform_is_macos(self):
        assert PLATFORM == "macOS"

    def test_base_headers_match_ua(self):
        client = XhsClient({"a1": "test"})
        try:
            headers = client._base_headers()
            # sec-ch-ua must reference Chrome, not Edge
            assert "Google Chrome" in headers["sec-ch-ua"]
            assert CHROME_VERSION in headers["sec-ch-ua"]
            assert "Edge" not in headers["sec-ch-ua"]
            # Platform must match
            assert "macOS" in headers["sec-ch-ua-platform"]
        finally:
            client.close()


class TestSigningHeaders:
    """Verify sign_main_api returns all required headers."""

    def test_returns_all_required_keys(self):
        cookies = {"a1": "test_a1_12345678901234567890123456789012345678901234"}
        headers = sign_main_api("GET", "/api/test", cookies)
        expected_keys = {"x-s", "x-s-common", "x-t", "x-b3-traceid", "x-xray-traceid"}
        assert set(headers.keys()) == expected_keys

    def test_xs_has_xys_prefix(self):
        cookies = {"a1": "test_a1_12345678901234567890123456789012345678901234"}
        headers = sign_main_api("GET", "/api/test", cookies)
        assert headers["x-s"].startswith("XYS_")

    def test_get_and_post_return_different_xs(self):
        cookies = {"a1": "test_a1_12345678901234567890123456789012345678901234"}
        h_get = sign_main_api("GET", "/api/test", cookies)
        h_post = sign_main_api("POST", "/api/test", cookies, payload={"key": "value"})
        # Different method/payload → different signature
        assert h_get["x-s"] != h_post["x-s"]


class TestClientJitter:
    """Verify jitter produces variable delays (not fixed intervals)."""

    def test_request_delay_default(self):
        client = XhsClient({"a1": "test"})
        try:
            assert client._request_delay == 1.0
            assert client._base_request_delay == 1.0
        finally:
            client.close()

    def test_verify_count_starts_at_zero(self):
        client = XhsClient({"a1": "test"})
        try:
            assert client._verify_count == 0
            assert client._request_count == 0
        finally:
            client.close()


class TestBaseHeadersCompleteness:
    """Ensure all anti-detection headers are present."""

    def test_has_dnt_header(self):
        client = XhsClient({"a1": "test"})
        try:
            headers = client._base_headers()
            assert headers.get("dnt") == "1"
        finally:
            client.close()

    def test_has_priority_header(self):
        client = XhsClient({"a1": "test"})
        try:
            headers = client._base_headers()
            assert "priority" in headers
        finally:
            client.close()

    def test_has_all_sec_fetch_headers(self):
        client = XhsClient({"a1": "test"})
        try:
            headers = client._base_headers()
            assert "sec-fetch-dest" in headers
            assert "sec-fetch-mode" in headers
            assert "sec-fetch-site" in headers
        finally:
            client.close()

"""Tests for CLI commands using Click's test runner."""

import pytest
import yaml
from click.testing import CliRunner

from xhs_cli.cli import cli
from xhs_cli.exceptions import NoCookieError, SessionExpiredError, UnsupportedOperationError

runner = CliRunner()

FAKE_NOTE_RESPONSE = {
    "items": [
        {
            "note_card": {
                "title": "Test Note",
                "desc": "body",
                "user": {"nickname": "Author"},
                "interact_info": {
                    "liked_count": "100",
                    "collected_count": "50",
                    "comment_count": "10",
                    "share_count": "5",
                },
                "tag_list": [],
                "image_list": [],
            }
        }
    ]
}


class TestCliBasic:
    """Test CLI basics without requiring cookies."""

    def test_version(self):
        result = runner.invoke(cli, ["--version"])
        assert result.exit_code == 0
        assert "0." in result.output  # dynamic version from importlib.metadata

    def test_help(self):
        result = runner.invoke(cli, ["--help"])
        assert result.exit_code == 0
        assert "xhs" in result.output
        assert "search" in result.output
        assert "read" in result.output

    def test_search_help(self):
        result = runner.invoke(cli, ["search", "--help"])
        assert result.exit_code == 0
        assert "keyword" in result.output.lower() or "KEYWORD" in result.output

    def test_read_help(self):
        result = runner.invoke(cli, ["read", "--help"])
        assert result.exit_code == 0

    def test_login_help(self):
        result = runner.invoke(cli, ["login", "--help"])
        assert result.exit_code == 0

    def test_status_help(self):
        result = runner.invoke(cli, ["status", "--help"])
        assert result.exit_code == 0

    def test_all_commands_registered(self):
        result = runner.invoke(cli, ["--help"])
        commands_expected = [
            # Auth
            "login", "status", "logout", "whoami",
            # Reading
            "search", "read", "comments", "sub-comments", "user", "user-posts",
            "feed", "hot", "topics", "search-user", "my-notes",
            "notifications", "unread",
            # Interactions
            "like", "favorite", "unfavorite", "comment", "reply", "delete-comment",
            # Social
            "follow", "unfollow", "favorites",
            # Creator
            "post", "delete",
        ]
        for cmd in commands_expected:
            assert cmd in result.output, f"Command '{cmd}' not found in CLI help"

    def test_whoami_help(self):
        result = runner.invoke(cli, ["whoami", "--help"])
        assert result.exit_code == 0

    def test_hot_help(self):
        result = runner.invoke(cli, ["hot", "--help"])
        assert result.exit_code == 0
        assert "category" in result.output.lower()

    def test_unread_help(self):
        result = runner.invoke(cli, ["unread", "--help"])
        assert result.exit_code == 0

    def test_my_notes_help(self):
        result = runner.invoke(cli, ["my-notes", "--help"])
        assert result.exit_code == 0

    def test_status_auto_yaml_when_stdout_is_not_tty(self, monkeypatch):
        monkeypatch.setenv("OUTPUT", "auto")
        monkeypatch.setattr(
            "xhs_cli.commands.auth.run_client_action",
            lambda ctx, action: {"nickname": "Alice", "red_id": "alice001"},
        )

        result = runner.invoke(cli, ["status"])

        assert result.exit_code == 0
        payload = yaml.safe_load(result.output)
        assert payload["ok"] is True
        assert payload["schema_version"] == "1"
        assert payload["data"]["authenticated"] is True
        assert payload["data"]["user"]["name"] == "Alice"

    def test_whoami_auto_yaml_when_stdout_is_not_tty(self, monkeypatch):
        monkeypatch.setenv("OUTPUT", "auto")
        monkeypatch.setattr(
            "xhs_cli.commands.auth.run_client_action",
            lambda ctx, action: {"nickname": "Alice", "red_id": "alice001", "user_id": "u-1"},
        )

        result = runner.invoke(cli, ["whoami"])

        assert result.exit_code == 0
        payload = yaml.safe_load(result.output)
        assert payload["ok"] is True
        assert payload["data"]["user"]["username"] == "alice001"

    def test_read_error_yaml_when_not_logged_in(self, monkeypatch):
        monkeypatch.setenv("OUTPUT", "auto")
        monkeypatch.setattr(
            "xhs_cli.commands._common.get_cookies",
            lambda source, force_refresh=False: (_ for _ in ()).throw(NoCookieError(source)),
        )

        result = runner.invoke(cli, ["read", "abc", "--yaml"])

        assert result.exit_code != 0
        payload = yaml.safe_load(result.output)
        assert payload["ok"] is False
        assert payload["error"]["code"] == "not_authenticated"

    def test_status_reports_not_authenticated_when_session_expired(self, monkeypatch):
        monkeypatch.setenv("OUTPUT", "auto")

        def fake_run_client_action(ctx, action):
            raise SessionExpiredError()

        monkeypatch.setattr("xhs_cli.commands.auth.run_client_action", fake_run_client_action)

        result = runner.invoke(cli, ["status", "--yaml"])

        assert result.exit_code != 0
        payload = yaml.safe_load(result.output)
        assert payload["ok"] is False
        assert payload["error"]["code"] == "not_authenticated"

    def test_logout_supports_structured_output(self):
        from xhs_cli.commands import auth

        original_clear_cookies = auth.clear_cookies
        auth.clear_cookies = lambda: None
        try:
            result = runner.invoke(cli, ["logout", "--yaml"])
        finally:
            auth.clear_cookies = original_clear_cookies

        assert result.exit_code == 0
        payload = yaml.safe_load(result.output)
        assert payload["ok"] is True
        assert payload["data"]["logged_out"] is True

    def test_delete_reports_unsupported_operation(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.commands.creator.run_client_action",
            lambda ctx, action: (_ for _ in ()).throw(
                UnsupportedOperationError("Delete note is currently unavailable from the public web API.")
            ),
        )

        result = runner.invoke(cli, ["delete", "note-123", "--yes", "--yaml"])

        assert result.exit_code != 0
        payload = yaml.safe_load(result.output)
        assert payload["ok"] is False
        assert payload["error"]["code"] == "unsupported_operation"

    def test_comments_rich_output_handles_string_reply_counts(self, monkeypatch):
        monkeypatch.setenv("OUTPUT", "rich")
        monkeypatch.setattr(
            "xhs_cli.commands.reading.run_client_action",
            lambda ctx, action: {
                "comments": [
                    {
                        "user_info": {"nickname": "tester"},
                        "content": "hello",
                        "like_count": "12",
                        "sub_comment_count": "2",
                    }
                ]
            },
        )

        result = runner.invoke(cli, ["comments", "note-123"])

        assert result.exit_code == 0
        assert "tester" in result.output
        assert "2 replies" in result.output

    def test_search_rich_output_shortens_visible_links(self, monkeypatch):
        monkeypatch.setenv("OUTPUT", "rich")
        monkeypatch.setattr(
            "xhs_cli.commands.reading.handle_command",
            lambda ctx, action, render, as_json, as_yaml: render({
                "items": [
                    {
                        "id": "69ad061d000000002603326d",
                        "xsec_token": "very-long-token-value",
                        "note_card": {
                            "title": "测试标题",
                            "user": {"nickname": "tester"},
                            "interact_info": {"liked_count": "12"},
                            "type": "normal",
                        },
                    }
                ],
                "has_more": False,
            }),
        )

        result = runner.invoke(cli, ["search", "openclaw"])

        assert result.exit_code == 0
        assert "search_result/69ad061d" in result.output
        assert "very-long-token-value" not in result.output

    def test_feed_rich_output_shortens_visible_links(self, monkeypatch):
        monkeypatch.setenv("OUTPUT", "rich")
        monkeypatch.setattr(
            "xhs_cli.commands.reading.handle_command",
            lambda ctx, action, render, as_json, as_yaml: render({
                "items": [
                    {
                        "id": "69ad061d000000002603326d",
                        "xsec_token": "another-very-long-token",
                        "note_card": {
                            "title": "推荐内容",
                            "user": {"nickname": "tester"},
                            "interact_info": {"liked_count": "9"},
                        },
                    }
                ]
            }),
        )

        result = runner.invoke(cli, ["feed"])

        assert result.exit_code == 0
        assert "explore/69ad061d" in result.output
        assert "another-very-long-token" not in result.output

    def test_read_help_mentions_short_index(self):
        result = runner.invoke(cli, ["read", "--help"])
        assert result.exit_code == 0
        assert "index" in result.output.lower()

    def test_comments_help_mentions_short_index(self):
        result = runner.invoke(cli, ["comments", "--help"])
        assert result.exit_code == 0
        assert "index" in result.output.lower()

    def test_read_index_resolves_note_context(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.note_refs.get_note_by_index",
            lambda idx: {
                "note_id": "note-abc",
                "xsec_token": "token-abc",
                "xsec_source": "pc_search",
            } if idx == 1 else None,
        )

        called = {}

        class FakeClient:
            def get_note_detail(self, note_id, **kwargs):
                called["note_id"] = note_id
                called["kwargs"] = kwargs
                return FAKE_NOTE_RESPONSE

        def fake_handle_command(ctx, action, render, as_json, as_yaml):
            action(FakeClient())
            return None

        monkeypatch.setattr("xhs_cli.commands.reading.handle_command", fake_handle_command)

        result = runner.invoke(cli, ["read", "1"])

        assert result.exit_code == 0
        assert called["note_id"] == "note-abc"
        assert called["kwargs"]["xsec_token"] == "token-abc"
        assert called["kwargs"]["xsec_source"] == "pc_search"

    def test_comments_index_resolves_note_context(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.note_refs.get_note_by_index",
            lambda idx: {
                "note_id": "note-abc",
                "xsec_token": "token-abc",
                "xsec_source": "pc_search",
            } if idx == 1 else None,
        )

        called = {}

        class FakeClient:
            def get_comments(self, note_id, cursor="", **kwargs):
                called["note_id"] = note_id
                called["cursor"] = cursor
                called["kwargs"] = kwargs
                return {"comments": []}

        def fake_run_client_action(ctx, action):
            return action(FakeClient())

        monkeypatch.setattr("xhs_cli.commands.reading.run_client_action", fake_run_client_action)

        result = runner.invoke(cli, ["comments", "1", "--yaml"])

        assert result.exit_code == 0
        assert called["note_id"] == "note-abc"
        assert called["kwargs"]["xsec_token"] == "token-abc"
        assert called["kwargs"]["xsec_source"] == "pc_search"

    def test_read_index_not_found_returns_usage_error(self, monkeypatch):
        monkeypatch.setattr("xhs_cli.note_refs.get_note_by_index", lambda idx: None)

        result = runner.invoke(cli, ["read", "999"])

        assert result.exit_code != 0
        assert "999" in result.output

    def test_search_empty_results_clear_previous_index(self, monkeypatch):
        from xhs_cli.note_refs import save_index_from_items

        saved = []
        monkeypatch.setattr("xhs_cli.note_refs.save_note_index", lambda items: saved.append(items))

        save_index_from_items({"items": []}, xsec_source="pc_search")

        assert saved == [[]]

    def test_user_posts_saves_index_entries(self, monkeypatch):
        saved = []
        monkeypatch.setattr("xhs_cli.note_refs.save_note_index", lambda items: saved.append(items))

        def fake_handle_command(ctx, action, render, as_json, as_yaml):
            class FakeClient:
                def get_user_notes(self, user_id, cursor=""):
                    return {
                        "notes": [
                            {"note_id": "note-1"},
                            {"note_id": "note-2", "xsec_token": "ignored"},
                        ],
                        "has_more": False,
                        "cursor": "",
                    }

            data = action(FakeClient())
            render(data)
            return None

        monkeypatch.setattr("xhs_cli.commands.reading.handle_command", fake_handle_command)

        result = runner.invoke(cli, ["user-posts", "user-1"])

        assert result.exit_code == 0
        assert saved == [[
            {"note_id": "note-1", "xsec_token": "", "xsec_source": ""},
            {"note_id": "note-2", "xsec_token": "ignored", "xsec_source": ""},
        ]]

    @pytest.mark.parametrize(
        ("command", "extra_args", "method_name"),
        [
            ("like", [], "like_note"),
            ("like", ["--undo"], "unlike_note"),
            ("favorite", [], "favorite_note"),
            ("unfavorite", [], "unfavorite_note"),
        ],
    )
    def test_short_index_resolves_for_note_actions(self, monkeypatch, command, extra_args, method_name):
        monkeypatch.setattr(
            "xhs_cli.note_refs.get_note_by_index",
            lambda idx: {
                "note_id": "note-abc",
                "xsec_token": "token-abc",
                "xsec_source": "pc_search",
            } if idx == 1 else None,
        )

        called = {}

        class FakeClient:
            def __getattr__(self, name):
                if name != method_name:
                    raise AttributeError(name)

                def _call(note_id):
                    called["method"] = name
                    called["note_id"] = note_id
                    return {"ok": True}

                return _call

        def fake_handle_command(ctx, action, render, as_json, as_yaml):
            action(FakeClient())
            return None

        monkeypatch.setattr("xhs_cli.commands.interactions.handle_command", fake_handle_command)

        result = runner.invoke(cli, [command, "1", *extra_args])

        assert result.exit_code == 0
        assert called == {"method": method_name, "note_id": "note-abc"}

    def test_short_index_resolves_for_comment(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.note_refs.get_note_by_index",
            lambda idx: {
                "note_id": "note-abc",
                "xsec_token": "token-abc",
                "xsec_source": "pc_search",
            } if idx == 1 else None,
        )

        called = {}

        class FakeClient:
            def post_comment(self, note_id, content):
                called["note_id"] = note_id
                called["content"] = content
                return {"ok": True}

        def fake_handle_command(ctx, action, render, as_json, as_yaml):
            action(FakeClient())
            return None

        monkeypatch.setattr("xhs_cli.commands.interactions.handle_command", fake_handle_command)

        result = runner.invoke(cli, ["comment", "1", "-c", "hello"])

        assert result.exit_code == 0
        assert called == {"note_id": "note-abc", "content": "hello"}

    def test_short_index_resolves_for_reply(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.note_refs.get_note_by_index",
            lambda idx: {
                "note_id": "note-abc",
                "xsec_token": "token-abc",
                "xsec_source": "pc_search",
            } if idx == 1 else None,
        )

        called = {}

        class FakeClient:
            def reply_comment(self, note_id, comment_id, content):
                called["note_id"] = note_id
                called["comment_id"] = comment_id
                called["content"] = content
                return {"ok": True}

        def fake_handle_command(ctx, action, render, as_json, as_yaml):
            action(FakeClient())
            return None

        monkeypatch.setattr("xhs_cli.commands.interactions.handle_command", fake_handle_command)

        result = runner.invoke(cli, ["reply", "1", "--comment-id", "c-1", "-c", "hello"])

        assert result.exit_code == 0
        assert called == {"note_id": "note-abc", "comment_id": "c-1", "content": "hello"}

    def test_favorites_saves_index_entries(self, monkeypatch):
        saved = []
        monkeypatch.setattr("xhs_cli.note_refs.save_note_index", lambda items: saved.append(items))
        monkeypatch.setattr("xhs_cli.commands.social._resolve_user_id", lambda ctx, user_id: "user-1")

        def fake_handle_command(ctx, action, render, as_json, as_yaml):
            class FakeClient:
                def get_user_favorites(self, user_id, cursor=""):
                    return {
                        "notes": [
                            {"note_id": "note-1"},
                            {"id": "note-2", "xsec_token": "token-2"},
                        ],
                        "has_more": False,
                        "cursor": "",
                    }

            data = action(FakeClient())
            render(data)
            return None

        monkeypatch.setattr("xhs_cli.commands.social.handle_command", fake_handle_command)

        result = runner.invoke(cli, ["favorites"])

        assert result.exit_code == 0
        assert saved == [[
            {"note_id": "note-1", "xsec_token": "", "xsec_source": ""},
            {"note_id": "note-2", "xsec_token": "token-2", "xsec_source": ""},
        ]]

    def test_my_notes_saves_index_entries(self, monkeypatch):
        saved = []
        monkeypatch.setattr("xhs_cli.note_refs.save_note_index", lambda items: saved.append(items))

        def fake_handle_command(ctx, action, render, as_json, as_yaml):
            class FakeClient:
                def get_creator_note_list(self, page=0):
                    return {
                        "note_list": [
                            {"note_id": "note-1"},
                            {"id": "note-2"},
                        ]
                    }

            data = action(FakeClient())
            render(data)
            return None

        monkeypatch.setattr("xhs_cli.commands.creator.handle_command", fake_handle_command)

        result = runner.invoke(cli, ["my-notes"])

        assert result.exit_code == 0
        assert saved == [[
            {"note_id": "note-1", "xsec_token": "", "xsec_source": ""},
            {"note_id": "note-2", "xsec_token": "", "xsec_source": ""},
        ]]

"""Unit tests for XHS client request payloads, cookies, and endpoint selection."""

from collections import OrderedDict

import httpx
import pytest

from xhs_cli.client import XhsClient
from xhs_cli.cookies import cache_note_context, get_cached_note_context
from xhs_cli.exceptions import UnsupportedOperationError, XhsApiError


class TestFavorites:
    def test_unfavorite_uses_note_ids_payload(self, monkeypatch):
        captured = {}

        def fake_post(self, uri, data, header_overrides=None):
            captured["uri"] = uri
            captured["data"] = data
            return True

        monkeypatch.setattr(XhsClient, "_main_api_post", fake_post)

        client = XhsClient({"a1": "cookie"})
        try:
            client.unfavorite_note("note-123")
        finally:
            client.close()

        assert captured["uri"] == "/api/sns/web/v1/note/uncollect"
        assert captured["data"] == {"note_ids": "note-123"}


class TestCreatorEndpoints:
    def test_creator_note_list_uses_v2_endpoint(self, monkeypatch):
        captured = {}

        def fake_get(self, uri, params=None):
            captured["uri"] = uri
            captured["params"] = params
            return {"notes": [], "page": -1}

        monkeypatch.setattr(XhsClient, "_creator_get", fake_get)

        client = XhsClient({"a1": "cookie"})
        try:
            result = client.get_creator_note_list(page=2)
        finally:
            client.close()

        assert result["page"] == -1
        assert captured["uri"] == "/api/galaxy/v2/creator/note/user/posted"
        assert captured["params"] == {"tab": 0, "page": 2}

    def test_delete_note_raises_unsupported_for_404(self, monkeypatch):
        def fake_post(self, uri, data):
            raise XhsApiError(
                "API error: {\"status\": 404}",
                response={"status": 404},
            )

        monkeypatch.setattr(XhsClient, "_creator_post", fake_post)

        client = XhsClient({"a1": "cookie"})
        try:
            with pytest.raises(UnsupportedOperationError, match="Delete note is currently unavailable"):
                client.delete_note("note-123")
        finally:
            client.close()


class TestTransportCookies:
    def test_request_with_retry_merges_response_cookies(self, monkeypatch):
        request = httpx.Request("POST", "https://edith.xiaohongshu.com/api/test")
        response = httpx.Response(
            200,
            headers=[
                ("set-cookie", "web_session=real-session; Path=/; Domain=.xiaohongshu.com"),
                ("set-cookie", "web_session_sec=real-sec; Path=/; Domain=.xiaohongshu.com"),
            ],
            json={"success": True, "data": {"ok": True}},
            request=request,
        )

        class _FakeHttpClient:
            def request(self, method, url, **kwargs):
                return response

            def close(self):
                return None

        client = XhsClient({"a1": "cookie", "web_session": "guest-session"}, request_delay=0)
        client._http = _FakeHttpClient()
        try:
            resp = client._request_with_retry("POST", "https://edith.xiaohongshu.com/api/test")
        finally:
            client.close()

        assert resp is response
        assert client.cookies["web_session"] == "real-session"
        assert client.cookies["web_session_sec"] == "real-sec"


class TestReadingEndpointBehavior:
    def test_get_note_detail_prefers_cached_xsec_source(self, monkeypatch):
        captured = {}

        def fake_get_note_by_id(self, note_id, xsec_token="", xsec_source="pc_feed"):
            captured["note_id"] = note_id
            captured["token"] = xsec_token
            captured["source"] = xsec_source
            return {"items": [{"note_card": {"title": "ok"}}]}

        monkeypatch.setattr(XhsClient, "get_note_by_id", fake_get_note_by_id)
        cache_note_context("note-123", "token-xyz", "pc_search")

        client = XhsClient({"a1": "cookie"})
        try:
            client.get_note_detail("note-123")
        finally:
            client.close()

        assert captured == {
            "note_id": "note-123",
            "token": "token-xyz",
            "source": "pc_search",
        }

    def test_search_notes_uses_browser_like_prewarm_sequence(self, monkeypatch):
        calls = []

        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE", OrderedDict())
        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE_LOADED", True)

        def fake_get(self, uri, params=None):
            calls.append(("GET", uri, params))
            return {"ok": True}

        def fake_post(self, uri, data, header_overrides=None):
            calls.append(("POST", uri, data))
            if uri == "/api/sns/web/v1/search/notes":
                return {"items": [], "has_more": False}
            return {"ok": True}

        monkeypatch.setattr(XhsClient, "_main_api_get", fake_get)
        monkeypatch.setattr(XhsClient, "_main_api_post", fake_post)

        client = XhsClient({"a1": "cookie"})
        try:
            client.search_notes("openclaw prompt", page=2)
        finally:
            client.close()

        assert [call[1] for call in calls] == [
            "/api/sns/web/v1/search/onebox",
            "/api/sns/web/v1/search/filter",
            "/api/sns/web/v1/search/notes",
            "/api/sns/web/v1/search/recommend",
        ]

        notes_payload = calls[2][2]
        assert notes_payload["page"] == 2
        assert notes_payload["filters"][0]["type"] == "sort_type"
        assert notes_payload["filters"][1]["type"] == "filter_note_type"

    def test_search_notes_reuses_search_id_across_pages(self, monkeypatch):
        calls = []

        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE", OrderedDict())
        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE_LOADED", True)

        def fake_get(self, uri, params=None):
            calls.append(("GET", uri, params))
            return {"ok": True}

        def fake_post(self, uri, data, header_overrides=None):
            calls.append(("POST", uri, data))
            if uri == "/api/sns/web/v1/search/notes":
                return {"items": [], "has_more": False}
            return {"ok": True}

        monkeypatch.setattr(XhsClient, "_main_api_get", fake_get)
        monkeypatch.setattr(XhsClient, "_main_api_post", fake_post)

        client = XhsClient({"a1": "cookie"})
        try:
            client.search_notes("openclaw", page=1)
            client.search_notes("openclaw", page=2)
        finally:
            client.close()

        notes_calls = [call for call in calls if call[1] == "/api/sns/web/v1/search/notes"]
        assert len(notes_calls) == 2
        assert notes_calls[0][2]["search_id"] == notes_calls[1][2]["search_id"]

        onebox_calls = [call for call in calls if call[1] == "/api/sns/web/v1/search/onebox"]
        filter_calls = [call for call in calls if call[1] == "/api/sns/web/v1/search/filter"]
        recommend_calls = [call for call in calls if call[1] == "/api/sns/web/v1/search/recommend"]
        assert len(onebox_calls) == 1
        assert len(filter_calls) == 1
        assert len(recommend_calls) == 1

    def test_search_notes_reuses_search_id_after_cache_reload(self, monkeypatch, tmp_path):
        first_calls = []
        second_calls = []

        monkeypatch.setattr("xhs_cli.client_mixins.get_config_dir", lambda: tmp_path)
        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE", OrderedDict())
        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE_LOADED", False)
        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE_PATH", None)

        def fake_get_first(self, uri, params=None):
            first_calls.append(("GET", uri, params))
            return {"ok": True}

        def fake_post_first(self, uri, data, header_overrides=None):
            first_calls.append(("POST", uri, data))
            if uri == "/api/sns/web/v1/search/notes":
                return {"items": [], "has_more": False}
            return {"ok": True}

        monkeypatch.setattr(XhsClient, "_main_api_get", fake_get_first)
        monkeypatch.setattr(XhsClient, "_main_api_post", fake_post_first)

        client = XhsClient({"a1": "cookie"})
        try:
            client.search_notes("openclaw", page=1)
        finally:
            client.close()

        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE", OrderedDict())
        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE_LOADED", False)
        monkeypatch.setattr("xhs_cli.client_mixins._SEARCH_SESSION_CACHE_PATH", None)

        def fake_get_second(self, uri, params=None):
            second_calls.append(("GET", uri, params))
            return {"ok": True}

        def fake_post_second(self, uri, data, header_overrides=None):
            second_calls.append(("POST", uri, data))
            if uri == "/api/sns/web/v1/search/notes":
                return {"items": [], "has_more": False}
            return {"ok": True}

        monkeypatch.setattr(XhsClient, "_main_api_get", fake_get_second)
        monkeypatch.setattr(XhsClient, "_main_api_post", fake_post_second)

        client = XhsClient({"a1": "cookie"})
        try:
            client.search_notes("openclaw", page=2)
        finally:
            client.close()

        first_notes_call = next(call for call in first_calls if call[1] == "/api/sns/web/v1/search/notes")
        second_notes_call = next(call for call in second_calls if call[1] == "/api/sns/web/v1/search/notes")
        assert first_notes_call[2]["search_id"] == second_notes_call[2]["search_id"]
        assert [call[1] for call in second_calls] == ["/api/sns/web/v1/search/notes"]

    def test_get_note_detail_invalidates_cached_context_after_feed_failure(self, monkeypatch):
        cache_note_context("note-123", "stale-token", "pc_search")

        def fake_get_note_by_id(self, note_id, xsec_token="", xsec_source="pc_feed"):
            raise XhsApiError("stale token")

        def fake_get_note_from_html(self, note_id, xsec_token="", xsec_source="pc_feed"):
            return {"items": [{"note_card": {"title": "html-ok"}}]}

        monkeypatch.setattr(XhsClient, "get_note_by_id", fake_get_note_by_id)
        monkeypatch.setattr(XhsClient, "get_note_from_html", fake_get_note_from_html)

        client = XhsClient({"a1": "cookie"})
        try:
            result = client.get_note_detail("note-123")
        finally:
            client.close()

        assert result["items"][0]["note_card"]["title"] == "html-ok"
        assert get_cached_note_context("note-123") == {}

"""Unit tests for cookie management (no network required)."""


import time

import pytest

from xhs_cli.cookies import (
    NOTE_CONTEXT_TTL_SECONDS,
    cache_note_context,
    clear_cookies,
    cookies_to_string,
    get_cached_note_context,
    get_cached_xsec_token,
    get_cookies,
    get_index_cache_path,
    get_note_by_index,
    get_token_cache_path,
    load_saved_cookies,
    load_token_cache,
    save_cookies,
    save_note_index,
)


@pytest.fixture
def tmp_config_dir(tmp_path, monkeypatch):
    """Override config dir to use temp directory."""
    monkeypatch.setattr("xhs_cli.cookies.get_config_dir", lambda: tmp_path)
    monkeypatch.setattr("xhs_cli.cookies.get_cookie_path", lambda: tmp_path / "cookies.json")
    monkeypatch.setattr("xhs_cli.cookies._TOKEN_CACHE_MEMORY", None)
    monkeypatch.setattr("xhs_cli.cookies._TOKEN_CACHE_PATH", None)
    return tmp_path


class TestSaveCookies:
    def test_save_and_load(self, tmp_config_dir):
        cookies = {"a1": "test_value", "web_session": "sess123"}
        save_cookies(cookies)

        loaded = load_saved_cookies()
        assert loaded is not None
        assert loaded["a1"] == "test_value"
        assert loaded["web_session"] == "sess123"

    def test_file_permissions(self, tmp_config_dir):
        cookies = {"a1": "test"}
        save_cookies(cookies)

        cookie_file = tmp_config_dir / "cookies.json"
        stat = cookie_file.stat()
        assert stat.st_mode & 0o777 == 0o600


class TestLoadSavedCookies:
    def test_no_file(self, tmp_config_dir):
        assert load_saved_cookies() is None

    def test_invalid_json(self, tmp_config_dir):
        (tmp_config_dir / "cookies.json").write_text("not json")
        assert load_saved_cookies() is None

    def test_missing_a1(self, tmp_config_dir):
        (tmp_config_dir / "cookies.json").write_text('{"web_session": "x"}')
        assert load_saved_cookies() is None


class TestClearCookies:
    def test_clear_existing(self, tmp_config_dir):
        save_cookies({"a1": "test"})
        clear_cookies()
        assert load_saved_cookies() is None

    def test_clear_nonexistent(self, tmp_config_dir):
        # Should not raise
        clear_cookies()


class TestCookiesToString:
    def test_format(self):
        result = cookies_to_string({"a1": "v1", "web_session": "v2"})
        assert "a1=v1" in result
        assert "web_session=v2" in result
        assert "; " in result


class TestGetCookies:
    def test_prefers_saved_cookies_by_default(self, monkeypatch):
        monkeypatch.setattr("xhs_cli.cookies.load_saved_cookies", lambda: {"a1": "saved"})
        monkeypatch.setattr(
            "xhs_cli.cookies.extract_browser_cookies",
            lambda source: ("chrome", {"a1": "fresh"}),
        )

        browser, cookies = get_cookies("chrome")
        assert browser == "saved"
        assert cookies == {"a1": "saved"}

    def test_force_refresh_bypasses_saved_cookies(self, monkeypatch):
        monkeypatch.setattr("xhs_cli.cookies.load_saved_cookies", lambda: {"a1": "saved"})
        monkeypatch.setattr(
            "xhs_cli.cookies.extract_browser_cookies",
            lambda source: ("chrome", {"a1": "fresh"}),
        )
        saved = []
        monkeypatch.setattr("xhs_cli.cookies.save_cookies", lambda cookies: saved.append(cookies))

        browser, cookies = get_cookies("chrome", force_refresh=True)
        assert browser == "chrome"
        assert cookies == {"a1": "fresh"}
        assert saved == [{"a1": "fresh"}]


class TestNoteContextCache:
    def test_cache_persists_token_and_source(self, tmp_config_dir):
        cache_note_context("note-1", "token-1", "pc_search", context="search")

        assert get_cached_xsec_token("note-1") == "token-1"
        context = get_cached_note_context("note-1")
        assert context["token"] == "token-1"
        assert context["source"] == "pc_search"
        assert context["context"] == "search"

    def test_load_token_cache_keeps_legacy_entries_compatible(self, tmp_config_dir):
        get_token_cache_path().write_text('{"note-1":"legacy-token"}')

        cache = load_token_cache()
        assert cache["note-1"]["token"] == "legacy-token"
        assert cache["note-1"]["source"] == ""

    def test_expired_note_context_is_not_returned(self, tmp_config_dir):
        stale_ts = time.time() - NOTE_CONTEXT_TTL_SECONDS - 10
        get_token_cache_path().write_text(
            f'{{"note-1":{{"token":"stale-token","source":"pc_search","ts":{stale_ts}}}}}'
        )

        assert get_cached_note_context("note-1") == {}


class TestNoteIndexCache:
    def test_save_and_resolve_index_with_source(self, tmp_config_dir):
        save_note_index([
            {
                "note_id": "note-1",
                "xsec_token": "token-1",
                "xsec_source": "pc_search",
            }
        ])

        assert get_note_by_index(1) == {
            "note_id": "note-1",
            "xsec_token": "token-1",
            "xsec_source": "pc_search",
        }

    def test_save_empty_index_clears_previous_entries(self, tmp_config_dir):
        save_note_index([
            {
                "note_id": "note-1",
                "xsec_token": "token-1",
                "xsec_source": "pc_search",
            }
        ])
        save_note_index([])

        assert get_note_by_index(1) is None
        assert get_index_cache_path().read_text() == "[]"

    def test_index_file_permissions(self, tmp_config_dir):
        save_note_index([{"note_id": "note-1", "xsec_token": "", "xsec_source": ""}])

        stat = get_index_cache_path().stat()
        assert stat.st_mode & 0o777 == 0o600

    def test_index_normalizes_missing_optional_fields(self, tmp_config_dir):
        get_index_cache_path().write_text('[{"note_id":"note-1"}]')

        assert get_note_by_index(1) == {
            "note_id": "note-1",
            "xsec_token": "",
            "xsec_source": "",
        }

"""Unit tests for creator signing (no network required)."""

import base64
import json

from xhs_cli.creator_signing import _aes_encrypt, sign_creator


class TestAesEncrypt:
    def test_returns_hex(self):
        result = _aes_encrypt("hello world")
        assert isinstance(result, str)
        # Should be valid hex
        int(result, 16)

    def test_deterministic(self):
        r1 = _aes_encrypt("test data")
        r2 = _aes_encrypt("test data")
        assert r1 == r2

    def test_different_inputs(self):
        r1 = _aes_encrypt("hello")
        r2 = _aes_encrypt("world")
        assert r1 != r2


class TestSignCreator:
    def test_generates_headers(self):
        result = sign_creator("url=/api/test", None, "test_a1")
        assert "x-s" in result
        assert "x-t" in result

    def test_xs_prefix(self):
        result = sign_creator("url=/api/test", None, "test_a1")
        assert result["x-s"].startswith("XYW_")

    def test_xt_is_timestamp(self):
        result = sign_creator("url=/api/test", None, "test_a1")
        ts = int(result["x-t"])
        import time
        now_ms = int(time.time() * 1000)
        assert abs(ts - now_ms) < 5000

    def test_with_data(self):
        result = sign_creator(
            "url=/web_api/sns/v1/search/topic",
            {"keyword": "test"},
            "test_a1",
        )
        assert result["x-s"].startswith("XYW_")

    def test_envelope_structure(self):
        result = sign_creator("url=/api/test", None, "test_a1")
        # Decode the XYW_ envelope
        xs = result["x-s"]
        assert xs.startswith("XYW_")
        envelope_b64 = xs[4:]
        envelope = json.loads(base64.b64decode(envelope_b64))

        assert envelope["signSvn"] == "56"
        assert envelope["signType"] == "x2"
        assert envelope["appId"] == "ugc"
        assert envelope["signVersion"] == "1"
        assert "payload" in envelope

    def test_get_vs_post(self):
        # GET: data is None
        r1 = sign_creator("url=/api/test", None, "a1_val")
        # POST: data is provided
        r2 = sign_creator("url=/api/test", {"key": "val"}, "a1_val")
        # Should produce different signatures because content differs
        # (x-t might also differ so just check x-s)
        # The payloads should be different
        env1 = json.loads(base64.b64decode(r1["x-s"][4:]))
        env2 = json.loads(base64.b64decode(r2["x-s"][4:]))
        assert env1["payload"] != env2["payload"]

from xhs_cli.commands.creator import extract_hashtags


def test_extract_hashtags():
    # Normal case
    assert extract_hashtags("This is a #test and another #hashtag") == ["test", "hashtag"]

    # Empty body
    assert extract_hashtags("") == []

    # URL fragment shouldn't match
    assert extract_hashtags("Visit https://example.com#section for more info") == []

    # Consecutive tags without spaces — only the first is preceded by whitespace
    assert extract_hashtags("Mixed #one#two#three") == ["one"]

    # Tags at start of line
    assert extract_hashtags("#start of line") == ["start"]

    # Mix of languages
    assert extract_hashtags("测试 #中文标签 和 #english tag") == ["中文标签", "english"]

    # Trailing hashtag
    assert extract_hashtags("This is #trailing") == ["trailing"]

    # Pure hashtag body
    assert extract_hashtags("#a #b #c") == ["a", "b", "c"]

    # Emoji hashtag
    assert extract_hashtags("Let's go #🎉party") == ["🎉party"]

"""Unit tests for formatter (no network required)."""

from xhs_cli.formatter import coerce_int, extract_note_id, format_count, parse_note_reference


class TestFormatCount:
    def test_small_number(self):
        assert format_count(123) == "123"

    def test_wan(self):
        assert format_count(12345) == "1.2万"

    def test_yi(self):
        assert format_count(123456789) == "1.2亿"

    def test_string_input(self):
        assert format_count("5678") == "5678"

    def test_string_large(self):
        assert format_count("50000") == "5.0万"


class TestCoerceInt:
    def test_int_input(self):
        assert coerce_int(3) == 3

    def test_string_input(self):
        assert coerce_int("42") == 42

    def test_invalid_string_falls_back(self):
        assert coerce_int("10+") == 0


class TestExtractNoteId:
    def test_plain_id(self):
        assert extract_note_id("abc123def") == "abc123def"

    def test_explore_url(self):
        result = extract_note_id("https://www.xiaohongshu.com/explore/abc123def")
        assert result == "abc123def"

    def test_url_with_params(self):
        result = extract_note_id("https://www.xiaohongshu.com/explore/abc123?xsec_token=xxx")
        assert result == "abc123"

    def test_discovery_url(self):
        result = extract_note_id("https://www.xiaohongshu.com/discovery/item/abc123")
        assert result == "abc123"

    def test_trailing_slash(self):
        result = extract_note_id("https://www.xiaohongshu.com/explore/abc123/")
        assert result == "abc123"


class TestParseNoteReference:
    def test_extracts_token_and_source(self):
        note_id, token, source = parse_note_reference(
            "https://www.xiaohongshu.com/explore/abc123?xsec_token=token-1&xsec_source=pc_search"
        )
        assert note_id == "abc123"
        assert token == "token-1"
        assert source == "pc_search"

"""
Integration tests for XHS API client.

These tests require actual XHS cookies (a logged-in browser session).
They test against the real XHS API to verify the complete signing + request pipeline.

Run with: uv run pytest tests/test_integration.py -v
Skip if no cookies available.

Write operations are tested as self-cleaning PAIRS:
  - like + unlike → net effect = zero
  - favorite + unfavorite → net effect = zero
  - comment + delete-comment → net effect = zero
  The undo operation always runs in a `finally` block.
"""

import time

import pytest

from xhs_cli.client import XhsClient
from xhs_cli.cookies import get_cookies
from xhs_cli.exceptions import NoCookieError, SessionExpiredError, XhsApiError


def _get_test_cookies():
    """Try to get valid cookies for integration testing."""
    try:
        cookies = get_cookies("chrome")
        with XhsClient(cookies) as client:
            client.get_self_info()
        return cookies
    except SessionExpiredError:
        try:
            cookies = get_cookies("chrome", force_refresh=True)
            with XhsClient(cookies) as client:
                client.get_self_info()
            return cookies
        except (NoCookieError, SessionExpiredError, XhsApiError, Exception):
            return None
    except (NoCookieError, Exception):
        return None


# Skip all integration tests if no valid cookies are available
cookies = _get_test_cookies()
pytestmark = pytest.mark.skipif(cookies is None, reason="No valid XHS cookies available for integration testing")

# Cached values (discovered once, reused across tests)
_CACHED_NOTE_ID = None
_CACHED_XSEC_TOKEN = None
_CACHED_USER_ID = None
_CACHED_OTHER_USER_ID = None


def _ensure_test_note(client: XhsClient) -> tuple[str, str]:
    """Discover a note ID from search results for testing. Cached across tests."""
    global _CACHED_NOTE_ID, _CACHED_XSEC_TOKEN
    if _CACHED_NOTE_ID:
        return _CACHED_NOTE_ID, _CACHED_XSEC_TOKEN or ""

    search_data = client.search_notes("美食")
    items = search_data.get("items", [])
    if not items:
        pytest.skip("No search results available for testing")

    _CACHED_NOTE_ID = items[0].get("id", "")
    _CACHED_XSEC_TOKEN = items[0].get("xsec_token", "")
    if not _CACHED_NOTE_ID:
        pytest.skip("No note ID in search results")

    return _CACHED_NOTE_ID, _CACHED_XSEC_TOKEN or ""


def _ensure_self_user_id(client: XhsClient) -> str:
    """Get current user's ID. Cached across tests."""
    global _CACHED_USER_ID
    if _CACHED_USER_ID:
        return _CACHED_USER_ID

    info = client.get_self_info()
    _CACHED_USER_ID = info.get("user_id", "")
    if not _CACHED_USER_ID:
        pytest.skip("Cannot determine current user ID")

    return _CACHED_USER_ID


def _ensure_other_user_id(client: XhsClient) -> str:
    """Find another user ID for safe follow/unfollow testing."""
    global _CACHED_OTHER_USER_ID
    if _CACHED_OTHER_USER_ID:
        return _CACHED_OTHER_USER_ID

    self_user_id = _ensure_self_user_id(client)
    data = client.search_users("美食达人")
    users = data.get("user_info_dtos", []) if isinstance(data, dict) else []
    for item in users:
        user_id = item.get("user_base_dto", {}).get("user_id", "")
        if user_id and user_id != self_user_id:
            _CACHED_OTHER_USER_ID = user_id
            return user_id

    pytest.skip("No other user found for follow/unfollow testing")


@pytest.fixture
def client():
    """Create a test client with real cookies."""
    assert cookies is not None
    c = XhsClient(cookies)
    yield c
    c.close()


# ─── Auth ────────────────────────────────────────────────────────────────────


class TestAuth:
    """Test authentication and user info."""

    def test_get_self_info(self, client: XhsClient):
        info = client.get_self_info()
        assert isinstance(info, dict)
        assert any(k in info for k in ["nickname", "red_id", "user_id"])

    def test_cookies_have_a1(self):
        assert cookies is not None
        assert "a1" in cookies
        assert len(cookies["a1"]) > 0


# ─── Search ──────────────────────────────────────────────────────────────────


class TestSearch:
    """Test search functionality."""

    def test_search_notes(self, client: XhsClient):
        data = client.search_notes("美食", page=1)
        items = data.get("items", [])
        assert len(items) > 0

    def test_search_with_sort(self, client: XhsClient):
        time.sleep(1)
        data = client.search_notes("旅行", sort="popularity_descending")
        assert data is not None

    def test_search_returns_note_cards(self, client: XhsClient):
        time.sleep(1)
        data = client.search_notes("咖啡")
        items = data.get("items", [])
        if items:
            first = items[0]
            assert "note_card" in first or "model_type" in first

    def test_search_users(self, client: XhsClient):
        """Search for users by keyword."""
        time.sleep(1)
        data = client.search_users("美食达人")
        assert data is not None

    def test_search_topics(self, client: XhsClient):
        time.sleep(1)
        data = client.search_topics("美食")
        assert data is not None


# ─── Feed ────────────────────────────────────────────────────────────────────


class TestFeed:
    """Test feed functionality."""

    def test_get_home_feed(self, client: XhsClient):
        time.sleep(1)
        data = client.get_home_feed()
        items = data.get("items", [])
        assert len(items) > 0

    def test_feed_items_have_structure(self, client: XhsClient):
        time.sleep(1)
        data = client.get_home_feed()
        items = data.get("items", [])
        if items:
            assert "note_card" in items[0] or "id" in items[0]

    def test_hot_feed(self, client: XhsClient):
        """Hot/trending feed by category."""
        time.sleep(1)
        data = client.get_hot_feed("homefeed.food_v3")
        items = data.get("items", [])
        assert len(items) > 0


# ─── Note Reading ────────────────────────────────────────────────────────────


class TestNoteRead:
    """Test reading individual notes."""

    def test_read_note(self, client: XhsClient):
        time.sleep(1)
        note_id, xsec_token = _ensure_test_note(client)
        time.sleep(1)
        data = client.get_note_by_id(note_id, xsec_token=xsec_token)
        assert data is not None

    def test_get_comments(self, client: XhsClient):
        time.sleep(1)
        note_id, xsec_token = _ensure_test_note(client)
        time.sleep(1)
        data = client.get_comments(note_id, xsec_token=xsec_token)
        assert data is not None


# ─── User Content ────────────────────────────────────────────────────────────


class TestUserContent:
    """Test user content queries."""

    def test_get_user_info(self, client: XhsClient):
        time.sleep(1)
        user_id = _ensure_self_user_id(client)
        data = client.get_user_info(user_id)
        assert isinstance(data, dict)

    def test_get_user_notes(self, client: XhsClient):
        time.sleep(1)
        user_id = _ensure_self_user_id(client)
        data = client.get_user_notes(user_id)
        assert data is not None

    def test_get_user_favorites(self, client: XhsClient):
        time.sleep(1)
        user_id = _ensure_self_user_id(client)
        data = client.get_user_favorites(user_id)
        assert data is not None

    def test_get_creator_note_list(self, client: XhsClient):
        time.sleep(1)
        data = client.get_creator_note_list()
        assert data is not None
        assert "notes" in data
        assert "page" in data


# ─── Notifications ───────────────────────────────────────────────────────────


class TestNotifications:
    """Test notification endpoints (reverse-engineered)."""

    def test_unread_count(self, client: XhsClient):
        time.sleep(1)
        data = client.get_unread_count()
        assert isinstance(data, dict)

    def test_notification_mentions(self, client: XhsClient):
        time.sleep(1)
        data = client.get_notification_mentions()
        assert data is not None

    def test_notification_likes(self, client: XhsClient):
        time.sleep(1)
        data = client.get_notification_likes()
        assert data is not None

    def test_notification_connections(self, client: XhsClient):
        time.sleep(1)
        data = client.get_notification_connections()
        assert data is not None


# ─── Paired Write Tests (self-cleaning) ──────────────────────────────────────


class TestLikePair:
    """like + unlike → net effect = zero."""

    def test_like_then_unlike(self, client: XhsClient):
        time.sleep(1)
        note_id, _ = _ensure_test_note(client)

        time.sleep(1)
        try:
            result = client.like_note(note_id)
            assert result is not None or result is True
        finally:
            time.sleep(1)
            try:
                client.unlike_note(note_id)
            except XhsApiError:
                pass  # Best effort cleanup


class TestFavoritePair:
    """favorite + unfavorite → net effect = zero."""

    def test_favorite_then_unfavorite(self, client: XhsClient):
        time.sleep(1)
        note_id, _ = _ensure_test_note(client)

        time.sleep(1)
        try:
            result = client.favorite_note(note_id)
            assert result is not None or result is True
        finally:
            time.sleep(1)
            try:
                client.unfavorite_note(note_id)
            except XhsApiError:
                pass  # Best effort cleanup


class TestCommentPair:
    """comment + delete-comment → net effect = zero."""

    def test_comment_then_delete(self, client: XhsClient):
        time.sleep(1)
        note_id, _ = _ensure_test_note(client)

        time.sleep(1)
        comment_data = client.post_comment(note_id, "🤖 Integration test — will delete shortly")
        assert comment_data is not None

        # Extract comment ID from response
        comment_id = ""
        if isinstance(comment_data, dict):
            comment_id = (
                comment_data.get("comment", {}).get("id", "")
                or comment_data.get("id", "")
                or comment_data.get("comment_id", "")
            )

        try:
            assert comment_id, f"Should get comment ID back, got: {comment_data}"
        finally:
            if comment_id:
                time.sleep(1)
                try:
                    client.delete_comment(note_id, comment_id)
                except XhsApiError:
                    pass  # Best effort cleanup


# ─── End-to-End Workflows ────────────────────────────────────────────────────


class TestEndToEnd:
    """Full workflow tests combining multiple operations."""

    def test_search_read_comments_workflow(self, client: XhsClient):
        """search → read note → get comments."""
        time.sleep(1)
        result = client.search_notes("Python编程")
        items = result.get("items", [])
        if not items:
            pytest.skip("No search results")

        first = items[0]
        note_id = first.get("id", "")
        xsec_token = first.get("xsec_token", "")
        assert note_id

        time.sleep(1)
        note = client.get_note_by_id(note_id, xsec_token=xsec_token)
        assert note is not None

        time.sleep(1)
        comments = client.get_comments(note_id, xsec_token=xsec_token)
        assert comments is not None

    def test_like_favorite_workflow(self, client: XhsClient):
        """Like + favorite → unfavorite + unlike. Full interaction lifecycle."""
        time.sleep(1)
        note_id, _ = _ensure_test_note(client)

        liked = False
        favorited = False

        try:
            time.sleep(1)
            client.like_note(note_id)
            liked = True

            time.sleep(1)
            client.favorite_note(note_id)
            favorited = True
        finally:
            if favorited:
                time.sleep(1)
                try:
                    client.unfavorite_note(note_id)
                except XhsApiError:
                    pass
            if liked:
                time.sleep(1)
                try:
                    client.unlike_note(note_id)
                except XhsApiError:
                    pass


class TestFollowPair:
    """follow + unfollow → net effect = zero."""

    def test_follow_then_unfollow(self, client: XhsClient):
        time.sleep(1)
        user_id = _ensure_other_user_id(client)

        time.sleep(1)
        followed = False
        try:
            result = client.follow_user(user_id)
            followed = True
            assert isinstance(result, dict)
            assert result.get("fstatus") in {"follows", "followed"}
        finally:
            if followed:
                time.sleep(1)
                try:
                    undo = client.unfollow_user(user_id)
                    assert isinstance(undo, dict)
                    assert undo.get("fstatus") in {"none", "unfollowed"}
                except XhsApiError:
                    pass

    def test_notifications_workflow(self, client: XhsClient):
        """unread → mentions → likes → connections."""
        time.sleep(1)
        unread = client.get_unread_count()
        assert isinstance(unread, dict)

        time.sleep(1)
        assert client.get_notification_mentions(num=5) is not None

        time.sleep(1)
        assert client.get_notification_likes(num=5) is not None

        time.sleep(1)
        assert client.get_notification_connections(num=5) is not None