shenleiandClaude Opus 5
6867a3cb01
每个分支各有一条独立轨道（轨道名即分支名，master/main 沿用历史名 release）， 轨道类型 kind 只决定构建号行为：release 上架自增、feature 自测不自增。 - branch_track 映射分支 → 轨道名，新增分支自动建同名轨道 - 迁移：旧的共用 release / feature 轨按分支拆开并复制已用构建号， 拆完没有分支引用的旧共用轨直接删除；无自测分支时保留 feature 轨作模板 - 分支管理页的下拉只改轨道类型（是否自增），不再影响版本号归属 - 打包设置页按分支分块填写版本号 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJaZ913U5GVdTkYHiRpvDU
568 lines
21 KiB
Python
568 lines
21 KiB
Python
"""打包服务核心逻辑测试"""
|
|
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,
|
|
run_pod_install,
|
|
_patch_afnetworking_private_headers,
|
|
_cleanup_old_builds,
|
|
_mark_superseded_tasks,
|
|
_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()
|
|
whitelist_script = source_dir / "AutoPacking" / "obfuscation" / "generate_image_whitelist.py"
|
|
whitelist_script.parent.mkdir(parents=True)
|
|
whitelist_script.write_text("# whitelist generator")
|
|
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 (result / "AutoPacking" / "obfuscation" / "generate_image_whitelist.py").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()
|
|
|
|
|
|
def test_patch_afnetworking_private_headers(tmp_path):
|
|
"""pod install 后移除新版 Xcode 禁止直接引用的 netinet6 私有头。"""
|
|
source_dir = tmp_path / "Pods" / "AFNetworking" / "AFNetworking"
|
|
source_dir.mkdir(parents=True)
|
|
source_file = source_dir / "AFNetworkReachabilityManager.m"
|
|
source_file.write_text(
|
|
"#import <netinet/in.h>\n#import <netinet6/in6.h>\n#import <arpa/inet.h>\n"
|
|
)
|
|
source_file.chmod(0o444)
|
|
|
|
assert _patch_afnetworking_private_headers(tmp_path) == 1
|
|
assert "#import <netinet6/in6.h>" not in source_file.read_text()
|
|
assert "#import <netinet/in.h>" in source_file.read_text()
|
|
assert source_file.stat().st_mode & 0o777 == 0o444
|
|
|
|
|
|
async def test_run_pod_install_patches_afnetworking(tmp_path, log_streamer):
|
|
source_dir = tmp_path / "Pods" / "AFNetworking"
|
|
source_dir.mkdir(parents=True)
|
|
source_file = source_dir / "AFHTTPSessionManager.m"
|
|
source_file.write_text("#import <netinet6/in6.h>\n")
|
|
|
|
with patch("asyncio.create_subprocess_exec", return_value=_make_mock_process()):
|
|
await run_pod_install("t1", tmp_path)
|
|
|
|
assert "netinet6/in6.h" not in source_file.read_text()
|
|
|
|
|
|
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")
|
|
|
|
|
|
def _versions_config(tracks: dict) -> dict:
|
|
"""构造带版本轨道的配置；tracks 为 轨道名 -> {kind, app_ver, build_map}。"""
|
|
return {
|
|
"apps": {"1": {"name": "App", "certificates": {}}},
|
|
"schemes": {"1": {"name": "sch", "ossFloder": "f"}},
|
|
"versions": {"tracks": tracks},
|
|
}
|
|
|
|
|
|
def _release(app_ver: str, build_map: dict = None) -> dict:
|
|
return {"kind": "release", "app_ver": app_ver, "build_map": build_map or {}}
|
|
|
|
|
|
def _feature(app_ver: str, build_map: dict = None) -> dict:
|
|
return {"kind": "feature", "app_ver": app_ver, "build_map": build_map or {}}
|
|
|
|
|
|
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
|
|
task.branch = "master"
|
|
|
|
mock_config = _versions_config({"release": _release("3.0.1"), "feature": _feature("9.9.9")})
|
|
|
|
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_appstore_increments_per_version(tmp_path, log_streamer):
|
|
"""App_Store 打包时，按版本号在其自身已用构建号上 +1（回退旧版本号亦从自身最大值继续）"""
|
|
build_dir = tmp_path / "build"
|
|
build_dir.mkdir()
|
|
|
|
task = MagicMock(app_id="1", scheme_id="1", build_type="App_Store",
|
|
obfuscation=False, branch="master")
|
|
|
|
# 2.195.0 已打到 .5，切到 2.196.0 后又回退到 2.195.0
|
|
mock_config = _versions_config({
|
|
"release": _release("2.195.0", {"2.195.0": 5, "2.196.0": 3}),
|
|
"feature": _feature("2.195.0"),
|
|
})
|
|
|
|
with patch("backend.routers.config.load_config", return_value=mock_config), \
|
|
patch("backend.routers.config.save_config") as save_mock:
|
|
result = await generate_config("t1", task, build_dir)
|
|
|
|
# 从 2.195.0 自身的 5 继续到 6，不受 2.196.0 影响，也不撞已传的 .5
|
|
release_map = mock_config["versions"]["tracks"]["release"]["build_map"]
|
|
assert result["BUILD_VERSION"] == "2.195.0.6"
|
|
assert release_map["2.195.0"] == 6
|
|
assert release_map["2.196.0"] == 3
|
|
save_mock.assert_called()
|
|
|
|
|
|
async def test_generate_config_release_branches_are_independent(tmp_path, log_streamer):
|
|
"""master 与 develop 各用自己的轨道，版本号与构建号互不影响"""
|
|
build_dir = tmp_path / "build"
|
|
build_dir.mkdir()
|
|
|
|
task = MagicMock(app_id="1", scheme_id="1", build_type="App_Store",
|
|
obfuscation=False, branch="develop")
|
|
|
|
mock_config = _versions_config({
|
|
"release": _release("2.195.0", {"2.195.0": 5}), # master
|
|
"develop": _release("2.196.0", {"2.196.0": 1}),
|
|
"feature": _feature("2.100.0"),
|
|
})
|
|
|
|
with patch("backend.routers.config.load_config", return_value=mock_config), \
|
|
patch("backend.routers.config.save_config"):
|
|
result = await generate_config("t1", task, build_dir)
|
|
|
|
assert result["VERSION"] == "2.196.0"
|
|
assert result["BUILD_VERSION"] == "2.196.0.2"
|
|
# master 的 release 轨完全不受影响
|
|
assert mock_config["versions"]["tracks"]["release"]["build_map"] == {"2.195.0": 5}
|
|
|
|
|
|
async def test_generate_config_build_no_skips_other_release_track(tmp_path, log_streamer):
|
|
"""两个上架分支临时用同一 App_Ver 时，构建号跨轨道取最大值 +1，避免 App Store 重号"""
|
|
build_dir = tmp_path / "build"
|
|
build_dir.mkdir()
|
|
|
|
task = MagicMock(app_id="1", scheme_id="1", build_type="App_Store",
|
|
obfuscation=False, branch="develop")
|
|
|
|
mock_config = _versions_config({
|
|
"release": _release("2.196.0", {"2.196.0": 7}), # master
|
|
"develop": _release("2.196.0", {"2.196.0": 2}),
|
|
"feature": _feature("2.100.0"),
|
|
})
|
|
|
|
with patch("backend.routers.config.load_config", return_value=mock_config), \
|
|
patch("backend.routers.config.save_config"):
|
|
result = await generate_config("t1", task, build_dir)
|
|
|
|
assert result["BUILD_VERSION"] == "2.196.0.8"
|
|
assert mock_config["versions"]["tracks"]["develop"]["build_map"]["2.196.0"] == 8
|
|
|
|
|
|
async def test_generate_config_feature_branch_never_increments(tmp_path, log_streamer):
|
|
"""自测分支用自己那条轨道的版本号，且构建号永不自增（即使是 App_Store 打包）"""
|
|
build_dir = tmp_path / "build"
|
|
build_dir.mkdir()
|
|
|
|
task = MagicMock(app_id="1", scheme_id="1", build_type="App_Store",
|
|
obfuscation=False, branch="feature/login")
|
|
|
|
mock_config = _versions_config({
|
|
"release": _release("2.195.0", {"2.195.0": 5}),
|
|
"feature/login": _feature("2.100.0"),
|
|
})
|
|
|
|
with patch("backend.routers.config.load_config", return_value=mock_config), \
|
|
patch("backend.routers.config.save_config") as save_mock:
|
|
result = await generate_config("t1", task, build_dir)
|
|
|
|
assert result["VERSION"] == "2.100.0"
|
|
assert result["BUILD_VERSION"] == "2.100.0.0"
|
|
assert mock_config["versions"]["tracks"]["feature/login"]["build_map"] == {}
|
|
save_mock.assert_not_called()
|
|
|
|
|
|
async def test_generate_config_branch_track_override(tmp_path, log_streamer):
|
|
"""branch_track 中的显式配置优先于按分支名推断"""
|
|
build_dir = tmp_path / "build"
|
|
build_dir.mkdir()
|
|
|
|
task = MagicMock(app_id="1", scheme_id="1", build_type="Ad_Hoc",
|
|
obfuscation=False, branch="release/2.196")
|
|
|
|
mock_config = _versions_config({
|
|
"release": _release("2.196.0"),
|
|
"feature/login": _feature("2.100.0"),
|
|
})
|
|
# release/2.196 按名字本应是上架轨，显式指到 feature/login 那条轨道
|
|
mock_config["branch_track"] = {"release/2.196": "feature/login"}
|
|
|
|
with patch("backend.routers.config.load_config", return_value=mock_config):
|
|
result = await generate_config("t1", task, build_dir)
|
|
|
|
assert result["VERSION"] == "2.100.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()
|
|
|
|
|
|
def test_mark_superseded_tasks_for_same_oss_url():
|
|
from backend.database import SessionLocal, init_db
|
|
from backend.models import Task
|
|
init_db()
|
|
db = SessionLocal()
|
|
old = Task(id="old", app_id="1", app_name="测试App", build_type="Ad_Hoc", scheme_id="1", scheme_name="readoor31", status="completed", oss_url="https://files.example.com/app.html")
|
|
latest = Task(id="latest", app_id="1", app_name="测试App", build_type="Ad_Hoc", scheme_id="1", scheme_name="readoor31", status="completed", oss_url="https://files.example.com/app.html")
|
|
db.add_all([old, latest])
|
|
db.commit()
|
|
|
|
_mark_superseded_tasks(db, latest)
|
|
assert db.get(Task, "old").superseded_at is not None
|
|
assert db.get(Task, "latest").superseded_at is None
|
|
db.close()
|