389 lines
16 KiB
Python
389 lines
16 KiB
Python
|
"""SDK 自测：清单校验 / 打包往返 / 装载策略 / 本地测试台（无需 worker）。
|
|||
|
|
|
|||
|
|
有设备的用例标 `@pytest.mark.device`（无 sophon.sail 时自动跳过）：
|
|||
|
|
|
|||
|
|
uv run pytest # 静态用例
|
|||
|
|
uv run pytest -m device # 设备上跑推理
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import subprocess
|
|||
|
|
import sys
|
|||
|
|
import tempfile
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
|
|||
|
|
from algpkg_sdk import (AlgoBuilder, AlgoError, AlgoPackage, AlgoPackageError, Manifest,
|
|||
|
|
ManifestError, install_sdk_alias, sha256_file)
|
|||
|
|
from algpkg_sdk import testkit
|
|||
|
|
from algpkg_sdk.cli import main as cli_main
|
|||
|
|
|
|||
|
|
EXAMPLES = Path(__file__).resolve().parents[1] / "examples"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _manifest(**overrides) -> dict:
|
|||
|
|
payload = {
|
|||
|
|
"manifest_format": 1, "id": "t.demo", "name": "测试包", "version": "1.0.0",
|
|||
|
|
"api_version": 1, "kind": "detector", "entry": "Algorithm",
|
|||
|
|
"code_package": "alg_t_demo",
|
|||
|
|
"models": [{"name": "default", "file": "models/t.bmodel", "precision": "fp16",
|
|||
|
|
"resolution": [640, 640], "batch": 1, "default": True}],
|
|||
|
|
"params_schema": [{"key": "conf", "type": "float", "default": 0.25,
|
|||
|
|
"min": 0.01, "max": 0.99, "label": "置信度"}],
|
|||
|
|
"defaults": {"conf": 0.25},
|
|||
|
|
}
|
|||
|
|
payload.update(overrides)
|
|||
|
|
return payload
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------- 清单
|
|||
|
|
|
|||
|
|
def test_manifest_required_fields() -> None:
|
|||
|
|
Manifest.parse(_manifest())
|
|||
|
|
for missing in ("id", "version", "models"):
|
|||
|
|
bad = _manifest()
|
|||
|
|
bad.pop(missing)
|
|||
|
|
with pytest.raises(ManifestError):
|
|||
|
|
Manifest.parse(bad)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_manifest_rejects_bad_values() -> None:
|
|||
|
|
with pytest.raises(ManifestError): # id 形态
|
|||
|
|
Manifest.parse(_manifest(id="Bad Id!"))
|
|||
|
|
with pytest.raises(ManifestError): # 模型路径越界
|
|||
|
|
Manifest.parse(_manifest(models=[{"name": "d", "file": "../x.bmodel"}]))
|
|||
|
|
with pytest.raises(ManifestError): # api_version 过高
|
|||
|
|
Manifest.parse(_manifest(api_version=99))
|
|||
|
|
with pytest.raises(ManifestError): # params_schema 重复键
|
|||
|
|
Manifest.parse(_manifest(params_schema=[
|
|||
|
|
{"key": "conf", "type": "float"}, {"key": "conf", "type": "float"}]))
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_manifest_default_model_pick() -> None:
|
|||
|
|
manifest = Manifest.parse(_manifest(models=[
|
|||
|
|
{"name": "fp16", "file": "models/a.bmodel"},
|
|||
|
|
{"name": "int8", "file": "models/b.bmodel", "default": True}]))
|
|||
|
|
assert manifest.default_model().name == "int8"
|
|||
|
|
assert manifest.model("fp16").file == "models/a.bmodel"
|
|||
|
|
with pytest.raises(ManifestError):
|
|||
|
|
manifest.model("nope")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_manifest_check_params_semantics() -> None:
|
|||
|
|
manifest = Manifest.parse(_manifest())
|
|||
|
|
merged = manifest.check_params({"conf": 0.5})
|
|||
|
|
assert merged["conf"] == 0.5
|
|||
|
|
assert manifest.check_params({})["conf"] == 0.25 # 补默认
|
|||
|
|
with pytest.raises(ManifestError): # 越界拒绝（不静默截断）
|
|||
|
|
manifest.check_params({"conf": 2.0})
|
|||
|
|
with pytest.raises(ManifestError): # 类型不符
|
|||
|
|
manifest.check_params({"conf": "abc"})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_manifest_legacy_and_entry_helpers() -> None:
|
|||
|
|
manifest = Manifest.parse(_manifest(entry="sub:Alg", legacy={"algorithm": "yolo"}))
|
|||
|
|
assert manifest.entry_module == "alg_t_demo.sub"
|
|||
|
|
assert manifest.entry_class == "Alg"
|
|||
|
|
assert manifest.legacy["algorithm"] == "yolo"
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------- 打包往返
|
|||
|
|
|
|||
|
|
def _make_source(root: Path, *, model_bytes: bytes = b"FAKE-BMODEL" * 512) -> Path:
|
|||
|
|
src = root / "t.demo"
|
|||
|
|
(src / "alg_t_demo").mkdir(parents=True)
|
|||
|
|
(src / "alg_t_demo" / "__init__.py").write_text(
|
|||
|
|
"from video_algpkg_api import Algorithm as Base, AlgoResult\n"
|
|||
|
|
"class Algorithm(Base):\n"
|
|||
|
|
" def load(self):\n"
|
|||
|
|
" self._data = self.ctx.model_bytes()\n"
|
|||
|
|
" self.ctx.logger.info('loaded %s bytes', len(self._data))\n"
|
|||
|
|
" def infer(self, frame):\n"
|
|||
|
|
" return AlgoResult(boxes=None, value=len(self._data))\n",
|
|||
|
|
encoding="utf-8",
|
|||
|
|
)
|
|||
|
|
(src / "models").mkdir()
|
|||
|
|
(src / "models" / "t.bmodel").write_bytes(model_bytes)
|
|||
|
|
(src / "assets").mkdir()
|
|||
|
|
(src / "assets" / "labels.txt").write_text("person\ncar\n", encoding="utf-8")
|
|||
|
|
(src / "manifest.json").write_text(json.dumps(_manifest()), encoding="utf-8")
|
|||
|
|
return src
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_build_roundtrip_and_bytes_fidelity() -> None:
|
|||
|
|
install_sdk_alias()
|
|||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|||
|
|
root = Path(tmp)
|
|||
|
|
src = _make_source(root)
|
|||
|
|
out = root / "t.demo-1.0.0.mbank"
|
|||
|
|
summary = AlgoBuilder(src, verbose=False).build(out)
|
|||
|
|
|
|||
|
|
assert out.is_file() and summary["algo_id"] == "t.demo"
|
|||
|
|
assert summary["sha256"] == sha256_file(out)
|
|||
|
|
|
|||
|
|
pkg = AlgoPackage(out)
|
|||
|
|
pkg.open()
|
|||
|
|
manifest = pkg.require_manifest()
|
|||
|
|
assert manifest.version == "1.0.0"
|
|||
|
|
data, meta = pkg.read_model("default")
|
|||
|
|
assert data == (src / "models" / "t.bmodel").read_bytes() # 包内模型与源文件逐字节相同
|
|||
|
|
assert meta["precision"] == "fp16"
|
|||
|
|
assert pkg.read_asset("assets/labels.txt").decode().splitlines() == ["person", "car"]
|
|||
|
|
files = {item["path"] for item in pkg.list_files()}
|
|||
|
|
assert {"manifest.json", "assets/labels.txt", "models/t.bmodel"} <= files
|
|||
|
|
pkg.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_build_creates_version_snapshot_and_signature() -> None:
|
|||
|
|
from sqlite_vfs.core import SQLiteVFS
|
|||
|
|
|
|||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|||
|
|
root = Path(tmp)
|
|||
|
|
src = _make_source(root)
|
|||
|
|
out = root / "signed.mbank"
|
|||
|
|
AlgoBuilder(src, verbose=False).build(out, sign_key="KEY-1", sign_key_id="ci")
|
|||
|
|
vfs = SQLiteVFS(str(out), readonly=True, check_same_thread=False)
|
|||
|
|
try:
|
|||
|
|
assert [s["name"] for s in vfs.list_snapshots()] == ["1.0.0"]
|
|||
|
|
report = vfs.verify_signature("KEY-1")
|
|||
|
|
assert report["ok"] and report["key_id"] == "ci"
|
|||
|
|
assert not vfs.verify_signature("KEY-WRONG")["ok"]
|
|||
|
|
finally:
|
|||
|
|
vfs.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_build_reports_missing_model_and_code() -> None:
|
|||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|||
|
|
root = Path(tmp)
|
|||
|
|
src = _make_source(root)
|
|||
|
|
(src / "models" / "t.bmodel").unlink()
|
|||
|
|
with pytest.raises(Exception) as exc:
|
|||
|
|
AlgoBuilder(src, verbose=False).build(root / "x.mbank")
|
|||
|
|
assert "模型" in str(exc.value)
|
|||
|
|
|
|||
|
|
src2 = _make_source(root / "other")
|
|||
|
|
import shutil
|
|||
|
|
shutil.rmtree(src2 / "alg_t_demo")
|
|||
|
|
with pytest.raises(Exception) as exc2:
|
|||
|
|
AlgoBuilder(src2, verbose=False).build(root / "y.mbank")
|
|||
|
|
assert "代码目录" in str(exc2.value)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_encrypt_removes_plain_source() -> None:
|
|||
|
|
from sqlite_vfs.core import SQLiteVFS
|
|||
|
|
|
|||
|
|
key = "uA9mYq0T2wGZ0dX1vJ8kL3pR5sN7bC4eF6hK2mP0qT8="
|
|||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|||
|
|
root = Path(tmp)
|
|||
|
|
src = _make_source(root)
|
|||
|
|
out = root / "enc.mbank"
|
|||
|
|
AlgoBuilder(src, key=key, verbose=False).build(out)
|
|||
|
|
vfs = SQLiteVFS(str(out), readonly=True, check_same_thread=False)
|
|||
|
|
try:
|
|||
|
|
files = [e["path"] for e in vfs.all_entries() if not e["is_directory"]]
|
|||
|
|
finally:
|
|||
|
|
vfs.close()
|
|||
|
|
assert not any(f.endswith(".py") for f in files) # 明文源码不得留在包里
|
|||
|
|
# 不带 key 打开结构没问题；装载（import）会因缺密钥失败
|
|||
|
|
pkg = AlgoPackage(out)
|
|||
|
|
pkg.open()
|
|||
|
|
with pytest.raises(AlgoPackageError) as exc:
|
|||
|
|
pkg.mount()
|
|||
|
|
assert "密钥" in str(exc.value)
|
|||
|
|
pkg.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_bad_file_rejected() -> None:
|
|||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|||
|
|
bad = Path(tmp) / "bad.mbank"
|
|||
|
|
bad.write_bytes(b"not a database at all")
|
|||
|
|
with pytest.raises(AlgoPackageError):
|
|||
|
|
AlgoPackage(bad).open()
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------- 多模型包
|
|||
|
|
|
|||
|
|
def _make_multi_model_source(root: Path) -> Path:
|
|||
|
|
"""一个包带两份模型（检测 + 二次判定）：代码按名字取第二份（多模型包的标准写法）。"""
|
|||
|
|
src = root / "t.multi"
|
|||
|
|
(src / "alg_t_multi").mkdir(parents=True)
|
|||
|
|
(src / "alg_t_multi" / "__init__.py").write_text(
|
|||
|
|
"from video_algpkg_api import Algorithm as Base, AlgoResult\n"
|
|||
|
|
"class Algorithm(Base):\n"
|
|||
|
|
" def load(self):\n"
|
|||
|
|
" self._det = self.ctx.model_bytes() # 选中模型（默认检测器）\n"
|
|||
|
|
" self._cls = self.ctx.model_bytes('fp_filter') # 按名字取第二份\n"
|
|||
|
|
" def infer(self, frame):\n"
|
|||
|
|
" return AlgoResult(boxes=None, value=len(self._det) + len(self._cls))\n",
|
|||
|
|
encoding="utf-8",
|
|||
|
|
)
|
|||
|
|
(src / "models").mkdir()
|
|||
|
|
(src / "models" / "det.bmodel").write_bytes(b"DET" * 256)
|
|||
|
|
(src / "models" / "fp.bmodel").write_bytes(b"FPFILTER" * 128)
|
|||
|
|
(src / "manifest.json").write_text(json.dumps(_manifest(
|
|||
|
|
id="t.multi", name="多模型包", code_package="alg_t_multi",
|
|||
|
|
models=[{"name": "detector", "file": "models/det.bmodel", "default": True},
|
|||
|
|
{"name": "fp_filter", "file": "models/fp.bmodel", "precision": "int8"}])),
|
|||
|
|
encoding="utf-8")
|
|||
|
|
return src
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_multi_model_package_context_access() -> None:
|
|||
|
|
"""`ctx.model_bytes(name)` 按 manifest.models[] 取包内任意模型；不给名 = 选中模型（旧写法兼容）。"""
|
|||
|
|
install_sdk_alias()
|
|||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|||
|
|
root = Path(tmp)
|
|||
|
|
src = _make_multi_model_source(root)
|
|||
|
|
out = root / "multi.mbank"
|
|||
|
|
AlgoBuilder(src, verbose=False).build(out)
|
|||
|
|
|
|||
|
|
pkg = AlgoPackage(out)
|
|||
|
|
pkg.open()
|
|||
|
|
pkg.mount()
|
|||
|
|
try:
|
|||
|
|
instance = pkg.new_instance("detector", {})
|
|||
|
|
ctx = instance.ctx
|
|||
|
|
det = (src / "models" / "det.bmodel").read_bytes()
|
|||
|
|
fp = (src / "models" / "fp.bmodel").read_bytes()
|
|||
|
|
assert ctx.model_name == "detector"
|
|||
|
|
assert ctx.model_bytes() == det # 选中模型（向后兼容）
|
|||
|
|
assert ctx.model_bytes("fp_filter") == fp # 按 name 取第二份
|
|||
|
|
assert ctx.model_bytes("models/fp.bmodel") == fp # 按 file 路径也认
|
|||
|
|
assert [m["name"] for m in ctx.models] == ["detector", "fp_filter"]
|
|||
|
|
assert ctx.model_spec("fp_filter")["precision"] == "int8"
|
|||
|
|
assert ctx.model_spec()["name"] == "detector"
|
|||
|
|
with pytest.raises(AlgoError) as exc:
|
|||
|
|
ctx.model_bytes("nope")
|
|||
|
|
assert "fp_filter" in str(exc.value) # 报错带可选清单
|
|||
|
|
assert instance.infer(None).value == len(det) + len(fp) # 包内代码真读到两份不同字节
|
|||
|
|
finally:
|
|||
|
|
pkg.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------- 契约名（video_algpkg_api）
|
|||
|
|
|
|||
|
|
def test_contract_module_importable_without_prior_registration() -> None:
|
|||
|
|
"""`video_algpkg_api` 是随 SDK 分发的**真实模块**，不只是运行时注册的别名。
|
|||
|
|
|
|||
|
|
新解释器里**先** import 它也要成功（本地开发/IDE/类型检查的常见姿势），
|
|||
|
|
且与 `algpkg_sdk.api` 是同一批对象（重导出，非副本）——否则 worker 侧
|
|||
|
|
`isinstance(instance, Algorithm)` 会出现两个基类（"本地过、设备挂"）。
|
|||
|
|
"""
|
|||
|
|
script = (
|
|||
|
|
"import video_algpkg_api as contract\n"
|
|||
|
|
"assert contract.__file__.endswith('video_algpkg_api/__init__.py'), contract.__file__\n"
|
|||
|
|
"import algpkg_sdk.api as api\n"
|
|||
|
|
"for name in contract.__all__:\n"
|
|||
|
|
" assert getattr(contract, name) is getattr(api, name), name\n"
|
|||
|
|
"print('contract-ok')\n"
|
|||
|
|
)
|
|||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|||
|
|
proc = subprocess.run([sys.executable, "-c", script], capture_output=True,
|
|||
|
|
text=True, timeout=30, cwd=tmp)
|
|||
|
|
assert proc.returncode == 0, proc.stderr
|
|||
|
|
assert "contract-ok" in proc.stdout
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_contract_alias_still_registered_to_api_module() -> None:
|
|||
|
|
"""worker 侧注册语义不变：导入 `algpkg_sdk` 后，`video_algpkg_api` 解析到 api 模块本身。
|
|||
|
|
|
|||
|
|
（`install_sdk_alias` 用 setdefault：设备上 worker 先注册，真实模块被惰性遮蔽，
|
|||
|
|
两条路径拿到的是同一批对象。）
|
|||
|
|
"""
|
|||
|
|
import video_algpkg_api as resolved
|
|||
|
|
|
|||
|
|
from algpkg_sdk import api
|
|||
|
|
|
|||
|
|
assert resolved is api
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------- CLI
|
|||
|
|
|
|||
|
|
def test_cli_init_build_inspect_check(tmp_path: Path) -> None:
|
|||
|
|
src = tmp_path / "cli.demo"
|
|||
|
|
assert cli_main(["init", str(src), "--id", "cli.demo", "--name", "CLI 示例"]) == 0
|
|||
|
|
assert (src / "manifest.json").is_file()
|
|||
|
|
code_pkg = next((src.glob("alg_cli_demo")), None)
|
|||
|
|
assert code_pkg is not None and (code_pkg / "__init__.py").is_file()
|
|||
|
|
|
|||
|
|
model = tmp_path / "fake.bmodel"
|
|||
|
|
model.write_bytes(b"FAKE" * 1024)
|
|||
|
|
out = tmp_path / "cli.demo-1.0.0.mbank"
|
|||
|
|
assert cli_main(["build", str(src), "--out", str(out), "--model", f"default={model}",
|
|||
|
|
"--quiet"]) == 0
|
|||
|
|
assert cli_main(["inspect", str(out), "--json"]) == 0
|
|||
|
|
assert cli_main(["verify", str(out), "--json"]) == 0
|
|||
|
|
assert cli_main(["list", str(tmp_path)]) == 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_cli_check_reports_static_results(tmp_path: Path) -> None:
|
|||
|
|
"""check 在无设备时也要给出"清单/结构 OK、试加载跳过"的诚实结论。"""
|
|||
|
|
src = tmp_path / "chk.demo"
|
|||
|
|
cli_main(["init", str(src), "--id", "chk.demo", "--name", "检查示例"])
|
|||
|
|
model = tmp_path / "fake.bmodel"
|
|||
|
|
model.write_bytes(b"FAKE" * 1024)
|
|||
|
|
out = tmp_path / "chk.mbank"
|
|||
|
|
cli_main(["build", str(src), "--out", str(out), "--model", f"default={model}", "--quiet"])
|
|||
|
|
report = testkit.check(out)
|
|||
|
|
names = {item["name"]: item for item in report["checks"]}
|
|||
|
|
assert names["manifest 合法"]["ok"]
|
|||
|
|
assert names["模型文件在包内"]["ok"]
|
|||
|
|
if not testkit.has_device():
|
|||
|
|
assert not names["代码挂载（试加载）"]["ok"] and "跳过" in names["代码挂载（试加载）"]["detail"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_module_entry_point_help() -> None:
|
|||
|
|
proc = subprocess.run([sys.executable, "-m", "algpkg_sdk.cli", "--help"],
|
|||
|
|
capture_output=True, text=True, timeout=30)
|
|||
|
|
assert proc.returncode == 0
|
|||
|
|
assert "check" in proc.stdout and "run" in proc.stdout and "compare" in proc.stdout
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------- 示例包（真实结构）
|
|||
|
|
|
|||
|
|
def test_example_person_count_manifest() -> None:
|
|||
|
|
example = EXAMPLES / "person-count"
|
|||
|
|
manifest = Manifest.parse(json.loads((example / "manifest.json").read_text()))
|
|||
|
|
assert manifest.id == "demo.person-count" and manifest.kind == "counter"
|
|||
|
|
assert manifest.output["type"] == "counter"
|
|||
|
|
assert (example / manifest.code_package).is_dir()
|
|||
|
|
# 模型文件由打包时 --model 指定（示例目录不带 bmodel），这里只校验声明形态
|
|||
|
|
assert manifest.default_model().file.startswith("models/")
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------- 设备用例
|
|||
|
|
|
|||
|
|
@pytest.mark.device
|
|||
|
|
def test_check_deep_on_device() -> None:
|
|||
|
|
"""设备：入口类可实例化（模型真加载）。
|
|||
|
|
|
|||
|
|
默认用**自包含**示例包（examples/person-count 打出来的包）；内置包（依赖 worker 内部模块）
|
|||
|
|
属"特权包"，需额外传 `extra_path=[worker 工程目录]`。
|
|||
|
|
"""
|
|||
|
|
import os
|
|||
|
|
|
|||
|
|
pkg_path = os.environ.get("ALGPKG_DEVICE_PACKAGE")
|
|||
|
|
if not pkg_path or not Path(pkg_path).is_file():
|
|||
|
|
pytest.skip("未提供 ALGPKG_DEVICE_PACKAGE（指向含真实模型的 .mbank）")
|
|||
|
|
extra = [p for p in (os.environ.get("ALGPKG_EXTRA_PATH") or "").split(":") if p]
|
|||
|
|
report = testkit.check(pkg_path, deep=True, extra_path=extra)
|
|||
|
|
assert report["ok"], report["checks"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.mark.device
|
|||
|
|
def test_run_on_device() -> None:
|
|||
|
|
"""设备：跑一段视频，拿到逐帧检出与单帧耗时。"""
|
|||
|
|
import os
|
|||
|
|
|
|||
|
|
pkg_path = os.environ.get("ALGPKG_DEVICE_PACKAGE")
|
|||
|
|
clip = os.environ.get("ALGPKG_DEVICE_CLIP")
|
|||
|
|
if not pkg_path or not clip:
|
|||
|
|
pytest.skip("未提供 ALGPKG_DEVICE_PACKAGE / ALGPKG_DEVICE_CLIP")
|
|||
|
|
extra = [p for p in (os.environ.get("ALGPKG_EXTRA_PATH") or "").split(":") if p]
|
|||
|
|
report = testkit.run(pkg_path, source=clip, frames=3, warmup=1, extra_path=extra)
|
|||
|
|
summary = report.summary()
|
|||
|
|
assert summary["frames"] >= 1
|
|||
|
|
assert summary["infer_ms"]["avg"] > 0
|