File size: 1,896 Bytes
ebeed67 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 | from __future__ import annotations
import io
import json
from camstack_train.cli import main
from camstack_train.events import Emitter
from camstack_train.run import Deps
from .conftest import frame, subject
from .fakes import FakeToolchain, detector_factory, fake_fetch, fake_trainer
def test_bad_spec_exits_64_with_a_failed_result(tmp_path, job_dict):
job_dict["schema"] = "camstack.training/v0"
(tmp_path / "job.json").write_text(json.dumps(job_dict))
buf = io.StringIO()
code = main(["run", "--job", str(tmp_path / "job.json"), "--dataset", str(tmp_path / "d.tar"),
"--out", str(tmp_path / "out"), "--cache", str(tmp_path / "c")], deps=Deps(emitter=Emitter(buf)))
assert code == 64
result = json.loads((tmp_path / "out" / "result.json").read_text())
assert result["status"] == "failed" and result["jobKey"] == job_dict["jobKey"]
assert result["error"]["code"] == "unknown-schema"
def test_run_ok_exits_0(tmp_path, job_dict, make_export):
path, digest = make_export([frame(f"a{i}", 1, i, [subject("person")]) for i in range(3)])
job_dict["dataset"]["sha256"] = digest
(tmp_path / "job.json").write_text(json.dumps(job_dict))
deps = Deps(fetch=fake_fetch, trainer=fake_trainer, toolchain=FakeToolchain(),
detector=detector_factory(), emitter=Emitter(io.StringIO()))
code = main(["run", "--job", str(tmp_path / "job.json"), "--dataset", str(path), "--out", str(tmp_path / "out"),
"--cache", str(tmp_path / "c"), "--work", str(tmp_path / "w")], deps=deps)
assert code == 0
def test_dataset_command(tmp_path, make_export, capsys):
path, _ = make_export([frame("a", 1, 0, [subject("person")]), frame("b", 1, 0, [subject("animal")])])
assert main(["dataset", "--export", str(path), "--out", str(tmp_path / "o")]) == 0
assert "1 frames, 1 boxes" in capsys.readouterr().out
|