iOSBuildServer/tests/test_build_service.py

"""打包服务核心逻辑测试"""
import asyncio
import json
import os
import plistlib
import time
import shutil
import tempfile
from pathlib import Path
from unittest.mock import patch, AsyncMock, MagicMock

import pytest

from backend.services.build_service import (
    update_source,
    copy_source_code,
    prepare_source_snapshot,
    build_project,
    generate_config,
    _cleanup_old_builds,
    _resolve_provisioning_profile,
)
from backend.services.log_streamer import LogStreamer


@pytest.fixture
def log_streamer():
    return LogStreamer()


@pytest.fixture
def tmp_dirs(tmp_path):
    """创建临时源码目录和打包目录"""
    source_dir = tmp_path / "source"
    source_dir.mkdir()
    (source_dir / "readoor").mkdir()
    (source_dir / "readoor" / "AppDelegate.swift").write_text("// app")
    (source_dir / "readoor.xcodeproj").mkdir()
    (source_dir / "Pods").mkdir()
    (source_dir / "Podfile.lock").write_text("PODFILE CHECKSUM: abc")
    (source_dir / "readoor.xcworkspace").mkdir()
    (source_dir / "AutoPacking").mkdir()
    vendor_dir = source_dir / "Vendor" / "RDEpubReaderView"
    vendor_dir.mkdir(parents=True)
    (vendor_dir / "RDEpubReaderView.podspec").write_text("Pod::Spec.new do |s| end")
    (source_dir / "readoorTests").mkdir()
    (source_dir / "Podfile").write_text("pod 'AFNetworking'")

    build_dir = tmp_path / "build"
    return source_dir, build_dir


def _make_mock_process(returncode=0, output=b"ok\n"):
    """创建 mock 进程，支持 async for stdout"""
    proc = AsyncMock()
    proc.returncode = returncode
    proc.wait = AsyncMock(return_value=returncode)

    async def iter_stdout():
        yield output

    proc.stdout = iter_stdout()
    return proc


# ---- update_source ----

async def test_update_source_clone(tmp_path, log_streamer):
    """目录不存在时执行 clone"""
    source_dir = tmp_path / "branches" / "main"
    source_dir.parent.mkdir()

    with patch("backend.services.build_service.get_git_remote_url", return_value="git@github.com:test/repo.git"):
        with patch("asyncio.create_subprocess_exec", return_value=_make_mock_process()) as mock_exec:
            await update_source("t1", source_dir, "main")
            assert mock_exec.call_count == 6  # clone, set-url, fetch, checkout, reset, clean
            assert mock_exec.call_args_list[0].args[:4] == ("git", "-c", "credential.helper=", "clone")


async def test_update_source_clone_no_remote(tmp_path, log_streamer):
    """目录不存在且未配置远程仓库时抛异常"""
    source_dir = tmp_path / "branches" / "main"

    with patch("backend.services.build_service.get_git_remote_url", return_value=""):
        with pytest.raises(Exception, match="GIT_REMOTE_URL"):
            await update_source("t1", source_dir, "main")


async def test_update_source_existing(tmp_path, log_streamer):
    """目录已存在时执行 fetch + checkout + reset + clean"""
    source_dir = tmp_path / "branches" / "dev"
    source_dir.parent.mkdir(parents=True)
    source_dir.mkdir()
    (source_dir / ".git").mkdir()

    call_count = 0

    async def mock_exec(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        return _make_mock_process()

    with patch("backend.services.build_service.get_git_remote_url", return_value=""):
        with patch("asyncio.create_subprocess_exec", side_effect=mock_exec):
            await update_source("t1", source_dir, "dev")

    assert call_count == 4  # fetch, checkout, reset, clean


async def test_update_source_fetch_failure(tmp_path, log_streamer):
    """fetch 失败时抛异常"""
    source_dir = tmp_path / "branches" / "dev"
    source_dir.parent.mkdir(parents=True)
    source_dir.mkdir()
    (source_dir / ".git").mkdir()

    async def mock_exec(*args, **kwargs):
        return _make_mock_process(returncode=1, output=b"error\n")

    with patch("backend.services.build_service.get_git_remote_url", return_value=""):
        with patch("asyncio.create_subprocess_exec", side_effect=mock_exec):
            with pytest.raises(Exception, match="git fetch 失败"):
                await update_source("t1", source_dir, "dev")


async def test_prepare_source_snapshot_serializes_shared_source(tmp_dirs, log_streamer):
    """共享源码模式下，更新和复制必须在同一把锁内完成。"""
    source_dir, build_dir_parent = tmp_dirs
    task = MagicMock(branch="main")

    with patch("backend.services.build_service.get_git_remote_url", return_value="git@github.com:test/repo.git"), \
         patch("backend.services.build_service.get_shared_source_dir", return_value=source_dir), \
         patch("backend.services.build_service.BUILD_BASE_DIR", build_dir_parent), \
         patch("backend.services.build_service.update_source", new_callable=AsyncMock), \
         patch("backend.services.build_service.get_source_commit", new_callable=AsyncMock, return_value="abc123"):
        build_dir, commit = await prepare_source_snapshot("t1", task)

    assert build_dir.exists()
    assert commit == "abc123"


# ---- copy_source_code ----

async def test_copy_source_code(tmp_dirs, log_streamer):
    """正常拷贝源码"""
    source_dir, build_dir_parent = tmp_dirs

    task = MagicMock()
    task.branch = "main"

    with patch("backend.services.build_service.BUILD_BASE_DIR", build_dir_parent):
        result = await copy_source_code("t1", task, source_dir)

    assert result.exists()
    assert (result / "readoor" / "AppDelegate.swift").exists()
    assert (result / "Pods").exists()
    assert (result / "Podfile").exists()
    assert not (result / "podfile").exists()
    assert (result / "Podfile.lock").exists()
    assert (result / "readoor.xcworkspace").exists()
    assert (result / "Vendor" / "RDEpubReaderView" / "RDEpubReaderView.podspec").exists()
    assert not (result / "AutoPacking").exists()


async def test_copy_source_code_normalizes_lowercase_podfile(tmp_dirs, log_streamer):
    """旧分支的 podfile 也必须复制为 Pods 工程引用的 Podfile。"""
    source_dir, build_dir_parent = tmp_dirs
    (source_dir / "Podfile").unlink()
    (source_dir / "podfile").write_text("pod 'AFNetworking'")
    task = MagicMock(branch="main")

    with patch("backend.services.build_service.BUILD_BASE_DIR", build_dir_parent):
        result = await copy_source_code("t1", task, source_dir)

    copied_names = {path.name for path in result.iterdir()}
    assert "Podfile" in copied_names
    assert "podfile" not in copied_names


async def test_copy_source_code_excludes_git(tmp_dirs, log_streamer):
    """拷贝时排除 .git 目录"""
    source_dir, build_dir_parent = tmp_dirs
    (source_dir / "readoor" / ".git").mkdir()

    task = MagicMock()

    with patch("backend.services.build_service.BUILD_BASE_DIR", build_dir_parent):
        result = await copy_source_code("t1", task, source_dir)

    assert not (result / "readoor" / ".git").exists()


async def test_copy_source_code_overwrites_existing(tmp_dirs, log_streamer):
    """打包目录已存在时覆盖"""
    source_dir, build_dir_parent = tmp_dirs
    old_dir = build_dir_parent / "build_readoor_t1"
    old_dir.mkdir(parents=True)
    (old_dir / "old_file.txt").write_text("old")

    task = MagicMock()

    with patch("backend.services.build_service.BUILD_BASE_DIR", build_dir_parent):
        result = await copy_source_code("t1", task, source_dir)

    assert not (result / "old_file.txt").exists()
    assert (result / "readoor").exists()


async def test_build_project_passes_scheme_as_exec_argument(tmp_path, log_streamer):
    """Scheme 中的 shell 特殊字符只能作为 xcodebuild 参数，不能被执行。"""
    build_dir = tmp_path / "build"
    build_dir.mkdir()
    task = MagicMock()
    config_data = {"SCHEME": "App; touch /tmp/should-not-run"}

    with patch("asyncio.create_subprocess_exec", return_value=_make_mock_process()) as mock_exec, \
         patch("asyncio.create_subprocess_shell") as mock_shell:
        with pytest.raises(Exception, match="未找到 IPA 文件"):
            await build_project("t1", task, config_data, build_dir)

    assert mock_shell.call_count == 0
    archive_args = mock_exec.call_args_list[1].args
    assert archive_args[0:2] == ("xcodebuild", "archive")
    assert "App; touch /tmp/should-not-run" in archive_args


# ---- generate_config ----

async def test_generate_config(tmp_path, log_streamer):
    """生成打包配置"""
    build_dir = tmp_path / "build"
    build_dir.mkdir()

    task = MagicMock()
    task.app_id = "1"
    task.scheme_id = "1"
    task.build_type = "Ad_Hoc"
    task.obfuscation = True

    mock_config = {
        "apps": {
            "1": {
                "name": "测试App",
                "server": "测试环境",
                "AppGuid": "guid-123",
                "API": "https://test.api.com",
                "certificates": {
                    "Ad_Hoc": {"name": "com.test.app", "cer": "cert", "pro": "pro", "theme": "t"}
                },
            }
        },
        "schemes": {"1": {"name": "testScheme", "ossFloder": "test"}},
    }

    with patch("backend.routers.config.load_config", return_value=mock_config), \
         patch(
             "backend.services.build_service._resolve_provisioning_profile",
             return_value=("pro", "TEAM123"),
         ):
        result = await generate_config("t1", task, build_dir)

    assert result["APPID"] == "guid-123"
    assert result["SCHEME"] == "testScheme"
    assert result["BUILD_TYPE"] == "Ad_Hoc"
    assert result["BUNDLE_ID"] == "com.test.app"
    assert result["CERTIFICATE"] == "cert"
    assert result["PROVISIONING_NAME"] == "pro"
    assert result["TEAM_ID"] == "TEAM123"

    config_file = build_dir / "config_output.json"
    assert config_file.exists()
    saved = json.loads(config_file.read_text())
    assert saved["APPID"] == "guid-123"


def test_resolve_provisioning_profile_extracts_name_and_team(tmp_path):
    profile_path = tmp_path / "profile.mobileprovision"
    profile_path.write_bytes(b"signed profile")
    decoded = plistlib.dumps({"Name": "a4YX061_adHoc", "TeamIdentifier": ["MG4Z7FU83W"]})

    with patch("backend.services.build_service.subprocess.check_output", return_value=decoded):
        assert _resolve_provisioning_profile(str(profile_path)) == ("a4YX061_adHoc", "MG4Z7FU83W")


async def test_generate_config_uses_saved_version(tmp_path, log_streamer):
    """从服务配置读取版本号，不依赖分支源码脚本"""
    build_dir = tmp_path / "build"
    build_dir.mkdir()

    task = MagicMock()
    task.app_id = "1"
    task.scheme_id = "1"
    task.build_type = "Ad_Hoc"
    task.obfuscation = False

    mock_config = {
        "apps": {"1": {"name": "App", "certificates": {}}},
        "schemes": {"1": {"name": "sch", "ossFloder": "f"}},
        "versions": {"app_ver": "3.0.1", "build_ver": "3.0.1.0"},
    }

    with patch("backend.routers.config.load_config", return_value=mock_config):
        result = await generate_config("t1", task, build_dir)

    assert result["VERSION"] == "3.0.1"
    assert result["BUILD_VERSION"] == "3.0.1.0"


async def test_generate_config_uses_server_theme_not_source_autopacking(tmp_path, log_streamer):
    """Web 打包不读取分支源码中的 AutoPacking 目录。"""
    build_dir = tmp_path / "build"
    (build_dir / "AutoPacking" / "ymh").mkdir(parents=True)
    automation_dir = tmp_path / "automation"
    default_theme = automation_dir / "themes" / "ymh"
    default_theme.mkdir(parents=True)

    task = MagicMock(app_id="1", scheme_id="1", build_type="Ad_Hoc", obfuscation=False)
    mock_config = {
        "apps": {
            "1": {
                "name": "App",
                "certificates": {"Ad_Hoc": {"name": "com.example.app", "theme": "AutoPacking/ymh"}},
            }
        },
        "schemes": {"1": {"name": "sch", "ossFloder": "f"}},
    }

    with patch("backend.routers.config.load_config", return_value=mock_config), \
         patch("backend.services.build_service.AUTOMATION_DIR", automation_dir), \
         patch("backend.services.build_service.SKINS_DIR", tmp_path / "skins"):
        result = await generate_config("t1", task, build_dir)

    assert result["THEME"] == str(default_theme)


# ---- _cleanup_old_builds ----

def test_cleanup_old_builds(tmp_path):
    """清理过期打包目录"""
    from backend.database import SessionLocal, init_db
    from backend.models import BuildConfig

    init_db()
    db = SessionLocal()

    config = db.query(BuildConfig).filter(BuildConfig.key == "build_dir_retention_hours").first()
    if config:
        config.value = "1"
    else:
        db.add(BuildConfig(key="build_dir_retention_hours", value="1"))
    db.commit()

    old_dir = tmp_path / "build_readoor_old"
    old_dir.mkdir()
    # 设置修改时间为 24 小时前（确保超过 1 小时保留期）
    old_time = time.time() - 86400
    os.utime(old_dir, (old_time, old_time))

    with patch("backend.services.build_service.BUILD_BASE_DIR", tmp_path):
        _cleanup_old_builds(db)

    assert not old_dir.exists()
    db.close()


def test_cleanup_keeps_recent(tmp_path):
    """保留未过期的打包目录"""
    from backend.database import SessionLocal, init_db
    from backend.models import BuildConfig

    init_db()
    db = SessionLocal()

    config = db.query(BuildConfig).filter(BuildConfig.key == "build_dir_retention_hours").first()
    if config:
        config.value = "24"
    else:
        db.add(BuildConfig(key="build_dir_retention_hours", value="24"))
    db.commit()

    recent_dir = tmp_path / "build_readoor_recent"
    recent_dir.mkdir()

    with patch("backend.services.build_service.BUILD_BASE_DIR", tmp_path):
        _cleanup_old_builds(db)

    assert recent_dir.exists()
    db.close()