iOSBuildServer/tests/test_build_service.py

"""打包服务核心逻辑测试"""
import asyncio
import json
import os
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,
    generate_config,
    _cleanup_old_builds,
)
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()
    (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.GIT_REMOTE_URL", "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")
            mock_exec.assert_called_once()
            args = mock_exec.call_args[0]
            assert "git" in args
            assert "clone" in args


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

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


async def test_update_source_pull_existing(tmp_path, log_streamer):
    """目录已存在时执行 fetch + checkout + pull"""
    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("asyncio.create_subprocess_exec", side_effect=mock_exec):
        await update_source("t1", source_dir, "dev")

    assert call_count == 3  # fetch, checkout, pull


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()

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

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


# ---- 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.lock").exists()
    assert (result / "readoor.xcworkspace").exists()


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()


# ---- 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.services.build_service.AUTOPACKING_DIR", tmp_path / "AutoPacking"):
        with patch("backend.routers.config.load_config", return_value=mock_config):
            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"

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


async def test_generate_config_reads_version(tmp_path, log_streamer):
    """从 start_build_app.py 读取版本号"""
    build_dir = tmp_path / "build"
    build_dir.mkdir()

    autopacking = tmp_path / "AutoPacking"
    autopacking.mkdir()
    (autopacking / "start_build_app.py").write_text(
        'App_Ver = "3.0.1"\nBuild_Ver = "3.0.1.0"\n'
    )

    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"}},
    }

    with patch("backend.services.build_service.AUTOPACKING_DIR", autopacking):
        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"


# ---- _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()