wscraper/tests/e2e/test_happyfappy_live.py

from __future__ import annotations

import json
import os
import subprocess
import sys
import time
from pathlib import Path

import pytest


pytestmark = [pytest.mark.e2e]


def _e2e_enabled() -> bool:
    return os.getenv("WSCRAPER_E2E", "").strip() == "1"


def _base_env() -> dict[str, str]:
    env = os.environ.copy()
    src_path = str(Path.cwd() / "src")
    current_pythonpath = env.get("PYTHONPATH", "").strip()
    env["PYTHONPATH"] = f"{src_path}{os.pathsep}{current_pythonpath}" if current_pythonpath else src_path
    return env


def _log(tr, message: str, kind: str = "info") -> None:
    icon = "•"
    style: dict[str, bool] = {}
    if kind == "ok":
        icon = "✅"
        style = {"green": True}
    elif kind == "err":
        icon = "❌"
        style = {"red": True}
    elif kind == "warn":
        icon = "⚠️"
        style = {"yellow": True}
    elif kind == "run":
        icon = "🚀"
        style = {"cyan": True}

    if tr is not None:
        tr.write_line(f"{icon} {message}", **style)
    else:
        print(f"{icon} {message}")


def _run_cli_live(args: list[str], tr, timeout: int = 900) -> tuple[int, str]:
    cmd = [sys.executable, "-m", "wscraper"] + args
    _log(tr, f"Running: {' '.join(cmd)}", kind="run")

    started = time.time()
    proc = subprocess.Popen(
        cmd,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        env=_base_env(),
    )

    output_lines: list[str] = []
    assert proc.stdout is not None
    for line in proc.stdout:
        output_lines.append(line)
        clean = line.rstrip("\n")
        if clean:
            if tr is not None:
                tr.write_line(f"   {clean}")
            else:
                print(f"   {clean}")

    return_code = proc.wait(timeout=timeout)
    duration = time.time() - started
    if return_code == 0:
        _log(tr, f"Command finished successfully in {duration:.2f}s", kind="ok")
    else:
        _log(tr, f"Command failed with exit code {return_code} in {duration:.2f}s", kind="err")

    return return_code, "".join(output_lines)


@pytest.fixture
def tr(request):
    return request.config.pluginmanager.getplugin("terminalreporter")


@pytest.mark.skipif(not _e2e_enabled(), reason="Set WSCRAPER_E2E=1 to run live tests")
def test_get_bookmarks_live(tmp_path: Path, tr) -> None:
    cookie_file = Path(os.getenv("WSCRAPER_COOKIE_FILE", "cookies.txt"))
    if not cookie_file.exists():
        pytest.skip(f"Cookie file not found: {cookie_file}")

    output_file = tmp_path / "bookmarks.json"
    _log(tr, f"Output file: {output_file}")

    return_code, output_text = _run_cli_live(
        [
            "happyfappy",
            "--action",
            "get-bookmarks",
            "-c",
            str(cookie_file),
            "-o",
            str(output_file),
        ],
        tr,
    )
    assert return_code == 0, f"CLI failed:\n{output_text}"
    assert output_file.exists(), "bookmarks.json was not created"

    data = json.loads(output_file.read_text(encoding="utf-8"))
    assert isinstance(data, list), "bookmarks output must be a JSON list"
    assert len(data) >= 1, "expected at least one bookmark record"
    _log(tr, f"Extracted records: {len(data)}", kind="ok")

    first = data[0]
    assert isinstance(first, dict), "bookmark entry must be an object"
    for required_key in ("pageURL", "isVR", "title", "backgroundImage"):
        assert required_key in first, f"missing key: {required_key}"
    assert isinstance(first["pageURL"], str) and first["pageURL"].startswith("http")
    assert isinstance(first["isVR"], bool)
    assert isinstance(first["title"], str) and first["title"].strip() != ""


@pytest.mark.skipif(not _e2e_enabled(), reason="Set WSCRAPER_E2E=1 to run live tests")
def test_download_torrent_file_live(tmp_path: Path, tr) -> None:
    cookie_file = Path(os.getenv("WSCRAPER_COOKIE_FILE", "cookies.txt"))
    if not cookie_file.exists():
        pytest.skip(f"Cookie file not found: {cookie_file}")

    test_url = os.getenv(
        "WSCRAPER_TEST_TORRENT_URL",
        "https://www.happyfappy.net/torrents.php?id=110178",
    )
    output_dir = tmp_path / "torrent"
    _log(tr, f"Output dir: {output_dir}")

    return_code, output_text = _run_cli_live(
        [
            "happyfappy",
            "--action",
            "download-torrent-files",
            "-u",
            test_url,
            "-c",
            str(cookie_file),
            "-o",
            str(output_dir),
        ],
        tr,
    )
    assert return_code == 0, f"CLI failed:\n{output_text}"
    assert output_dir.exists(), "torrent output directory was not created"

    torrent_files = list(output_dir.glob("*.torrent"))
    assert len(torrent_files) >= 1, "expected at least one .torrent file"
    _log(tr, f"Downloaded .torrent files: {len(torrent_files)}", kind="ok")

    content = torrent_files[0].read_bytes()
    assert content.startswith(b"d"), "torrent file should start with bencode dictionary token 'd'"
    assert b"4:info" in content[:4096], "torrent file should include 'info' dictionary marker"