"""Offline unit tests for hyperstack_burst.py. No network, no VM creation, no GPU: manifest parsing/validation, workflow filling, frame math, cost estimation, and key sourcing are tested in isolation by importing the module (hyperstack SDK import is deferred inside functions). """ import json import re import sys from pathlib import Path import pytest SCRIPTS = Path(__file__).resolve().parents[2] / "moldovan-qwen" / "scripts" sys.path.insert(0, str(SCRIPTS)) import hyperstack_burst as hb # noqa: E402 @pytest.fixture def artwork_dir(tmp_path): d = tmp_path / "artwork" d.mkdir() for i in (1, 2, 3): (d / f"cover_{i}.png").write_bytes(b"\x89PNG\r\n\x1a\n" + b"stub" * 8) return d @pytest.fixture def scenes(artwork_dir): return [ {"scene_num": 1, "artwork": "cover_1.png", "prompt": "zoom", "duration": 5, "resolution": "480p"}, {"scene_num": 2, "artwork": "cover_2.png", "prompt": "pan", "duration": 4.5, "resolution": "720p"}, ] def test_load_manifest_list(scenes, tmp_path): p = tmp_path / "m.json" p.write_text(json.dumps(scenes)) assert hb.load_manifest(p) == scenes def test_load_manifest_dict_wrapper(scenes, tmp_path): p = tmp_path / "m.json" p.write_text(json.dumps({"scenes": scenes})) assert hb.load_manifest(p) == scenes def test_load_manifest_rejects_empty(tmp_path): p = tmp_path / "m.json" p.write_text(json.dumps([])) with pytest.raises(ValueError): hb.load_manifest(p) def test_validate_scenes_ok(scenes, artwork_dir): assert hb.validate_scenes(scenes, artwork_dir) == [] def test_validate_scenes_missing_artwork(scenes, artwork_dir): scenes[0]["artwork"] = "nope.png" errors = hb.validate_scenes(scenes, artwork_dir) assert any("cover" not in e and "missing" in e for e in errors) def test_validate_scenes_missing_prompt(artwork_dir): scenes = [{"scene_num": 1, "artwork": "cover_1.png"}] errors = hb.validate_scenes(scenes, artwork_dir) assert any("prompt" in e for e in errors) def test_validate_scenes_bad_resolution(scenes, artwork_dir): scenes[0]["resolution"] = "1080p" assert any("resolution" in e for e in hb.validate_scenes(scenes, artwork_dir)) def test_validate_scenes_bad_duration(scenes, artwork_dir): scenes[0]["duration"] = 30 assert any("duration" in e for e in hb.validate_scenes(scenes, artwork_dir)) def test_frames_for_is_4n_plus_1(): assert hb.frames_for(5) == 81 # 5s * 16fps / 4 -> 20 -> 4*20+1 for d in (1, 2, 3, 4.5, 5, 7.5, 10): f = hb.frames_for(d) assert f >= 5 and (f - 1) % 4 == 0 def test_fill_workflow_injects_scene_fields(scenes, artwork_dir): wf = hb.fill_workflow(scenes[0], artwork_dir) assert wf["42"]["inputs"]["text"] == "zoom" assert wf["44"]["inputs"]["image"] == "cover_1.png" assert wf["45"]["inputs"]["length"] == 81 assert wf["45"]["inputs"]["width"] == 832 assert wf["46"]["inputs"]["seed"] == 1001 assert wf["49"]["inputs"]["filename_prefix"] == "burst-scene_001" # negative default present assert "blurry" in wf["43"]["inputs"]["text"] def test_fill_workflow_720p_uses_720_unet(scenes, artwork_dir): wf = hb.fill_workflow(scenes[1], artwork_dir) assert "720p" in wf["37"]["inputs"]["unet_name"] assert wf["45"]["inputs"]["width"] == 1280 def test_fill_workflow_template_not_mutated(scenes, artwork_dir): before = json.dumps(hb.WORKFLOW_TEMPLATE) hb.fill_workflow(scenes[0], artwork_dir) assert json.dumps(hb.WORKFLOW_TEMPLATE) == before def test_workflow_graph_integrity(scenes, artwork_dir): """All internal references point to existing nodes with correct output index arity.""" wf = hb.fill_workflow(scenes[0], artwork_dir) for node in wf.values(): for v in node["inputs"].values(): if isinstance(v, list) and len(v) == 2 and isinstance(v[0], str): assert v[0] in wf, f"dangling reference {v}" # KSampler consumes WanImageToVideo's 3 outputs (pos/neg/latent) assert wf["46"]["inputs"]["positive"] == ["45", 0] assert wf["46"]["inputs"]["negative"] == ["45", 1] assert wf["46"]["inputs"]["latent_image"] == ["45", 2] def test_estimate_cost_scales(): e1 = hb.estimate_cost([{}] * 10) e2 = hb.estimate_cost([{}] * 50) assert e2["estimated_cost_usd"] > e1["estimated_cost_usd"] assert e1["rate_per_hour"] == 1.28 def test_bootstrap_embeds_hf_repo(): assert "Comfy-Org/Wan_2.1_ComfyUI_repackaged" in hb.BOOTSTRAP assert "COMFY_REPO" not in hb.BOOTSTRAP assert "main.py --listen 127.0.0.1" in hb.BOOTSTRAP def test_constants_match_proven_recipe(): assert hb.BASE == "https://infrahub-api.nexgencloud.com/v1" assert hb.FLAVOR_SPOT == "n3-A100x1-spot" assert hb.ENV_NAME == "default-CANADA-1" def test_no_secrets_in_source(): """Guardrail: no hardcoded key value, only env/file sourcing of the key name.""" src = Path(hb.__file__).read_text() assert not re.search(r"HYPERSTACK_API_KEY\s*=\s*[\"']?[A-Za-z0-9_\-]{20,}", src) assert not re.search(r"[\"']sk-[A-Za-z0-9]{16,}[\"']", src) def test_build_manifest_from_dir(artwork_dir): m = hb.build_manifest_from_dir(artwork_dir) assert len(m) == 3 assert [s["artwork"] for s in m] == ["cover_1.png", "cover_2.png", "cover_3.png"] assert [s["scene_num"] for s in m] == [1, 2, 3] assert all(s["prompt"] for s in m) def test_build_manifest_custom_prompt_duration(artwork_dir): (artwork_dir / "sketch.txt").write_text("ignore me") m = hb.build_manifest_from_dir(artwork_dir, prompt="drift", duration=6, resolution="720p") assert len(m) == 3 # .txt filtered out assert m[0]["prompt"] == "drift" and m[0]["duration"] == 6 and m[0]["resolution"] == "720p" def test_build_manifest_empty_dir_raises(tmp_path): d = tmp_path / "empty" d.mkdir() with pytest.raises(ValueError): hb.build_manifest_from_dir(d) def test_notify_is_silent_no_crash(): hb.notify("t", "b") # osascript missing/blocked must not raise def test_run_burst_detached_spawns_child(tmp_path, monkeypatch): """Detach writes manifest + spawns the CLI; we stub Popen to avoid a real run.""" launched = {} class FakePopen: def __init__(self, cmd, **kw): launched["cmd"] = cmd monkeypatch.setattr(hb.subprocess, "Popen", FakePopen) scenes = [{"scene_num": 1, "artwork": "x.png", "prompt": "p"}] hb.run_burst_detached(scenes, tmp_path, tmp_path / "out") assert "--go" in launched["cmd"] and "--notify-done" in launched["cmd"] saved = json.loads((hb.STATE_DIR / "manifest.json").read_text()) assert saved == scenes def test_check_cost_guard_blocks_over_cap(): est = {"estimated_cost_usd": 4.0, "estimated_minutes": 225, "scenes": 50, "rate_per_hour": 1.28} with pytest.raises(SystemExit): hb.check_cost_guard(est, 3.0) def test_check_cost_guard_allows_under_cap(): est = {"estimated_cost_usd": 1.39, "estimated_minutes": 65, "scenes": 10, "rate_per_hour": 1.28} hb.check_cost_guard(est, 5.0) # no raise def test_model_sampling_sd3_matches_official_template(): """Official Comfy-Org Wan2.1 I2V template routes UNET through ModelSamplingSD3 (shift=8).""" wf = hb.fill_workflow({"scene_num": 1, "artwork": "cover_1.png", "prompt": "x"}, Path(".")) assert wf["36"]["class_type"] == "ModelSamplingSD3" assert wf["36"]["inputs"]["model"] == ["37", 0] assert wf["36"]["inputs"]["shift"] == 8 assert wf["46"]["inputs"]["model"] == ["36", 0] # KSampler consumes shifted model def test_official_template_crosscheck(): """Cross-check node types and key widgets against the official template file if cached.""" official = Path.home() / ".flomaster" / "scratch" / "wan_official.json" if not official.exists(): import pytest pytest.skip("official template not cached") d = json.loads(official.read_text())["nodes"] off_types = {n["type"] for n in d if n.get("type") != "MarkdownNote"} wf = hb.fill_workflow({"scene_num": 1, "artwork": "cover_1.png", "prompt": "x"}, Path(".")) my_types = {v["class_type"] for v in wf.values()} missing = off_types - my_types assert not missing, f"workflow missing official node types: {missing}"