"""发布包边界与部署管理的回归测试；使用临时目录，不触碰共享数据。""" from __future__ import annotations from pathlib import Path import shutil import subprocess import sys import tarfile import pytest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "release")) import build as release_build import common as release_common import deploy as release_deploy @pytest.fixture def source(tmp_path): root = tmp_path / "source" for original in release_build.source_files(ROOT): dest = root / original.relative_to(ROOT) dest.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(original, dest) return root @pytest.mark.parametrize("value", ["../2.0.0", "2.0", "01.2.3", "2.0.0/evil", "2.0.0+local", ""]) def test_invalid_versions(value): with pytest.raises(ValueError): release_common.version(value) def test_archive_boundary_and_integrity(source, tmp_path): (source / ".env").write_text("secret") (source / "data").mkdir() (source / "data/private.sqlite").write_text("private") (source / "agents/.env").write_text("secret") archive = release_build.build(source, tmp_path / "dist", "2.1.0-rc.1", checks=False) assert archive.with_suffix(".gz.sha256").read_text().split()[0] == release_common.sha256(archive) extracted = tmp_path / "extract" with tarfile.open(archive) as tar: tar.extractall(extracted, filter="data") root = extracted / "vent-agent-2.1.0-rc.1-linux" manifest = release_common.validate_package(root) assert manifest["checks"] == "skipped" assert (root / "VERSION").read_text().strip() == "2.1.0-rc.1" assert not (root / ".env").exists() assert not (root / "data").exists() assert not (root / "agents/.env").exists() assert (root / "algorithms/characteristic_curve_recognition/service.py").is_file() assert (root / "algorithms/characteristic_curve_recognition/vendor/chart_axis_detector/current_image_pipeline.py").is_file() assert (source / "VERSION").read_text() != "2.1.0-rc.1\n" # 校验脚本本身不应生成清单外 __pycache__ 导致第二次校验失败。 for _ in range(2): subprocess.run([sys.executable, str(root / "release/verify.py")], check=True, capture_output=True) with pytest.raises(ValueError, match="不覆盖"): release_build.build(source, tmp_path / "dist", "2.1.0-rc.1", checks=False) (root / "main.py").write_text("tampered") with pytest.raises(ValueError, match="校验失败"): release_common.validate_package(root) def test_source_symlink_rejected(source, tmp_path): (source / "agents/injected.py").symlink_to(tmp_path / "outside.py") with pytest.raises(ValueError, match="符号链接"): release_build.source_files(source) def test_stale_lock_rejected(source): with (source / "requirements.txt").open("a") as stream: stream.write("\nnew-package==1.0\n") with pytest.raises(ValueError, match="已变化"): release_build.verify_lock(source, False) def test_git_is_optional(tmp_path, monkeypatch): def missing(*args, **kwargs): raise FileNotFoundError("git") monkeypatch.setattr(release_build.subprocess, "run", missing) assert release_build.git_info(tmp_path) == {"commit": None, "dirty": None} def test_extraneous_file_rejected(source, tmp_path): archive = release_build.build(source, tmp_path / "dist", "2.0.1", checks=False) with tarfile.open(archive) as tar: tar.extractall(tmp_path / "extract", filter="data") root = tmp_path / "extract/vent-agent-2.0.1-linux" (root / ".env").write_text("unexpected") with pytest.raises(ValueError, match="清单外"): release_common.validate_package(root) release_common.validate_package(root, installed=True) def test_instance_lock_prevents_activation(tmp_path): assert not release_deploy.running(tmp_path) with release_deploy.locked(tmp_path / "shared/run/instance.lock"): assert release_deploy.running(tmp_path) with pytest.raises(ValueError, match="先 stop"): release_deploy.activate(tmp_path, "2.0.0") assert not release_deploy.running(tmp_path) def test_reused_pid_not_signalled(monkeypatch): monkeypatch.setattr(release_deploy, "start_ticks", lambda pid: "new") assert not release_deploy.same_process({"pid": 123, "start_ticks": "old"}) assert not release_deploy.same_process({"pid": 1, "start_ticks": "new"}) assert release_deploy.same_process({"pid": 123, "start_ticks": "new"}) @pytest.fixture def package(source, tmp_path): archive = release_build.build(source, tmp_path / "dist", "2.0.1", checks=False) with tarfile.open(archive) as tar: tar.extractall(tmp_path / "extract", filter="data") return tmp_path / "extract/vent-agent-2.0.1-linux" def test_install_preserves_shared_and_activation(package, tmp_path, monkeypatch): prefix = tmp_path / "installed" shared = prefix / "shared" (shared / "data").mkdir(parents=True) (shared / ".env").write_text("existing-secret") (shared / "data/sentinel").write_text("keep") monkeypatch.setattr(release_deploy, "PACKAGE", package) def fake_run(command, **kwargs): if "venv" in command: python = Path(command[-1]) / "bin/python" python.parent.mkdir(parents=True) python.write_text("test") return subprocess.CompletedProcess(command, 0) monkeypatch.setattr(release_deploy.subprocess, "run", fake_run) release_deploy.install(prefix) target = prefix / "releases/2.0.1" assert (target / ".env").resolve() == shared / ".env" assert (target / "data").resolve() == shared / "data" assert (shared / ".env").read_text() == "existing-secret" assert (shared / "data/sentinel").read_text() == "keep" assert not (prefix / "current").exists() with pytest.raises(ValueError, match="不覆盖"): release_deploy.install(prefix) release_deploy.activate(prefix, "2.0.1") assert release_deploy.current_release(prefix) == target def test_failed_install_cleans_only_new_version(package, tmp_path, monkeypatch): prefix = tmp_path / "installed" shared = prefix / "shared" shared.mkdir(parents=True) (shared / ".env").write_text("keep") monkeypatch.setattr(release_deploy, "PACKAGE", package) def fail(command, **kwargs): raise subprocess.CalledProcessError(1, command) monkeypatch.setattr(release_deploy.subprocess, "run", fail) with pytest.raises(subprocess.CalledProcessError): release_deploy.install(prefix) assert not (prefix / "releases/2.0.1").exists() assert (shared / ".env").read_text() == "keep"