icrdr/AnyImage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add unit tests for service functionality and model caching

Browse Source 
- Implement tests for ben2 and da3 model caching to ensure models are reused for matching runtime keys.
- Verify service state management, including client registration, job cancellation, and client expiration.
- Test discovery file creation and removal based on service ownership.
- Ensure frame extraction functionality works without models and produces expected results.
This commit is contained in:
马南
2026-08-13 00:35:40 +08:00
parent 7d0983529e
commit 92d2cea1d1
21 changed files with 935 additions and 887 deletions
Download Patch File Download Diff File Expand all files Collapse all files

489
tests/test_addon.py
View File

File diff suppressed because it is too large Load Diff

303
tests/test_bootstrap_runtime.py → tests/test_install_env.py
View File

@@ -16,17 +16,17 @@ from unittest.mock import patch
PROJECT_ROOT = Path(__file__).resolve().parents[1]
def load_bootstrap_runtime():
def load_install_env():
specification = importlib.util.spec_from_file_location(
"snapform_bootstrap_runtime",
PROJECT_ROOT / "src" / "snapform" / "bootstrap_runtime.py",
"snapform_install_env",
PROJECT_ROOT / "src" / "snapform" / "install_env.py",
)
module = importlib.util.module_from_spec(specification)
specification.loader.exec_module(module)
return module
class BootstrapRuntimeTest(unittest.TestCase):
class InstallEnvTest(unittest.TestCase):
def create_bundled_vendor(self, root):
vendor = root / "vendor"
vendor.mkdir()
@@ -35,23 +35,19 @@ class BootstrapRuntimeTest(unittest.TestCase):
return vendor
def test_installs_da3_from_archives_without_gsplat(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
da3_uri = (
root / "vendor" / "depth-anything-3.tar.gz"
).as_uri()
ben2_uri = (root / "vendor" / "ben2.tar.gz").as_uri()
uv_path = root / "uv"
uv_path.write_bytes(b"uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -82,7 +78,13 @@ class BootstrapRuntimeTest(unittest.TestCase):
f"addict=={bootstrap.ADDICT_VERSION}",
da3_command,
)
self.assertIn(f"depth-anything-3 @ {da3_uri}", da3_command)
self.assertTrue(
any(
"depth-anything-3 @ file:" in str(value)
and "snapform_vendor_" in str(value)
for value in da3_command
)
)
self.assertFalse(
any("git+" in str(value) for value in da3_command)
)
@@ -105,20 +107,26 @@ class BootstrapRuntimeTest(unittest.TestCase):
for command in commands
if any("ben2 @" in str(value) for value in command)
)
self.assertIn(f"ben2 @ {ben2_uri}", ben2_command)
self.assertTrue(
any(
"ben2 @ file:" in str(value)
and "snapform_vendor_" in str(value)
for value in ben2_command
)
)
self.assertFalse(
any("git+" in str(value) for value in ben2_command)
)
host_command = next(
service_command = next(
command
for command in commands
if f"fastapi=={bootstrap.FASTAPI_VERSION}" in command
)
self.assertIn(f"uvicorn=={bootstrap.UVICORN_VERSION}", host_command)
self.assertIn("setuptools", host_command)
self.assertNotIn("ninja", host_command)
self.assertIn(f"uvicorn=={bootstrap.UVICORN_VERSION}", service_command)
self.assertIn("setuptools", service_command)
self.assertNotIn("ninja", service_command)
check_command = bootstrap.verify_command(
bootstrap.environment_python(runtime_dir)
bootstrap.venv_python(venv_dir)
)
self.assertTrue(
any("import torch;" in str(value) for value in check_command)
@@ -179,19 +187,19 @@ class BootstrapRuntimeTest(unittest.TestCase):
)
def test_darwin_install_skips_cuda_index_and_xformers(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
uv_path = root / "uv"
uv_path.write_bytes(b"uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -238,12 +246,12 @@ class BootstrapRuntimeTest(unittest.TestCase):
)
def test_intel_mac_is_rejected_with_clear_error(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
args = SimpleNamespace(
uv="uv",
runtime_dir="runtime",
venv_dir=".venv",
models_dir="models",
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status="status.json",
@@ -263,7 +271,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
mocked_rmtree.assert_not_called()
def test_run_stage_writes_stage_progress(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
status_path = Path(directory) / "status.json"
with patch.object(bootstrap, "run") as mocked_run:
@@ -288,7 +296,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertEqual(status["stage_label"], "pytorch")
def test_run_stage_writes_progress_range(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
status_path = Path(directory) / "status.json"
with patch.object(bootstrap, "run"), patch.object(
@@ -314,7 +322,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertEqual(progresses[-1], 0.4)
def test_run_stage_passes_progress_range_to_download_command(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
status_path = Path(directory) / "status.json"
command = [
@@ -351,19 +359,19 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertEqual(called_command[-1], "da3 small")
def test_install_allocates_contiguous_progress_slices(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
uv_path = root / "uv"
uv_path.write_bytes(b"uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -424,7 +432,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertEqual(status["progress"], 1.0)
def test_run_install_tracks_package_download_events(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
class FakeProcess:
def __init__(self, lines):
@@ -474,7 +482,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertEqual(installed, ["torch==2.9.1", "numpy==1.26.4"])
def test_run_install_smooths_progress_between_events(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
class SlowLines:
def __init__(self, lines, delay):
@@ -529,7 +537,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertIn("Downloading packages 3/3", messages)
def test_run_install_maps_progress_into_stage_range(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
class FakeProcess:
def __init__(self, lines):
@@ -567,7 +575,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertEqual(first.kwargs["stage_label"], "pytorch deps")
def test_run_install_retries_once_with_mirror(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
previous = bootstrap._resolved_pypi_index
bootstrap._resolved_pypi_index = None
try:
@@ -597,13 +605,13 @@ class BootstrapRuntimeTest(unittest.TestCase):
)
def test_formats_elapsed_time(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
self.assertEqual(bootstrap.format_elapsed(12), "12s")
self.assertEqual(bootstrap.format_elapsed(83), "1m 23s")
def test_ai_component_estimate_continues_across_internal_steps(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
first = bootstrap.estimated_remaining(
"AI components",
@@ -619,7 +627,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertLess(second, first)
def test_resolve_indexes_uses_configured_pypi_index(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
bootstrap.resolve_indexes(
SimpleNamespace(pypi_index="https://pypi.tuna.tsinghua.edu.cn/simple")
)
@@ -629,7 +637,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
)
def test_resolve_indexes_uses_environment_override(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with patch.dict(
os.environ,
{"PIP_INDEX_URL": "https://pypi.example/simple"},
@@ -641,7 +649,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
)
def test_child_environment_injects_resolved_pypi_index(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
previous_pypi = bootstrap._resolved_pypi_index
bootstrap._resolved_pypi_index = (
"https://pypi.tuna.tsinghua.edu.cn/simple"
@@ -657,19 +665,19 @@ class BootstrapRuntimeTest(unittest.TestCase):
)
def test_ready_json_records_installed_packages(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
uv_path = root / "uv"
uv_path.write_bytes(b"uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -689,7 +697,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
bootstrap.install(args)
ready = json.loads(
(runtime_dir / "environment_manifest.json").read_text(
(root / "manifest.json").read_text(
encoding="utf-8"
)
)
@@ -699,19 +707,19 @@ class BootstrapRuntimeTest(unittest.TestCase):
)
self.assertEqual(ready["devices"], ["cuda", "cpu"])
def test_broken_runtime_is_cleaned_before_reinstall(self):
bootstrap = load_bootstrap_runtime()
def test_broken_env_is_cleaned_before_reinstall(self):
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
(runtime_dir / "partial-python.exe").write_text("junk")
(venv_dir / "partial-python.exe").write_text("junk")
args = SimpleNamespace(
uv="uv",
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -729,36 +737,30 @@ class BootstrapRuntimeTest(unittest.TestCase):
):
bootstrap.install(args)
self.assertFalse((runtime_dir / "partial-python.exe").exists())
self.assertFalse((venv_dir / "partial-python.exe").exists())
self.assertTrue(
(runtime_dir / "environment_manifest.json").is_file()
(root / "manifest.json").is_file()
)
def test_ready_runtime_is_not_cleaned(self):
bootstrap = load_bootstrap_runtime()
def test_installed_env_is_not_cleaned(self):
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir(parents=True)
venv_dir = root / ".venv"
venv_dir.mkdir(parents=True)
self.create_bundled_vendor(root)
(runtime_dir / "Scripts").mkdir()
(runtime_dir / "Scripts" / "python.exe").write_text("python")
(runtime_dir / "environment_manifest.json").write_text(
json.dumps(
{
"environment_manifest_revision": (
bootstrap.ENVIRONMENT_MANIFEST_REVISION
)
}
),
(venv_dir / "Scripts").mkdir()
(venv_dir / "Scripts" / "python.exe").write_text("python")
(root / "manifest.json").write_text(
json.dumps({}),
encoding="utf-8",
)
(runtime_dir / "marker.txt").write_text("keep")
(venv_dir / "marker.txt").write_text("keep")
args = SimpleNamespace(
uv="uv",
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -776,10 +778,10 @@ class BootstrapRuntimeTest(unittest.TestCase):
):
bootstrap.install(args)
self.assertTrue((runtime_dir / "marker.txt").is_file())
self.assertTrue((venv_dir / "marker.txt").is_file())
def test_model_download_failure_keeps_runtime_ready(self):
bootstrap = load_bootstrap_runtime()
def test_model_download_failure_keeps_env_installed(self):
bootstrap = load_install_env()
commands = []
def fake_run(command, environment=None):
@@ -789,14 +791,14 @@ class BootstrapRuntimeTest(unittest.TestCase):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
args = SimpleNamespace(
uv="uv",
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -813,13 +815,13 @@ class BootstrapRuntimeTest(unittest.TestCase):
bootstrap.install(args)
self.assertTrue(
(runtime_dir / "environment_manifest.json").is_file()
(root / "manifest.json").is_file()
)
status = json.loads(
(root / "status.json").read_text(encoding="utf-8")
)
self.assertEqual(status["progress"], 1.0)
self.assertIn("Backend Environment is ready", status["message"])
self.assertIn("Environment is installed", status["message"])
self.assertTrue(
any(
"PramaLLC/BEN2" in map(str, command)
@@ -828,18 +830,18 @@ class BootstrapRuntimeTest(unittest.TestCase):
)
def test_install_fails_when_bundled_uv_missing(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
uv_path = root / "uv" / "uv.exe"
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -869,11 +871,11 @@ class BootstrapRuntimeTest(unittest.TestCase):
bootstrap.install(args)
def test_install_uses_bundled_vendor_and_uv_without_network(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
bundled = root / "bundled"
bundled.mkdir()
(bundled / "depth-anything-3.tar.gz").write_bytes(b"da3")
@@ -892,9 +894,9 @@ class BootstrapRuntimeTest(unittest.TestCase):
uv_path = root / "uv" / ("uv.exe" if os.name == "nt" else "uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -956,25 +958,35 @@ class BootstrapRuntimeTest(unittest.TestCase):
for command in install_commands
if any("depth-anything-3 @" in str(value) for value in command)
)
da3_uri = (root / "vendor" / "depth-anything-3.tar.gz").as_uri()
ben2_uri = (root / "vendor" / "ben2.tar.gz").as_uri()
self.assertIn(f"depth-anything-3 @ {da3_uri}", ai_command)
self.assertIn(f"ben2 @ {ben2_uri}", ai_command)
self.assertTrue(
any(
"depth-anything-3 @ file:" in str(value)
and "snapform_vendor_" in str(value)
for value in ai_command
)
)
self.assertTrue(
any(
"ben2 @ file:" in str(value)
and "snapform_vendor_" in str(value)
for value in ai_command
)
)
def test_install_reuses_existing_uv(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
uv_path = root / "uv.exe"
uv_path.write_bytes(b"uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -993,7 +1005,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
bootstrap.install(args)
def test_extract_uv_unpacks_standalone_binary(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
archive_path = root / "uv.zip"
@@ -1005,8 +1017,8 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertEqual(uv_path.read_bytes(), b"binary")
def test_sync_bundled_vendor_copies_missing_archives(self):
bootstrap = load_bootstrap_runtime()
def test_copy_vendor_inputs_copies_missing_archives(self):
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
bundled = root / "bundled"
@@ -1015,7 +1027,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
(bundled / "ben2.tar.gz").write_bytes(b"ben2")
vendor = root / "vendor"
bootstrap.sync_bundled_vendor(bundled, vendor)
bootstrap.copy_vendor_inputs(bundled, vendor)
self.assertEqual(
(vendor / "depth-anything-3.tar.gz").read_bytes(),
@@ -1026,8 +1038,8 @@ class BootstrapRuntimeTest(unittest.TestCase):
b"ben2",
)
def test_sync_bundled_vendor_replaces_same_size_stale_archive(self):
bootstrap = load_bootstrap_runtime()
def test_copy_vendor_inputs_replaces_same_size_stale_archive(self):
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
bundled = root / "bundled"
@@ -1038,19 +1050,38 @@ class BootstrapRuntimeTest(unittest.TestCase):
vendor.mkdir()
(vendor / "depth-anything-3.tar.gz").write_bytes(b"old")
bootstrap.sync_bundled_vendor(bundled, vendor)
bootstrap.copy_vendor_inputs(bundled, vendor)
self.assertEqual(
(vendor / "depth-anything-3.tar.gz").read_bytes(),
b"new",
)
def test_install_stages_uses_local_source_archives(self):
bootstrap = load_bootstrap_runtime()
def test_remove_legacy_vendor_removes_only_install_inputs(self):
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
vendor = root / "vendor"
vendor.mkdir()
(vendor / "depth-anything-3.tar.gz").write_bytes(b"da3")
(root / "depth-anything-3.tar.gz").write_bytes(b"da3")
(root / "ben2.tar.gz").write_bytes(b"ben2")
models = root / "models"
models.mkdir()
bootstrap.remove_legacy_vendor(root)
self.assertFalse(vendor.exists())
self.assertFalse((root / "depth-anything-3.tar.gz").exists())
self.assertFalse((root / "ben2.tar.gz").exists())
self.assertTrue(models.is_dir())
def test_install_stages_uses_local_source_archives(self):
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
venv_dir = root / ".venv"
venv_dir.mkdir()
vendor = root / "vendor"
vendor.mkdir()
(vendor / "depth-anything-3.tar.gz").write_bytes(b"da3")
@@ -1059,9 +1090,9 @@ class BootstrapRuntimeTest(unittest.TestCase):
uv_path.write_bytes(b"uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -1069,7 +1100,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
stages = bootstrap.install_stages(
args,
runtime_dir,
venv_dir,
root / "python",
uv_path,
)
@@ -1086,11 +1117,11 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertIn(f"ben2 @ {ben2_local}", ai_command)
def test_install_stages_download_only_default_models(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
vendor = root / "vendor"
vendor.mkdir()
(vendor / "depth-anything-3.tar.gz").write_bytes(b"da3")
@@ -1099,9 +1130,9 @@ class BootstrapRuntimeTest(unittest.TestCase):
uv_path.write_bytes(b"uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -1109,7 +1140,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
stages = bootstrap.install_stages(
args,
runtime_dir,
venv_dir,
root / "python",
uv_path,
)
@@ -1123,20 +1154,20 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertIn("ben2_base", " ".join(map(str, downloads["BEN2 Base"])))
self.assertNotIn("DA3 Base", downloads)
def test_install_does_not_verify_runtime(self):
bootstrap = load_bootstrap_runtime()
def test_install_does_not_verify_env(self):
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
runtime_dir = root / "runtime"
runtime_dir.mkdir()
venv_dir = root / ".venv"
venv_dir.mkdir()
self.create_bundled_vendor(root)
uv_path = root / "uv"
uv_path.write_bytes(b"uv")
args = SimpleNamespace(
uv=str(uv_path),
runtime_dir=str(runtime_dir),
venv_dir=str(venv_dir),
models_dir=str(root / "models"),
backend="cu128",
torch_variant="cu128",
da3_commit="test-commit",
ben2_commit="test-ben2-commit",
status=str(root / "status.json"),
@@ -1157,7 +1188,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
mocked_verify.assert_not_called()
def test_run_verify_returns_detected_devices(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
class FakeProcess:
def __init__(self, lines):
@@ -1184,7 +1215,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertTrue(report["mps"])
def test_run_verify_writes_heartbeat_progress(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
class FakeProcess:
def __init__(self, lines):
@@ -1219,14 +1250,14 @@ class BootstrapRuntimeTest(unittest.TestCase):
messages = [call.args[2] for call in mocked_write.call_args_list]
self.assertTrue(
any(
"Checking Backend Environment" in message
"Checking Environment" in message
for message in messages
)
)
self.assertEqual(report, {})
def test_install_uv_extracts_bundled_archive(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
uv_path = root / "uv"
@@ -1253,7 +1284,7 @@ class BootstrapRuntimeTest(unittest.TestCase):
self.assertEqual(uv_path.read_bytes(), b"uv-binary")
def test_install_uv_raises_when_bundled_missing(self):
bootstrap = load_bootstrap_runtime()
bootstrap = load_install_env()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
uv_path = root / "uv"

2
tests/test_manifest.py
View File

@@ -18,7 +18,7 @@ class ManifestTest(unittest.TestCase):
self.assertEqual(self.manifest["type"], "add-on")
self.assertNotIn("bl_info", self.manifest)
def test_declares_required_runtime_permissions(self):
def test_declares_required_environment_permissions(self):
self.assertEqual(
set(self.manifest["permissions"]),
{"network", "files"},

2
tests/test_pack.py
View File

@@ -61,7 +61,7 @@ class PackTest(unittest.TestCase):
names,
)
self.assertIn("worker.py", names)
self.assertIn("ai_host.py", names)
self.assertIn("service.py", names)
self.assertNotIn("snapform/__init__.py", names)
self.assertNotIn("importer.py", names)
self.assertIn("vendor/depth-anything-3.tar.gz", names)

54
tests/test_ai_host.py → tests/test_service.py
View File

@@ -10,10 +10,10 @@ from unittest.mock import patch
PROJECT_ROOT = Path(__file__).resolve().parents[1]
def load_ai_host():
def load_service():
specification = importlib.util.spec_from_file_location(
"snapform_ai_host",
PROJECT_ROOT / "src" / "snapform" / "ai_host.py",
"snapform_service",
PROJECT_ROOT / "src" / "snapform" / "service.py",
)
module = importlib.util.module_from_spec(specification)
specification.loader.exec_module(module)
@@ -33,13 +33,13 @@ class FakeCuda:
return None
class HostTest(unittest.TestCase):
class ServiceTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.host = load_ai_host()
cls.service = load_service()
def create_cache(self):
cache = self.host.ModelCache.__new__(self.host.ModelCache)
cache = self.service.ModelCache.__new__(self.service.ModelCache)
cache.torch = SimpleNamespace(cuda=FakeCuda())
cache.ben2_model = None
cache.ben2_key = None
@@ -111,13 +111,13 @@ class HostTest(unittest.TestCase):
self.assertEqual(loads, ["model"])
self.assertEqual(cache.status()["da3"], "model")
def test_host_tracks_clients_and_rejects_shared_shutdown(self):
def test_service_tracks_clients_and_rejects_shared_shutdown(self):
fake_models = SimpleNamespace(
status=lambda: {"ben2": "", "da3": ""},
close=lambda: None,
)
with patch.object(self.host, "ModelCache", return_value=fake_models):
state = self.host.HostState("token", "instance")
with patch.object(self.service, "ModelCache", return_value=fake_models):
state = self.service.ServiceState("token", "instance")
state.register_client("first", "Blender A")
state.register_client("second", "Blender B")
@@ -131,13 +131,13 @@ class HostTest(unittest.TestCase):
state.reserve("other", "second", "reconstruct", "other.json")
)
def test_cancelling_job_keeps_host_and_model_cache_alive(self):
def test_cancelling_job_keeps_service_and_model_cache_alive(self):
fake_models = SimpleNamespace(
status=lambda: {"ben2": "BEN2_BASE", "da3": ""},
close=lambda: None,
)
with patch.object(self.host, "ModelCache", return_value=fake_models):
state = self.host.HostState("token", "instance")
with patch.object(self.service, "ModelCache", return_value=fake_models):
state = self.service.ServiceState("token", "instance")
state.register_client("client", "Blender")
with tempfile.TemporaryDirectory() as directory:
status_path = Path(directory) / "status.json"
@@ -158,7 +158,7 @@ class HostTest(unittest.TestCase):
"parameters": {"status": str(status_path)},
}
)
status = self.host.json.loads(
status = self.service.json.loads(
status_path.read_text(encoding="utf-8")
)
@@ -171,44 +171,44 @@ class HostTest(unittest.TestCase):
status=lambda: {"ben2": "", "da3": ""},
close=lambda: None,
)
with patch.object(self.host, "ModelCache", return_value=fake_models):
state = self.host.HostState("token", "instance")
with patch.object(self.service, "ModelCache", return_value=fake_models):
state = self.service.ServiceState("token", "instance")
state.register_client("client", "Blender")
state.clients["client"]["last_seen"] -= (
self.host.CLIENT_TIMEOUT_SECONDS + 1.0
self.service.CLIENT_TIMEOUT_SECONDS + 1.0
)
state.expire_clients()
self.assertEqual(state.snapshot()["clients"], 0)
def test_discovery_is_only_removed_by_matching_host(self):
def test_discovery_is_only_removed_by_matching_service(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "host.json"
path = Path(directory) / "service.json"
args = SimpleNamespace(
discovery=str(path),
port=51234,
token="token",
instance_id="owner",
)
with patch.object(self.host.os, "getpid", return_value=42):
self.host.write_discovery(args)
with patch.object(self.service.os, "getpid", return_value=42):
self.service.write_discovery(args)
payload = self.host.json.loads(path.read_text(encoding="utf-8"))
payload = self.service.json.loads(path.read_text(encoding="utf-8"))
self.assertEqual(payload["port"], 51234)
self.assertEqual(payload["pid"], 42)
self.host.remove_discovery(path, "other")
self.service.remove_discovery(path, "other")
self.assertTrue(path.is_file())
self.host.remove_discovery(path, "owner")
self.service.remove_discovery(path, "owner")
self.assertFalse(path.exists())
def test_host_dispatches_frame_extraction_without_models(self):
def test_service_dispatches_frame_extraction_without_models(self):
fake_models = SimpleNamespace(
status=lambda: {"ben2": "", "da3": ""},
close=lambda: None,
)
with patch.object(self.host, "ModelCache", return_value=fake_models):
state = self.host.HostState("token", "instance")
with patch.object(self.service, "ModelCache", return_value=fake_models):
state = self.service.ServiceState("token", "instance")
state.register_client("client", "Blender")
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
@@ -238,7 +238,7 @@ class HostTest(unittest.TestCase):
},
}
)
status = self.host.json.loads(
status = self.service.json.loads(
status_path.read_text(encoding="utf-8")
)
self.assertTrue(