Spaces:
Running on Zero
Running on Zero
| import app | |
| from nexus_visual_weaver.planner import build_command_center_run | |
| # --- _checkpoint_seed tests --- | |
| def test_checkpoint_seed_parses_valid_hex_suffix() -> None: | |
| # "nw-" prefix + "abcdef12" -> int("abcdef12", 16) % 1_000_000 | |
| checkpoint_id = "nw-abcdef12" | |
| result = app._checkpoint_seed(checkpoint_id) | |
| assert result == int("abcdef12", 16) % 1_000_000 | |
| def test_checkpoint_seed_handles_empty_suffix() -> None: | |
| assert app._checkpoint_seed("") == 0 | |
| assert app._checkpoint_seed("nw-") == 0 | |
| def test_checkpoint_seed_handles_non_hex_suffix() -> None: | |
| # No hex chars -> return 0 | |
| result = app._checkpoint_seed("nw-zzzzzzzz") | |
| assert result == 0 | |
| def test_checkpoint_seed_ignores_non_hex_chars_in_last_8() -> None: | |
| # Strips non-hex: "abc-xyz" -> only "abcef" are hex in the last 8 chars ("-xyz" strips to "abc") | |
| # "0000000g" -> only "0000000" are valid hex | |
| result = app._checkpoint_seed("0000000g") | |
| assert result == int("0000000", 16) % 1_000_000 | |
| assert result == 0 | |
| def test_checkpoint_seed_returns_value_within_range() -> None: | |
| checkpoint_id = "nw-deadbeef" | |
| result = app._checkpoint_seed(checkpoint_id) | |
| assert 0 <= result < 1_000_000 | |
| def test_checkpoint_seed_uses_only_last_8_chars() -> None: | |
| # Long checkpoint id: only last 8 chars considered | |
| # "nw-aabbccddee11223344" -> last 8 is "11223344" | |
| result = app._checkpoint_seed("nw-aabbccddee11223344") | |
| assert result == int("11223344", 16) % 1_000_000 | |
| # --- _wardrobe_summary tests --- | |
| def test_wardrobe_summary_returns_string_with_slot_fields() -> None: | |
| run = build_command_center_run("gothic patent leather platform boots, crimson hardware") | |
| summary = app._wardrobe_summary(run) | |
| # Summary is a semicolon-delimited string of slot info | |
| assert isinstance(summary, str) | |
| assert len(summary) > 0 | |
| # Should contain material= and palette= from slot info | |
| assert "material=" in summary | |
| assert "palette=" in summary | |
| assert "locked=" in summary | |
| def test_wardrobe_summary_handles_run_with_no_outfit() -> None: | |
| class FakeRun: | |
| outfit = None | |
| summary = app._wardrobe_summary(FakeRun()) | |
| assert summary == "" | |
| def test_wardrobe_summary_handles_none_run() -> None: | |
| # Pass something with no outfit attribute | |
| class Empty: | |
| pass | |
| summary = app._wardrobe_summary(Empty()) | |
| assert summary == "" | |
| def test_wardrobe_summary_includes_slot_names() -> None: | |
| run = build_command_center_run("dark couture archivist, patent leather, platform boots") | |
| summary = app._wardrobe_summary(run) | |
| # Slot names like footwear, outerwear, upper_body should appear | |
| assert any(slot_name in summary for slot_name in ["footwear", "outerwear", "upper_body", "jewelry"]) | |
| def test_run_weave_returns_stateful_dashboard_packet() -> None: | |
| result = app.run_weave( | |
| "gothic patent leather platform boots, crimson hardware", | |
| "Strict", | |
| "Wan2.2 I2V", | |
| False, | |
| None, | |
| "Forge", | |
| ) | |
| assert len(result) == 19 | |
| assert result[13].checkpoint.checkpoint_id.startswith("nw-") | |
| assert result[15]["provider_state"] == "dry-run" | |
| assert result[16]["interactive"] is False | |
| assert result[17]["interactive"] is False | |
| assert result[18]["interactive"] is False | |
| assert result[15]["creator_controls"]["wardrobe"]["footwear"] == "platform boots" | |
| assert result[15]["generation"]["lora_status"] == "disabled" | |
| def test_run_weave_persists_additive_creator_controls_and_reference_url_metadata() -> None: | |
| result = app.run_weave( | |
| "gothic patent leather platform boots, crimson hardware", | |
| "Frontier", | |
| "LTX-2.3", | |
| False, | |
| None, | |
| "Wardrobe", | |
| "layered tactical silhouette", | |
| "faux fur collar coat", | |
| "black mesh layer", | |
| "patent leather heels", | |
| "obsidian, pearl, crimson", | |
| "silver occult buckles", | |
| "https://shop.example.test/item/123", | |
| ) | |
| state = result[15] | |
| run = result[13] | |
| assert state["creator_controls"]["reasoning_mode"] == "Frontier" | |
| assert state["creator_controls"]["wardrobe"]["footwear"] == "patent leather heels" | |
| assert state["creator_controls"]["generation"]["seed"] >= 0 | |
| assert state["creator_controls"]["generation"]["style_strength"] == "High Fashion" | |
| assert state["creator_controls"]["generation"]["aspect"] == "Portrait" | |
| assert state["reference_metadata"][0]["source"] == "url" | |
| assert state["reference_metadata"][0]["domain"] == "shop.example.test" | |
| assert "url_hash" in state["reference_metadata"][0] | |
| assert run.request.creator_controls["wardrobe"]["outerwear"] == "faux fur collar coat" | |
| def test_operator_actions_transition_checkpoint_export_and_stop() -> None: | |
| from pathlib import Path | |
| result = app.run_weave( | |
| "gothic patent leather platform boots, crimson hardware", | |
| "Strict", | |
| "Wan2.2 I2V", | |
| False, | |
| None, | |
| "Forge", | |
| ) | |
| run = result[13] | |
| artifact_path = Path("outputs/test-generated-artifact.png") | |
| artifact_path.parent.mkdir(parents=True, exist_ok=True) | |
| artifact_path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100) | |
| operator_state = {**result[15], "generation": {**result[15]["generation"], "output_path": str(artifact_path)}} | |
| clean_scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []} | |
| approved = app.approve_checkpoint(run, False, clean_scan, "Forge", operator_state) | |
| assert approved[13]["checkpoint"] == "approved" | |
| assert approved[13]["provider_state"] == "export_ready" | |
| exported = app.export_packet(run, False, clean_scan, "Forge", approved[13]) | |
| assert exported[13]["provider_state"] == "exported" | |
| assert exported[13]["export"] == "clear" | |
| stopped = app.stop_provider_job(run, False, clean_scan, "Forge", operator_state) | |
| assert stopped[13]["provider_state"] == "stopped" | |
| def test_clear_export_ignores_stale_override_reason(monkeypatch) -> None: | |
| monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports") | |
| result = app.run_weave( | |
| "gothic patent leather platform boots, crimson hardware", | |
| "Strict", | |
| "Wan2.2 I2V", | |
| False, | |
| None, | |
| "Forge", | |
| ) | |
| run = result[13] | |
| artifact_path = app.ROOT / "outputs" / "test-clear-override-artifact.png" | |
| artifact_path.parent.mkdir(parents=True, exist_ok=True) | |
| artifact_path.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82") | |
| operator_state = { | |
| **result[15], | |
| "generation": {**result[15]["generation"], "output_path": str(artifact_path)}, | |
| } | |
| clean_scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []} | |
| approved = app.approve_checkpoint(run, False, clean_scan, "Forge", operator_state) | |
| exported = app.export_packet(run, False, clean_scan, "Forge", approved[13], "stale text from prior blocked run") | |
| assert exported[13]["provider_state"] == "exported" | |
| assert exported[13]["export"] == "clear" | |
| assert "st3gg_override_reason" not in exported[13] | |
| def test_export_blocks_without_checkpoint() -> None: | |
| result = app.run_weave( | |
| "gothic patent leather platform boots, crimson hardware", | |
| "Strict", | |
| "Wan2.2 I2V", | |
| False, | |
| None, | |
| "Forge", | |
| ) | |
| clean_scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []} | |
| blocked = app.export_packet(result[13], False, clean_scan, "Forge", result[15]) | |
| assert blocked[13]["provider_state"] == "blocked" | |
| assert "checkpoint" in blocked[13]["message"].lower() | |
| def test_reference_scan_cannot_clear_blocked_generated_artifact() -> None: | |
| base = app.ROOT / "outputs" / "test-app-callbacks" | |
| base.mkdir(parents=True, exist_ok=True) | |
| blocked_artifact = base / "blocked-generated.png" | |
| blocked_artifact.write_bytes( | |
| b"\x89PNG\r\n\x1a\n" | |
| b"\x00\x00\x00\rIEND\xaeB`\x82" | |
| b"NEXUS_TRAILING_PAYLOAD" | |
| ) | |
| clean_reference = base / "clean-reference.png" | |
| clean_reference.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82") | |
| run = build_command_center_run("gothic patent leather platform boots") | |
| state = { | |
| "provider_state": "checkpointed", | |
| "checkpoint": "pending_review", | |
| "export": "blocked", | |
| "generation": {"status": "success", "output_path": str(blocked_artifact)}, | |
| "generated_scan": app.scan_file(str(blocked_artifact)), | |
| } | |
| scanned = app.scan_reference(run, False, str(clean_reference), "Forge", state) | |
| assert scanned[13]["reference_scan"]["export_gate"] == "clear" | |
| assert scanned[13]["export"] == "blocked" | |
| assert scanned[17]["export_gate"] == "blocked" | |
| approved = app.approve_checkpoint(run, False, scanned[17], "Forge", scanned[13]) | |
| assert approved[13]["provider_state"] == "checkpointed" | |
| assert approved[13]["export"] == "blocked" | |
| def test_blocked_reference_scan_does_not_block_clear_generated_artifact() -> None: | |
| base = app.ROOT / "outputs" / "test-app-callbacks" | |
| base.mkdir(parents=True, exist_ok=True) | |
| generated = base / "clear-generated.png" | |
| generated.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82") | |
| blocked_reference = base / "blocked-reference.png" | |
| blocked_reference.write_bytes( | |
| b"\x89PNG\r\n\x1a\n" | |
| b"\x00\x00\x00\rIEND\xaeB`\x82" | |
| b"NEXUS_TRAILING_PAYLOAD" | |
| ) | |
| run = build_command_center_run("gothic patent leather platform boots") | |
| state = { | |
| "provider_state": "checkpointed", | |
| "checkpoint": "pending_review", | |
| "export": "clear", | |
| "generation": {"status": "success", "output_path": str(generated)}, | |
| "generated_scan": app.scan_file(str(generated)), | |
| } | |
| scanned = app.scan_reference(run, False, str(blocked_reference), "Forge", state) | |
| assert scanned[13]["reference_scan"]["export_gate"] == "blocked" | |
| assert scanned[13]["export"] == "clear" | |
| assert scanned[17]["export_gate"] == "clear" | |
| approved = app.approve_checkpoint(run, False, scanned[17], "Forge", scanned[13]) | |
| assert approved[13]["provider_state"] == "export_ready" | |
| assert approved[13]["export"] == "clear" | |
| def test_checkpoint_blocks_without_generated_artifact() -> None: | |
| result = app.run_weave( | |
| "gothic patent leather platform boots, crimson hardware", | |
| "Strict", | |
| "Wan2.2 I2V", | |
| False, | |
| None, | |
| "Forge", | |
| ) | |
| clean_scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []} | |
| blocked = app.approve_checkpoint(result[13], False, clean_scan, "Forge", result[15]) | |
| assert blocked[13]["provider_state"] == "blocked" | |
| assert "no generated artifact" in blocked[13]["message"].lower() | |
| # --- _safe_file_hash tests --- | |
| def test_safe_file_hash_returns_sha256_and_size_for_valid_file() -> None: | |
| import hashlib | |
| from pathlib import Path | |
| path = app.ROOT / "outputs" / "test-hash-target.png" | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32 | |
| path.write_bytes(data) | |
| file_hash, size = app._safe_file_hash(str(path)) | |
| assert file_hash == hashlib.sha256(data).hexdigest() | |
| assert size == len(data) | |
| def test_safe_file_hash_returns_none_for_none_path() -> None: | |
| assert app._safe_file_hash(None) == (None, None) | |
| def test_safe_file_hash_returns_none_for_empty_string() -> None: | |
| assert app._safe_file_hash("") == (None, None) | |
| def test_safe_file_hash_returns_none_for_missing_file() -> None: | |
| result = app._safe_file_hash("/nonexistent/path/no-such-file.png") | |
| assert result == (None, None) | |
| def test_safe_file_hash_hash_is_64_hex_chars() -> None: | |
| from pathlib import Path | |
| path = app.ROOT / "outputs" / "test-hash-hex.png" | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x01" * 8) | |
| file_hash, _ = app._safe_file_hash(str(path)) | |
| assert file_hash is not None | |
| assert len(file_hash) == 64 | |
| assert all(c in "0123456789abcdef" for c in file_hash) | |
| # --- _safe_reference_url_metadata tests --- | |
| def test_safe_reference_url_metadata_returns_none_for_falsy() -> None: | |
| assert app._safe_reference_url_metadata(None) is None | |
| assert app._safe_reference_url_metadata("") is None | |
| def test_safe_reference_url_metadata_returns_metadata_for_valid_https_url() -> None: | |
| result = app._safe_reference_url_metadata("https://shop.example.test/item/123") | |
| assert result is not None | |
| assert result["status"] == "metadata_only" | |
| assert result["domain"] == "shop.example.test" | |
| assert "url_hash" in result | |
| assert len(result["url_hash"]) == 64 | |
| def test_safe_reference_url_metadata_returns_metadata_for_valid_http_url() -> None: | |
| result = app._safe_reference_url_metadata("http://example.test/path?q=1") | |
| assert result is not None | |
| assert result["status"] == "metadata_only" | |
| assert result["domain"] == "example.test" | |
| def test_safe_reference_url_metadata_rejects_file_scheme() -> None: | |
| result = app._safe_reference_url_metadata("file:///etc/passwd") | |
| assert result is not None | |
| assert result["status"] == "invalid_url" | |
| def test_safe_reference_url_metadata_rejects_ftp_scheme() -> None: | |
| result = app._safe_reference_url_metadata("ftp://ftp.example.test/file.txt") | |
| assert result is not None | |
| assert result["status"] == "invalid_url" | |
| def test_safe_reference_url_metadata_rejects_missing_host() -> None: | |
| result = app._safe_reference_url_metadata("http:///no-host") | |
| assert result is not None | |
| assert result["status"] == "invalid_url" | |
| def test_safe_reference_url_metadata_domain_is_lowercase() -> None: | |
| result = app._safe_reference_url_metadata("https://SHOP.EXAMPLE.TEST/Item/123") | |
| assert result is not None | |
| assert result["domain"] == "shop.example.test" | |
| def test_safe_reference_url_metadata_url_hash_is_stable() -> None: | |
| url = "https://shop.example.test/stable-hash-test" | |
| first = app._safe_reference_url_metadata(url) | |
| second = app._safe_reference_url_metadata(url) | |
| assert first is not None and second is not None | |
| assert first["url_hash"] == second["url_hash"] | |
| def test_safe_reference_url_metadata_different_urls_produce_different_hashes() -> None: | |
| url_a = "https://shop.example.test/item/1" | |
| url_b = "https://shop.example.test/item/2" | |
| result_a = app._safe_reference_url_metadata(url_a) | |
| result_b = app._safe_reference_url_metadata(url_b) | |
| assert result_a is not None and result_b is not None | |
| assert result_a["url_hash"] != result_b["url_hash"] | |
| # --- _reference_metadata tests --- | |
| def test_reference_metadata_returns_empty_for_no_inputs() -> None: | |
| scan = {"status": "pass", "export_gate": "clear"} | |
| result = app._reference_metadata(None, None, scan) | |
| assert result == [] | |
| def test_reference_metadata_includes_file_record_when_uploaded() -> None: | |
| from pathlib import Path | |
| path = app.ROOT / "outputs" / "test-ref-meta.png" | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 16) | |
| scan = {"status": "pass", "export_gate": "clear", "magic": "PNG", "extension": ".png"} | |
| result = app._reference_metadata(str(path), None, scan) | |
| assert len(result) == 1 | |
| assert result[0]["source"] == "upload" | |
| assert result[0]["basename"] == "test-ref-meta.png" | |
| assert result[0]["sha256"] is not None | |
| assert result[0]["st3gg_status"] == "pass" | |
| assert result[0]["export_gate"] == "clear" | |
| assert result[0]["magic"] == "PNG" | |
| assert result[0]["extension"] == ".png" | |
| def test_reference_metadata_includes_url_record_when_url_given() -> None: | |
| scan = {} | |
| result = app._reference_metadata(None, "https://shop.example.test/item/99", scan) | |
| assert len(result) == 1 | |
| assert result[0]["source"] == "url" | |
| assert result[0]["domain"] == "shop.example.test" | |
| def test_reference_metadata_includes_both_file_and_url() -> None: | |
| from pathlib import Path | |
| path = app.ROOT / "outputs" / "test-ref-both.png" | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82") | |
| scan = {"status": "pass", "export_gate": "clear"} | |
| result = app._reference_metadata(str(path), "https://shop.example.test/item/77", scan) | |
| assert len(result) == 2 | |
| sources = {r["source"] for r in result} | |
| assert sources == {"upload", "url"} | |
| def test_reference_metadata_size_bytes_is_integer() -> None: | |
| from pathlib import Path | |
| data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 24 | |
| path = app.ROOT / "outputs" / "test-ref-size.png" | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(data) | |
| scan = {} | |
| result = app._reference_metadata(str(path), None, scan) | |
| assert result[0]["size_bytes"] == len(data) | |
| # --- _creator_controls tests --- | |
| def test_creator_controls_returns_dict_with_wardrobe_and_generation() -> None: | |
| result = app._creator_controls("Strict", "Wan2.2 I2V") | |
| assert "wardrobe" in result | |
| assert "generation" in result | |
| assert result["reasoning_mode"] == "Strict" | |
| assert result["video_preset"] == "Wan2.2 I2V" | |
| def test_creator_controls_default_wardrobe_values() -> None: | |
| result = app._creator_controls("Strict", "Wan2.2 I2V") | |
| wardrobe = result["wardrobe"] | |
| assert wardrobe["footwear"] == "platform boots" | |
| assert wardrobe["outerwear"] == "black patent leather long coat" | |
| assert wardrobe["palette"] == "black, crimson, cyan neon" | |
| def test_creator_controls_overrides_wardrobe_slots() -> None: | |
| result = app._creator_controls( | |
| "Frontier", | |
| "LTX-2.3", | |
| footwear="patent leather heels", | |
| outerwear="faux fur collar coat", | |
| palette="obsidian, pearl, crimson", | |
| ) | |
| wardrobe = result["wardrobe"] | |
| assert wardrobe["footwear"] == "patent leather heels" | |
| assert wardrobe["outerwear"] == "faux fur collar coat" | |
| assert wardrobe["palette"] == "obsidian, pearl, crimson" | |
| def test_creator_controls_locked_slots_present() -> None: | |
| result = app._creator_controls("Strict", "Wan2.2 I2V") | |
| wardrobe = result["wardrobe"] | |
| assert "locked_slots" in wardrobe | |
| assert isinstance(wardrobe["locked_slots"], list) | |
| assert "outerwear" in wardrobe["locked_slots"] | |
| def test_creator_controls_generation_specifies_flux_primary() -> None: | |
| result = app._creator_controls("Strict", "Wan2.2 I2V") | |
| generation = result["generation"] | |
| assert "black-forest-labs/FLUX.2-klein-9B" in generation["flux_primary"] | |
| assert "black-forest-labs/FLUX.2-klein-4B" in generation["flux_sidecar"] | |
| def test_creator_controls_generation_persists_seed_style_and_aspect() -> None: | |
| result = app._creator_controls( | |
| "Strict", | |
| "Wan2.2 I2V", | |
| seed=123, | |
| style_strength="Cinematic", | |
| aspect="Square", | |
| ) | |
| generation = result["generation"] | |
| assert generation["seed"] == 123 | |
| assert generation["style_strength"] == "Cinematic" | |
| assert generation["aspect"] == "Square" | |
| # --- _prompt_with_controls tests --- | |
| def test_prompt_with_controls_appends_wardrobe_suffix() -> None: | |
| controls = app._creator_controls("Strict", "Wan2.2 I2V", footwear="platform boots") | |
| result = app._prompt_with_controls("gothic couture portrait", controls) | |
| assert result.startswith("gothic couture portrait") | |
| assert "Wardrobe controls:" in result | |
| assert "platform boots" in result | |
| assert "Style direction:" in result | |
| def test_prompt_with_controls_returns_prompt_unchanged_when_no_wardrobe() -> None: | |
| # A controls dict with no wardrobe key | |
| result = app._prompt_with_controls("gothic couture portrait", {}) | |
| assert result == "gothic couture portrait" | |
| def test_prompt_with_controls_includes_all_wardrobe_fields() -> None: | |
| controls = app._creator_controls( | |
| "Strict", | |
| "Wan2.2 I2V", | |
| silhouette="structured long coat", | |
| outerwear="black patent leather long coat", | |
| upper_body="Chantilly lace neckline", | |
| footwear="platform boots", | |
| palette="black, crimson, cyan neon", | |
| hardware="crimson hardware", | |
| ) | |
| result = app._prompt_with_controls("brief", controls) | |
| assert "structured long coat" in result | |
| assert "black patent leather long coat" in result | |
| assert "Chantilly lace neckline" in result | |
| assert "platform boots" in result | |
| assert "crimson hardware" in result | |
| def test_generation_dimensions_support_portrait_and_square() -> None: | |
| assert app._generation_dimensions("Portrait") == (832, 1216) | |
| assert app._generation_dimensions("Square") == (1024, 1024) | |
| assert app._generation_dimensions("unknown") == (832, 1216) | |
| def test_resolve_seed_randomizes_negative_and_uses_explicit_seed() -> None: | |
| assert app._resolve_seed(42) == 42 | |
| assert app._resolve_seed("42") == 42 | |
| assert 0 <= app._resolve_seed(-1) < 1_000_000_000 | |
| def test_button_updates_unlock_checkpoint_then_export() -> None: | |
| state = { | |
| "provider_state": "generated", | |
| "checkpoint": "pending_review", | |
| "generation": {"status": "success", "output_path": "outputs/runtime/test.png"}, | |
| } | |
| checkpoint, export, stop = app._button_updates(None, state) | |
| assert checkpoint["interactive"] is True | |
| assert export["interactive"] is False | |
| assert stop["interactive"] is False | |
| approved_state = {**state, "checkpoint": "approved"} | |
| checkpoint, export, stop = app._button_updates(None, approved_state) | |
| assert checkpoint["interactive"] is False | |
| assert export["interactive"] is True | |
| assert stop["interactive"] is False | |
| def test_prompt_with_controls_does_not_duplicate_prompt() -> None: | |
| controls = app._creator_controls("Strict", "Wan2.2 I2V") | |
| prompt = "gothic couture brief" | |
| result = app._prompt_with_controls(prompt, controls) | |
| assert result.count(prompt) == 1 | |
| # --- _generated_output_path tests --- | |
| def test_generated_output_path_returns_none_when_no_state() -> None: | |
| assert app._generated_output_path(None) is None | |
| def test_generated_output_path_returns_none_when_no_generation_key() -> None: | |
| assert app._generated_output_path({"provider_state": "idle"}) is None | |
| def test_generated_output_path_returns_none_when_no_output_path() -> None: | |
| state = {"generation": {"status": "disabled"}} | |
| assert app._generated_output_path(state) is None | |
| def test_generated_output_path_returns_string_when_present() -> None: | |
| state = {"generation": {"output_path": "/data/artifact.png"}} | |
| result = app._generated_output_path(state) | |
| assert result == "/data/artifact.png" | |
| def test_generated_output_path_returns_none_for_empty_path() -> None: | |
| state = {"generation": {"output_path": ""}} | |
| assert app._generated_output_path(state) is None | |
| # --- _authoritative_generated_scan tests --- | |
| def test_authoritative_generated_scan_returns_default_for_no_state() -> None: | |
| result = app._authoritative_generated_scan(None) | |
| assert isinstance(result, dict) | |
| assert "status" in result | |
| def test_authoritative_generated_scan_scans_output_path_when_present() -> None: | |
| from pathlib import Path | |
| path = app.ROOT / "outputs" / "test-auth-scan.png" | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82") | |
| state = {"generation": {"output_path": str(path)}} | |
| result = app._authoritative_generated_scan(state) | |
| assert result["status"] in {"pass", "review"} | |
| assert "export_gate" in result | |
| def test_authoritative_generated_scan_falls_back_to_stored_scan() -> None: | |
| stored_scan = {"status": "review", "export_gate": "blocked", "findings": ["trailing data"]} | |
| state = {"generated_scan": stored_scan} | |
| result = app._authoritative_generated_scan(state) | |
| assert result == stored_scan | |
| def test_authoritative_generated_scan_uses_output_path_over_stored_scan() -> None: | |
| from pathlib import Path | |
| path = app.ROOT / "outputs" / "test-auth-scan-prefer.png" | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82") | |
| stale_scan = {"status": "review", "export_gate": "blocked"} | |
| state = { | |
| "generation": {"output_path": str(path)}, | |
| "generated_scan": stale_scan, | |
| } | |
| result = app._authoritative_generated_scan(state) | |
| # Should re-scan from file, not use stored scan | |
| assert result["status"] == "pass" | |
| assert result["export_gate"] == "clear" | |
| def test_export_packet_allows_blocked_st3gg_with_explicit_override(monkeypatch) -> None: | |
| monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports") | |
| run = build_command_center_run("gothic patent leather platform boots") | |
| blocked_scan = { | |
| "status": "review", | |
| "export_gate": "blocked", | |
| "findings": ["metadata review"], | |
| "purification_actions": ["strip metadata"], | |
| } | |
| state = { | |
| **app._default_operator_state(), | |
| "checkpoint": "approved", | |
| "provider_state": "generated", | |
| "generated_scan": blocked_scan, | |
| "generation": { | |
| "status": "success", | |
| "provider_state": "generated", | |
| "output_path": "outputs/runtime/nexus_flux_test.png", | |
| "lora_status": "skipped_incompatible", | |
| }, | |
| } | |
| blocked = app.export_packet(run, False, blocked_scan, "Forge", state, "") | |
| assert blocked[13]["provider_state"] == "blocked" | |
| assert "override reason" in blocked[13]["message"].lower() | |
| exported = app.export_packet(run, False, blocked_scan, "Forge", state, "Operator reviewed ST3GG and wrote audit evidence only.") | |
| assert exported[13]["provider_state"] == "exported" | |
| assert exported[13]["export"] == "override" | |
| assert "export_packet" in exported[13] | |