432 lines
17 KiB
Python
432 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
FNOS_ROOT = PROJECT_ROOT / "fnos"
|
|
|
|
|
|
def _mock_app(tmp_path: Path) -> tuple[Path, Path, Path]:
|
|
app_root = tmp_path / "app"
|
|
runtime = app_root / "runtime"
|
|
(runtime / "wheels").mkdir(parents=True)
|
|
(runtime / "VERSION").write_text("0.4.30001\n0.4.3\n" + "a" * 64 + "\n")
|
|
(runtime / "runtime-core.txt").write_text("# mocked offline core\n")
|
|
(runtime / "imagefind-0.4.3-py3-none-any.whl").write_bytes(b"mock wheel")
|
|
(runtime / "wheels" / "dependency.whl").write_bytes(b"mock dependency")
|
|
python = app_root / "fake-python312"
|
|
python.write_text(
|
|
"""#!/bin/bash
|
|
set -eu
|
|
if [ "${1:-}" = "-c" ]; then
|
|
case "${2:-}" in
|
|
*imagefind.__version__*) printf '%s\n' '0.4.3'; exit 0 ;;
|
|
esac
|
|
if [ "$#" -gt 2 ]; then
|
|
exec "$MOCK_REAL_PYTHON" "$@"
|
|
fi
|
|
printf '%s\n' '3.12'
|
|
exit 0
|
|
fi
|
|
if [ "${1:-}" = "-m" ] && [ "${2:-}" = "venv" ]; then
|
|
mkdir -p "$3/bin"
|
|
cp "$0" "$3/bin/python"
|
|
chmod 0755 "$3/bin/python"
|
|
exit 0
|
|
fi
|
|
if [ "${1:-}" = "-m" ] && [ "${2:-}" = "pip" ]; then
|
|
exit 0
|
|
fi
|
|
if [ "${1:-}" = "-m" ] && [ "${2:-}" = "imagefind.main" ]; then
|
|
shift 2
|
|
fi
|
|
if [ "${1:-}" = "--version" ]; then
|
|
printf '%s\n' '0.4.3'
|
|
exit 0
|
|
fi
|
|
if [ "${1:-}" = "postgres-enroll" ]; then
|
|
rm -f "$TRIM_PKGVAR/postgres-enrollment-token.seed"
|
|
exit 0
|
|
fi
|
|
if [ "${1:-}" = "admin-password" ]; then
|
|
printf '%s' "$*" >"$MOCK_ARGS_FILE"
|
|
[ -z "${wizard_admin_password:-}" ]
|
|
[ -z "${wizard_admin_password_confirm:-}" ]
|
|
PASSWORD=''
|
|
IFS= read -r PASSWORD || true
|
|
printf '%s' "$PASSWORD" >"$MOCK_PASSWORD_FILE"
|
|
fi
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
python.chmod(0o755)
|
|
return app_root, tmp_path / "mock-args", tmp_path / "mock-password"
|
|
|
|
|
|
def _environment(
|
|
tmp_path: Path,
|
|
app_root: Path,
|
|
args_file: Path,
|
|
password_file: Path,
|
|
) -> dict[str, str]:
|
|
return {
|
|
**os.environ,
|
|
"TRIM_APPDEST": str(app_root),
|
|
"TRIM_PKGVAR": str(tmp_path / "var"),
|
|
"TRIM_TEMP_LOGFILE": str(tmp_path / "fnos-error.log"),
|
|
"MOCK_ARGS_FILE": str(args_file),
|
|
"MOCK_PASSWORD_FILE": str(password_file),
|
|
"MOCK_REAL_PYTHON": sys.executable,
|
|
"IMAGEFIND_PYTHON_PATH": str(app_root / "fake-python312"),
|
|
}
|
|
|
|
|
|
def _run(script: str, env: dict[str, str]) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["bash", str(FNOS_ROOT / "cmd" / script)],
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def test_fnos_wizard_contracts():
|
|
install = json.loads((FNOS_ROOT / "wizard" / "install").read_text())
|
|
upgrade = json.loads((FNOS_ROOT / "wizard" / "upgrade").read_text())
|
|
uninstall = json.loads((FNOS_ROOT / "wizard" / "uninstall").read_text())
|
|
config = json.loads((FNOS_ROOT / "wizard" / "config").read_text())
|
|
|
|
assert [item["type"] for item in install[0]["items"]] == ["tips", "password", "password", "password"]
|
|
assert install[0]["items"][1]["field"] == "wizard_postgres_enrollment_token"
|
|
assert [item["type"] for item in upgrade[0]["items"]] == ["tips"]
|
|
clear_data = uninstall[0]["items"][1]
|
|
assert clear_data["field"] == "wizard_clear_data"
|
|
# fnpack 1.2.3 requires switch defaults to be encoded as strings.
|
|
assert clear_data["initValue"] == "false"
|
|
assert [item["type"] for item in config[0]["items"]] == ["tips", "switch", "text"]
|
|
assert config[0]["items"][1]["initValue"] == "false"
|
|
|
|
desktop = json.loads((FNOS_ROOT / "app" / "ui" / "config").read_text())[".url"]["imagefind.Application"]
|
|
assert desktop["gatewayPrefix"] == "/app/imagefind"
|
|
assert desktop["gatewaySocket"] == "imagefind.sock"
|
|
assert desktop["url"] == "/app/imagefind/"
|
|
manifest = (FNOS_ROOT / "manifest").read_text()
|
|
assert "checkport=false" in manifest
|
|
assert "service_port=" not in manifest
|
|
manifest_version = re.search(r"^version\s*=\s*([^\s]+)$", manifest, re.MULTILINE).group(1)
|
|
project_version = tomllib.loads((PROJECT_ROOT / "pyproject.toml").read_text())["project"]["version"]
|
|
frontend_version = json.loads((PROJECT_ROOT / "frontend" / "package.json").read_text())["version"]
|
|
backend_version = re.search(
|
|
r'^__version__\s*=\s*"([^"]+)"$',
|
|
(PROJECT_ROOT / "backend" / "imagefind" / "__init__.py").read_text(),
|
|
re.MULTILINE,
|
|
).group(1)
|
|
assert {manifest_version, project_version, frontend_version, backend_version} == {"0.5.45"}
|
|
assert "install_dep_apps=python312,nxsir.postgresql" in manifest
|
|
assert os.access(FNOS_ROOT / "cmd" / "config_callback", os.X_OK)
|
|
assert os.access(FNOS_ROOT / "cmd" / "runtime_bootstrap", os.X_OK)
|
|
build_script = (PROJECT_ROOT / "scripts" / "build-fnos.sh").read_text()
|
|
assert "PyInstaller" not in build_script
|
|
assert "requirements/runtime-core.txt" in build_script
|
|
assert ".fnos-wheel-cache/python312" in build_script
|
|
assert "vendor/qdrant" not in build_script
|
|
assert "vendor/ffmpeg" not in build_script
|
|
assert f"imagefind-{manifest_version}-x86_64.fpk" in build_script
|
|
|
|
|
|
def test_generated_fnos_icons_fill_the_opaque_canvas(tmp_path: Path):
|
|
stage = tmp_path / "stage"
|
|
subprocess.run(
|
|
[sys.executable, str(PROJECT_ROOT / "scripts" / "make_icons.py"), str(stage)],
|
|
check=True,
|
|
)
|
|
expected = {
|
|
stage / "ICON.PNG": 128,
|
|
stage / "ICON_256.PNG": 256,
|
|
stage / "app" / "ui" / "images" / "icon_64.png": 64,
|
|
stage / "app" / "ui" / "images" / "icon_256.png": 256,
|
|
}
|
|
for path, size in expected.items():
|
|
with Image.open(path) as icon:
|
|
assert icon.mode == "RGBA"
|
|
assert icon.size == (size, size)
|
|
assert icon.getpixel((0, 0)) == (103, 145, 244, 255)
|
|
assert icon.getpixel((size - 1, size - 1)) == (103, 145, 244, 255)
|
|
assert icon.getpixel((size // 2, size // 2))[3] == 255
|
|
|
|
|
|
def test_install_initializes_password_without_logging_it(tmp_path: Path):
|
|
app_root, args_file, password_file = _mock_app(tmp_path)
|
|
env = _environment(tmp_path, app_root, args_file, password_file)
|
|
password = "administrator passphrase"
|
|
postgres_token = "test-postgresql-enrollment-token"
|
|
env.update(
|
|
wizard_postgres_enrollment_token=postgres_token,
|
|
wizard_admin_password=password,
|
|
wizard_admin_password_confirm=password,
|
|
)
|
|
|
|
result = _run("install_callback", env)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
assert args_file.read_text() == "admin-password --stdin --replace"
|
|
assert password_file.read_text() == password
|
|
assert (tmp_path / "var" / ".imagefind-data-root").read_text().strip() == "imagefind-data-v1"
|
|
assert json.loads((tmp_path / "var" / "access.json").read_text())["direct_access"] == {
|
|
"enabled": False,
|
|
"port": 8765,
|
|
}
|
|
assert password not in result.stdout
|
|
assert password not in result.stderr
|
|
assert password not in (tmp_path / "fnos-error.log").read_text()
|
|
assert postgres_token not in result.stdout
|
|
assert postgres_token not in result.stderr
|
|
assert postgres_token not in (tmp_path / "fnos-error.log").read_text()
|
|
assert not (tmp_path / "var" / "postgres-enrollment-token.seed").exists()
|
|
|
|
|
|
def test_install_rejects_mismatched_passwords_without_invoking_server(tmp_path: Path):
|
|
app_root, args_file, password_file = _mock_app(tmp_path)
|
|
env = _environment(tmp_path, app_root, args_file, password_file)
|
|
env.update(
|
|
wizard_postgres_enrollment_token="test-postgresql-enrollment-token",
|
|
wizard_admin_password="administrator passphrase",
|
|
wizard_admin_password_confirm="different administrator passphrase",
|
|
)
|
|
|
|
result = _run("install_callback", env)
|
|
|
|
assert result.returncode != 0
|
|
assert "不一致" in result.stderr
|
|
assert not args_file.exists()
|
|
assert not password_file.exists()
|
|
error_log = (tmp_path / "fnos-error.log").read_text()
|
|
assert "不一致" in error_log
|
|
assert env["wizard_admin_password"] not in error_log
|
|
assert env["wizard_admin_password_confirm"] not in error_log
|
|
|
|
|
|
def test_upgrade_is_idempotent_and_never_changes_password(tmp_path: Path):
|
|
app_root, args_file, password_file = _mock_app(tmp_path)
|
|
env = _environment(tmp_path, app_root, args_file, password_file)
|
|
data_root = tmp_path / "var"
|
|
(data_root / "data").mkdir(parents=True)
|
|
database = data_root / "data" / "imagefind.sqlite3"
|
|
database.write_text("existing database")
|
|
old_runtime = data_root / "runtime" / "ai-current"
|
|
old_runtime.mkdir(parents=True)
|
|
(old_runtime / "sentinel").write_text("preserve old runtime data")
|
|
(data_root / ".imagefind-data-root").write_text("imagefind-data-v1\n")
|
|
(data_root / "access.json").write_text('{"direct_access":{"enabled":false,"port":9876}}\n')
|
|
# Cached fnOS wizard values from an older package must not turn a file
|
|
# replacement into a failed update or reset the administrator password.
|
|
env.update(wizard_admin_password="placeholder", wizard_admin_password_confirm="different")
|
|
|
|
preserve = _run("upgrade_callback", env)
|
|
again = _run("upgrade_callback", env)
|
|
assert preserve.returncode == 0, preserve.stderr
|
|
assert again.returncode == 0, again.stderr
|
|
assert not args_file.exists()
|
|
assert not password_file.exists()
|
|
assert database.read_text() == "existing database"
|
|
assert (old_runtime / "sentinel").read_text() == "preserve old runtime data"
|
|
assert json.loads((data_root / "access.json").read_text())["direct_access"] == {
|
|
"enabled": False,
|
|
"port": 9876,
|
|
}
|
|
assert "升级数据检查完成" in (tmp_path / "fnos-error.log").read_text()
|
|
|
|
|
|
def test_upgrade_init_does_not_migrate_or_modify_legacy_sqlite_data(tmp_path: Path):
|
|
app_root = tmp_path / "app"
|
|
(app_root / "bin").mkdir(parents=True)
|
|
qdrant = app_root / "bin" / "qdrant"
|
|
qdrant.write_bytes(b"legacy-qdrant-binary")
|
|
qdrant.chmod(0o755)
|
|
(app_root / "qdrant.yaml").write_text("storage:\n storage_path: ./storage\n")
|
|
data_root = tmp_path / "var"
|
|
database = data_root / "data" / "imagefind.sqlite3"
|
|
database.parent.mkdir(parents=True)
|
|
database.write_bytes(b"legacy sqlite data remains outside the PostgreSQL release")
|
|
env = {
|
|
**os.environ,
|
|
"TRIM_APPDEST": str(app_root),
|
|
"TRIM_PKGVAR": str(data_root),
|
|
"TRIM_TEMP_LOGFILE": str(tmp_path / "fnos-upgrade.log"),
|
|
"IMAGEFIND_PYTHON_PATH": sys.executable,
|
|
}
|
|
|
|
result = _run("upgrade_init", env)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
assert database.read_bytes() == b"legacy sqlite data remains outside the PostgreSQL release"
|
|
assert qdrant.read_bytes() == b"legacy-qdrant-binary"
|
|
assert not (data_root / "migration").exists()
|
|
assert "不执行旧数据迁移" in (tmp_path / "fnos-upgrade.log").read_text()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("current_kind", "old_name"),
|
|
[
|
|
("symlink", "core-0.3.14"),
|
|
("symlink", "core-0.3.16"),
|
|
("dangling", "core-missing"),
|
|
("directory", "current"),
|
|
],
|
|
)
|
|
def test_runtime_bootstrap_atomically_replaces_existing_current(
|
|
tmp_path: Path,
|
|
current_kind: str,
|
|
old_name: str,
|
|
):
|
|
app_root, args_file, password_file = _mock_app(tmp_path)
|
|
env = _environment(tmp_path, app_root, args_file, password_file)
|
|
runtime_root = tmp_path / "var" / "runtime"
|
|
runtime_root.mkdir(parents=True)
|
|
current = runtime_root / "current"
|
|
|
|
if current_kind == "symlink":
|
|
old_runtime = runtime_root / old_name
|
|
old_runtime.mkdir()
|
|
(old_runtime / "sentinel").write_text("preserve old runtime")
|
|
current.symlink_to(old_name)
|
|
elif current_kind == "dangling":
|
|
old_runtime = runtime_root / old_name
|
|
current.symlink_to(old_name)
|
|
else:
|
|
old_runtime = current
|
|
old_runtime.mkdir()
|
|
(old_runtime / "sentinel").write_text("preserve old runtime")
|
|
|
|
result = _run("runtime_bootstrap", env)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
assert current.is_symlink()
|
|
assert current.resolve() == runtime_root / "core-0.4.3-aaaaaaaaaaaaaaaa"
|
|
version = subprocess.run(
|
|
[str(current / "bin" / "python"), "-m", "imagefind.main", "--version"],
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
assert version.stdout.strip() == "0.4.3"
|
|
if current_kind == "symlink":
|
|
assert (old_runtime / "sentinel").read_text() == "preserve old runtime"
|
|
assert not list(old_runtime.glob(".current.*"))
|
|
elif current_kind == "dangling":
|
|
assert not old_runtime.exists()
|
|
else:
|
|
previous = list(runtime_root.glob("current.previous.*"))
|
|
assert len(previous) == 1
|
|
assert (previous[0] / "sentinel").read_text() == "preserve old runtime"
|
|
|
|
|
|
def test_runtime_bootstrap_reuses_content_id_across_package_builds_and_prunes_old_runtimes(tmp_path: Path):
|
|
app_root, args_file, password_file = _mock_app(tmp_path)
|
|
env = _environment(tmp_path, app_root, args_file, password_file)
|
|
runtime_root = tmp_path / "var" / "runtime"
|
|
runtime_root.mkdir(parents=True)
|
|
for name in ("core-0.3.14-old", "core-0.3.16-newer"):
|
|
old_runtime = runtime_root / name
|
|
old_runtime.mkdir()
|
|
(old_runtime / "sentinel").write_text(name)
|
|
|
|
first = _run("runtime_bootstrap", env)
|
|
target = runtime_root / "core-0.4.3-aaaaaaaaaaaaaaaa"
|
|
assert first.returncode == 0, first.stderr
|
|
assert target.is_dir()
|
|
assert len([path for path in runtime_root.glob("core-*") if path.is_dir()]) == 2
|
|
|
|
(app_root / "runtime" / "VERSION").write_text("0.4.30002\n0.4.3\n" + "a" * 64 + "\n")
|
|
second = _run("runtime_bootstrap", env)
|
|
assert second.returncode == 0, second.stderr
|
|
assert (runtime_root / "current").resolve() == target
|
|
assert len([path for path in runtime_root.glob("core-*") if path.is_dir()]) == 2
|
|
|
|
|
|
def test_config_callback_writes_validated_direct_access_atomically(tmp_path: Path):
|
|
data_root = tmp_path / "var"
|
|
env = {
|
|
**os.environ,
|
|
"TRIM_PKGVAR": str(data_root),
|
|
"TRIM_TEMP_LOGFILE": str(tmp_path / "fnos-error.log"),
|
|
"wizard_direct_access": "true",
|
|
"wizard_direct_port": "9876",
|
|
}
|
|
configured = _run("config_callback", env)
|
|
assert configured.returncode == 0, configured.stderr
|
|
assert json.loads((data_root / "access.json").read_text()) == {
|
|
"direct_access": {"enabled": True, "port": 9876}
|
|
}
|
|
assert (data_root / "access.json").stat().st_mode & 0o777 == 0o600
|
|
assert not list(data_root.glob("access.json.tmp.*"))
|
|
|
|
env["wizard_direct_port"] = "70000"
|
|
invalid = _run("config_callback", env)
|
|
assert invalid.returncode != 0
|
|
assert json.loads((data_root / "access.json").read_text())["direct_access"]["port"] == 9876
|
|
|
|
|
|
def test_uninstall_preserves_by_default_and_only_clears_internal_data(tmp_path: Path):
|
|
data_root = tmp_path / "var"
|
|
data_root.mkdir()
|
|
(data_root / ".imagefind-data-root").write_text("imagefind-data-v1\n")
|
|
(data_root / "data").mkdir()
|
|
(data_root / "data" / "imagefind.sqlite3").write_text("database")
|
|
(data_root / "models").mkdir()
|
|
(data_root / "models" / "model.bin").write_text("model")
|
|
shared_root = tmp_path / "shared" / "imagefind" / "videos"
|
|
shared_root.mkdir(parents=True)
|
|
source_video = shared_root / "source.mp4"
|
|
source_video.write_bytes(b"source-video")
|
|
app_root = tmp_path / "app"
|
|
env = {
|
|
**os.environ,
|
|
"TRIM_APPDEST": str(app_root),
|
|
"TRIM_PKGVAR": str(data_root),
|
|
"TRIM_TEMP_LOGFILE": str(tmp_path / "fnos-error.log"),
|
|
"TRIM_DATA_SHARE_PATHS": str(shared_root),
|
|
"TRIM_DATA_ACCESSIBLE_PATHS": str(tmp_path / "authorized-videos"),
|
|
}
|
|
|
|
preserve = _run("uninstall_init", env)
|
|
assert preserve.returncode == 0, preserve.stderr
|
|
assert (data_root / "data" / "imagefind.sqlite3").exists()
|
|
assert source_video.read_bytes() == b"source-video"
|
|
|
|
env["wizard_clear_data"] = "true"
|
|
clear = _run("uninstall_init", env)
|
|
assert clear.returncode == 0, clear.stderr
|
|
assert list(data_root.iterdir()) == []
|
|
assert source_video.read_bytes() == b"source-video"
|
|
|
|
|
|
def test_uninstall_refuses_data_directory_without_safety_marker(tmp_path: Path):
|
|
data_root = tmp_path / "not-imagefind-data"
|
|
data_root.mkdir()
|
|
protected = data_root / "keep.txt"
|
|
protected.write_text("keep")
|
|
env = {
|
|
**os.environ,
|
|
"TRIM_PKGVAR": str(data_root),
|
|
"TRIM_TEMP_LOGFILE": str(tmp_path / "fnos-error.log"),
|
|
"wizard_clear_data": "true",
|
|
}
|
|
|
|
result = _run("uninstall_init", env)
|
|
|
|
assert result.returncode != 0
|
|
assert "安全标记" in result.stderr
|
|
assert protected.read_text() == "keep"
|