| import app |
| from nexus_visual_weaver.planner import build_command_center_run |
|
|
|
|
| |
|
|
| def test_checkpoint_seed_parses_valid_hex_suffix() -> None: |
| |
| 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: |
| |
| result = app._checkpoint_seed("nw-zzzzzzzz") |
| assert result == 0 |
|
|
|
|
| def test_checkpoint_seed_ignores_non_hex_chars_in_last_8() -> None: |
| |
| |
| 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: |
| |
| |
| result = app._checkpoint_seed("nw-aabbccddee11223344") |
| assert result == int("11223344", 16) % 1_000_000 |
|
|
|
|
| |
|
|
| 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) |
|
|
| |
| assert isinstance(summary, str) |
| assert len(summary) > 0 |
| |
| 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: |
| |
| 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) |
|
|
| |
| 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() |
|
|
|
|
| |
|
|
| 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) |
|
|
|
|
| |
|
|
| 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"] |
|
|
|
|
| |
|
|
| 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) |
|
|
|
|
| |
|
|
| 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" |
|
|
|
|
| |
|
|
| 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: |
| |
| 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 |
|
|
|
|
| |
|
|
| 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 |
|
|
|
|
| |
|
|
| 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) |
|
|
| |
| 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] |
|
|