iOSBuildServer/tests/test_downloads.py

"""文件下载接口测试"""
import os
import tempfile
from unittest.mock import patch, AsyncMock

from backend.database import SessionLocal
from backend.models import Task


def _create_task_with_files(client, tmp_config, **extra_fields):
    """创建一个带文件路径的任务"""
    with patch("backend.services.build_queue.build_queue") as mock_queue:
        mock_queue.submit = AsyncMock()
        resp = client.post("/api/tasks", json={
            "app_id": "1",
            "build_type": "Ad_Hoc",
            "scheme_id": "1",
        })
        task_id = resp.json()["id"]

    # 直接更新数据库中的文件路径
    db = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()
    for k, v in extra_fields.items():
        setattr(task, k, v)
    db.commit()
    db.close()
    return task_id


# ---- dSYM 下载 ----

def test_download_dsym_not_exist(client, tmp_config):
    """dSYM 路径未设置"""
    task_id = _create_task_with_files(client, tmp_config)
    resp = client.get(f"/api/tasks/{task_id}/dsym")
    assert resp.status_code == 404


def test_download_dsym_file_missing(client, tmp_config):
    """dSYM 路径设置了但文件不存在"""
    task_id = _create_task_with_files(client, tmp_config, dsym_path="/tmp/nonexistent.dSYM")
    resp = client.get(f"/api/tasks/{task_id}/dsym")
    assert resp.status_code == 404
    assert "已被清理" in resp.json()["detail"]


def test_download_dsym_success(client, tmp_config):
    """dSYM 文件正常下载"""
    # 创建临时文件
    fd, path = tempfile.mkstemp(suffix=".dSYM")
    os.write(fd, b"fake dsym content")
    os.close(fd)

    try:
        task_id = _create_task_with_files(client, tmp_config, dsym_path=path)
        resp = client.get(f"/api/tasks/{task_id}/dsym")
        assert resp.status_code == 200
        assert resp.content == b"fake dsym content"
    finally:
        os.unlink(path)


def test_download_dsym_nonexistent_task(client, tmp_config):
    resp = client.get("/api/tasks/nonexistent/dsym")
    assert resp.status_code == 404


def test_download_dsym_remote_fallback(client, tmp_config):
    """本地文件已被清理，但保存了远端地址时应返回远端下载链接（而非直接重定向，
    因为该接口需要认证信息，浏览器无法直接跳转到 302 目标）"""
    task_id = _create_task_with_files(
        client, tmp_config,
        dsym_path="/tmp/nonexistent.dSYM",
        dsym_url="https://oss.example.com/ios-builds/iOS/app_1_0_0_main_adhoc.dSYM.zip",
    )
    with patch("backend.services.distribution.get_published_dsym_url", return_value="https://signed.example.com/app.dSYM.zip?signature=1"):
        resp = client.get(f"/api/tasks/{task_id}/dsym")
    assert resp.status_code == 200
    assert resp.json()["url"] == "https://signed.example.com/app.dSYM.zip?signature=1"


# ---- 混淆映射表下载 ----

def test_download_obfuscation_maps_not_exist(client, tmp_config):
    """混淆映射表路径未设置"""
    task_id = _create_task_with_files(client, tmp_config)
    resp = client.get(f"/api/tasks/{task_id}/obfuscation-maps")
    assert resp.status_code == 404


def test_download_obfuscation_maps_file_missing(client, tmp_config):
    """路径设置了但文件不存在"""
    task_id = _create_task_with_files(client, tmp_config, obfuscation_maps_path="/tmp/nonexistent")
    resp = client.get(f"/api/tasks/{task_id}/obfuscation-maps")
    assert resp.status_code == 404


def test_download_obfuscation_maps_single_file(client, tmp_config):
    """单文件下载"""
    fd, path = tempfile.mkstemp(suffix=".json")
    os.write(fd, b'{"mappings": []}')
    os.close(fd)

    try:
        task_id = _create_task_with_files(client, tmp_config, obfuscation_maps_path=path)
        resp = client.get(f"/api/tasks/{task_id}/obfuscation-maps")
        assert resp.status_code == 200
        assert resp.content == b'{"mappings": []}'
    finally:
        os.unlink(path)


def test_download_obfuscation_maps_directory(client, tmp_config):
    """目录打包成 zip 下载"""
    tmpdir = tempfile.mkdtemp()
    with open(os.path.join(tmpdir, "map.json"), "w") as f:
        f.write("{}")

    try:
        task_id = _create_task_with_files(
            client, tmp_config,
            obfuscation_maps_path=tmpdir,
            app_name="测试App",
        )
        resp = client.get(f"/api/tasks/{task_id}/obfuscation-maps")
        assert resp.status_code == 200
        assert resp.headers["content-type"] == "application/zip"
        # filename is URL-encoded, check the raw header
        disposition = resp.headers["content-disposition"]
        assert ".zip" in disposition
    finally:
        import shutil
        shutil.rmtree(tmpdir)


# ---- 二维码下载 ----

def test_download_qrcode_not_exist(client, tmp_config):
    """二维码路径未设置"""
    task_id = _create_task_with_files(client, tmp_config)
    resp = client.get(f"/api/tasks/{task_id}/qrcode")
    assert resp.status_code == 404


def test_download_qrcode_file_missing(client, tmp_config):
    """路径设置了但文件不存在"""
    task_id = _create_task_with_files(client, tmp_config, qr_code_path="/tmp/nonexistent.png")
    resp = client.get(f"/api/tasks/{task_id}/qrcode")
    assert resp.status_code == 404


def test_download_qrcode_success(client, tmp_config):
    """二维码正常下载"""
    fd, path = tempfile.mkstemp(suffix=".png")
    os.write(fd, b"\x89PNG fake")
    os.close(fd)

    try:
        task_id = _create_task_with_files(client, tmp_config, qr_code_path=path)
        resp = client.get(f"/api/tasks/{task_id}/qrcode")
        assert resp.status_code == 200
        assert resp.headers["content-type"] == "image/png"
    finally:
        os.unlink(path)