vent_agent_2.0/tests/test_release.py

"""发布包边界与部署管理的回归测试；使用临时目录，不触碰共享数据。"""
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"