224 lines
8.4 KiB
Python
224 lines
8.4 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from app.ota.diagnostics import (
|
|
MAX_FAILURE_LOG_BYTES,
|
|
DiagnosticLog,
|
|
bounded_failure_log,
|
|
clear_failure_log,
|
|
failure_log_metadata,
|
|
failure_log_path,
|
|
persist_failure_log,
|
|
read_failure_log,
|
|
)
|
|
from app.ota.state import read_last_result
|
|
from scripts import ota_worker
|
|
|
|
|
|
SOURCE_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_failure_log_normalizes_invalid_utf8_and_truncates_head_and_tail():
|
|
body = b"HEADER\xff\n" + (b"middle\n" * 200_000) + b"LATEST FAILURE\n"
|
|
rendered = bounded_failure_log(body)
|
|
assert len(rendered) <= MAX_FAILURE_LOG_BYTES
|
|
text = rendered.decode("utf-8")
|
|
assert "HEADER�" in text
|
|
assert "日志已截断" in text
|
|
assert text.endswith("LATEST FAILURE\n")
|
|
|
|
|
|
def test_failure_log_is_atomically_replaced_gated_by_result_and_cleared(tmp_path):
|
|
data_root = tmp_path / "data"
|
|
runtime_log = tmp_path / "ota-worker.log"
|
|
runtime_log.write_bytes(b"first")
|
|
target = persist_failure_log(data_root, runtime_log)
|
|
failed = {"status": "failed"}
|
|
assert target == failure_log_path(data_root)
|
|
assert failure_log_metadata(data_root, failed) == (True, 5)
|
|
assert read_failure_log(data_root, failed) == b"first"
|
|
assert read_failure_log(data_root, {"status": "success"}) is None
|
|
|
|
runtime_log.write_bytes(b"second failure")
|
|
persist_failure_log(data_root, runtime_log)
|
|
assert read_failure_log(data_root, failed) == b"second failure"
|
|
assert list(target.parent.glob("*")) == [target]
|
|
|
|
clear_failure_log(data_root)
|
|
assert failure_log_metadata(data_root, failed) == (False, 0)
|
|
|
|
|
|
def test_diagnostic_command_captures_success_failure_non_utf8_and_start_error(tmp_path):
|
|
log = DiagnosticLog(tmp_path / "worker.log")
|
|
log.reset()
|
|
result = log.run([sys.executable, "-c", "print('pytest output')"], failure_label="success")
|
|
assert result.returncode == 0
|
|
|
|
with pytest.raises(RuntimeError, match="7"):
|
|
log.run(
|
|
[sys.executable, "-c", "import os,sys; os.write(1, b'bad-\\xff'); sys.exit(7)"],
|
|
failure_label="single test",
|
|
)
|
|
with pytest.raises(RuntimeError, match="无法启动"):
|
|
log.run([str(tmp_path / "missing-command")], failure_label="startup")
|
|
|
|
text = log.path.read_text(encoding="utf-8")
|
|
assert "pytest output" in text
|
|
assert "bad-�" in text
|
|
assert "exit_code=7" in text
|
|
assert "command_start_failed=" in text
|
|
|
|
secret_log = DiagnosticLog(tmp_path / "redacted.log")
|
|
secret_log.reset()
|
|
secret_log.run(
|
|
[sys.executable, "-c", "print('ssid=private-network ip=192.0.2.10')"],
|
|
log_output=False,
|
|
)
|
|
redacted = secret_log.path.read_text(encoding="utf-8")
|
|
assert "private-network" not in redacted
|
|
assert "192.0.2.10" not in redacted
|
|
assert "已按脱敏规则隐藏" in redacted
|
|
|
|
|
|
def test_legacy_103_worker_environment_keeps_all_test_temporary_paths_in_transaction(tmp_path):
|
|
transaction = tmp_path / "work.legacy103"
|
|
environment = os.environ.copy()
|
|
for name in ("MATRIX_TEST_ROOT", "TMPDIR", "TEMP", "TMP", "PYTEST_ADDOPTS"):
|
|
environment.pop(name, None)
|
|
environment.update(
|
|
MATRIX_DATA_DIR=str(transaction / "test-data"),
|
|
MATRIX_RUNTIME_DIR=str(transaction / "test-runtime"),
|
|
MATRIX_SOURCE_ONLY_UPDATE_TESTS="1",
|
|
MATRIX_LEGACY_ISOLATION_PROBE="1",
|
|
PYTHONPATH=str(SOURCE_ROOT),
|
|
)
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"pytest",
|
|
"-p",
|
|
"no:cacheprovider",
|
|
"-q",
|
|
"tests/test_ota_isolation_probe.py",
|
|
"--tb=short",
|
|
"--disable-warnings",
|
|
],
|
|
cwd=SOURCE_ROOT,
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
assert "1 passed" in result.stdout
|
|
assert not (transaction.parent / ".pytest_cache").exists()
|
|
|
|
strict_environment = dict(environment, MATRIX_STRICT_PYTEST_FAILURE_PROBE="1")
|
|
strict = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"pytest",
|
|
"-p",
|
|
"no:cacheprovider",
|
|
"-q",
|
|
"tests/test_ota_isolation_probe.py",
|
|
"--tb=short",
|
|
"--disable-warnings",
|
|
],
|
|
cwd=SOURCE_ROOT,
|
|
env=strict_environment,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
)
|
|
assert strict.returncode != 0
|
|
assert "strict legacy-worker pytest probe executed" in strict.stdout
|
|
|
|
|
|
def test_development_source_contains_no_diagnostic_pytest_bypass():
|
|
conftest = (SOURCE_ROOT / "tests" / "conftest.py").read_text(encoding="utf-8")
|
|
retired_marker = "OTA_" + "DIAGNOSTIC_" + "BOOTSTRAP"
|
|
retired_symbol = "diagnostic_" + "bootstrap"
|
|
pytest_early_exit = "pytest" + ".exit"
|
|
retired_override = "MATRIX_FORCE_" + "LEGACY_ISOLATION_PROBE"
|
|
assert not (SOURCE_ROOT / retired_marker).exists()
|
|
assert retired_symbol not in conftest
|
|
assert pytest_early_exit not in conftest
|
|
assert retired_override not in conftest
|
|
|
|
|
|
def test_worker_and_manual_entry_keep_strict_pytest_and_transaction_paths():
|
|
worker = (SOURCE_ROOT / "scripts" / "ota_worker.py").read_text(encoding="utf-8")
|
|
manual = (SOURCE_ROOT / "scripts" / "update_walnutpi.sh").read_text(encoding="utf-8")
|
|
for option in ("no:cacheprovider", "--basetemp", "--tb=short", "--disable-warnings"):
|
|
assert option in worker
|
|
assert option in manual
|
|
for variable in ("MATRIX_TEST_ROOT", "TMPDIR", "MATRIX_SOURCE_ONLY_UPDATE_TESTS"):
|
|
assert variable in worker
|
|
assert variable in manual
|
|
assert "check=False" not in worker.split('failure_label="新版本 pytest"')[0].rsplit("self.run", 1)[1]
|
|
|
|
|
|
def test_pre_switch_failure_preserves_installed_state_and_records_rollback_problem(tmp_path, monkeypatch):
|
|
target = tmp_path / "opt" / "matrix-screen-controller"
|
|
releases = tmp_path / "opt" / "releases"
|
|
data_root = tmp_path / "var" / "data"
|
|
runtime_root = tmp_path / "run"
|
|
work_root = tmp_path / "opt" / "ota"
|
|
unit = tmp_path / "etc" / "matrix-screen-controller.service"
|
|
package = tmp_path / "uploaded.ota"
|
|
for directory in (target, releases, data_root, runtime_root, work_root, unit.parent):
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
(target / "old-program").write_text("unchanged", encoding="utf-8")
|
|
(data_root / "user-data").write_text("unchanged", encoding="utf-8")
|
|
unit.write_text("old unit", encoding="utf-8")
|
|
package.write_bytes(b"candidate")
|
|
|
|
monkeypatch.setattr(ota_worker, "TARGET", target)
|
|
monkeypatch.setattr(ota_worker, "RELEASES", releases)
|
|
monkeypatch.setattr(ota_worker, "DATA_ROOT", data_root)
|
|
monkeypatch.setattr(ota_worker, "RUNTIME_ROOT", runtime_root)
|
|
monkeypatch.setattr(ota_worker, "OTA_WORK_ROOT", work_root)
|
|
monkeypatch.setattr(ota_worker, "UNIT_PATH", unit)
|
|
monkeypatch.setattr(ota_worker, "REQUEST_PATH", runtime_root / "ota-request.json")
|
|
request = {
|
|
"job_id": "rollback-test",
|
|
"current_version": "1.0.4",
|
|
"target_version": "1.0.5",
|
|
"package_path": str(package),
|
|
"status_path": str(runtime_root / "ota-status.json"),
|
|
"request_path": str(runtime_root / "ota-request.json"),
|
|
}
|
|
transaction = ota_worker.Transaction(request)
|
|
transaction.work_root.mkdir(parents=True)
|
|
transaction.release.mkdir(parents=True)
|
|
transaction.data_candidate.mkdir(parents=True)
|
|
monkeypatch.setattr(
|
|
transaction,
|
|
"stop_visual",
|
|
lambda: (_ for _ in ()).throw(RuntimeError("simulated rollback fault")),
|
|
)
|
|
|
|
transaction.rollback(RuntimeError("pytest failed"))
|
|
|
|
assert (target / "old-program").read_text(encoding="utf-8") == "unchanged"
|
|
assert (data_root / "user-data").read_text(encoding="utf-8") == "unchanged"
|
|
assert unit.read_text(encoding="utf-8") == "old unit"
|
|
assert not transaction.release.exists()
|
|
assert not transaction.data_candidate.exists()
|
|
assert not transaction.work_root.exists()
|
|
result = read_last_result(data_root)
|
|
assert result is not None and result["status"] == "failed"
|
|
saved = failure_log_path(data_root).read_text(encoding="utf-8")
|
|
assert "pytest failed" in saved
|
|
assert "simulated rollback fault" in saved
|
|
assert "final_recovery_state=" in saved
|