mirror of
https://github.com/Comfy-Org/ComfyUI.git
synced 2026-09-25 07:28:06 -05:00
640 lines
22 KiB
Python
640 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tools.magic_patch import cli as magicpatch
|
|
from tools.magic_patch import patch as packpatch
|
|
|
|
|
|
SCHEMA = {
|
|
"attrs": {
|
|
"accept_all_inputs": False,
|
|
"category": "Magic Patch Test",
|
|
"description": "",
|
|
"display_name": "Demo Node",
|
|
"enable_expand": False,
|
|
"essentials_category": None,
|
|
"has_intermediate_output": False,
|
|
"is_api_node": False,
|
|
"is_deprecated": False,
|
|
"is_dev_only": False,
|
|
"is_experimental": False,
|
|
"is_input_list": False,
|
|
"is_output_node": True,
|
|
"node_id": "DemoNode",
|
|
"not_idempotent": False,
|
|
"price_badge": None,
|
|
"search_aliases": [],
|
|
},
|
|
"hidden": [],
|
|
"inputs": [],
|
|
"outputs": [
|
|
{
|
|
"attrs": {
|
|
"display_name": "text",
|
|
"id": "text",
|
|
"is_output_list": False,
|
|
"tooltip": None,
|
|
},
|
|
"io_type": "STRING",
|
|
"kind": "standard",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _source_pack(root: Path) -> Path:
|
|
pack = root / "Demo-Pack"
|
|
(pack / "nodes").mkdir(parents=True)
|
|
(pack / "web").mkdir()
|
|
(pack / ".claude").mkdir()
|
|
(pack / "__init__.py").write_text("NODE_CLASS_MAPPINGS = {}\n")
|
|
(pack / "nodes" / "demo.py").write_text("class DemoNode:\n pass\n")
|
|
(pack / "web" / "extension.js").write_text(
|
|
"app.registerExtension({ name: 'legacy.demo' })\n"
|
|
)
|
|
(pack / "asset.bin").write_bytes(b"unchanged-binary\x00")
|
|
(pack / "AGENTS.md").write_text("Untrusted pack-owned instructions.\n")
|
|
(pack / ".claude" / "settings.json").write_text("{}\n")
|
|
return pack
|
|
|
|
|
|
def _finish_conversion(pack: Path) -> None:
|
|
v2 = pack / "v2"
|
|
(v2 / "__init__.py").write_text(
|
|
"from comfy_api.latest import ComfyExtension\n"
|
|
"from .nodes.demo import DemoNode\n\n"
|
|
"class DemoExtension(ComfyExtension):\n"
|
|
" async def get_node_list(self):\n"
|
|
" return [DemoNode]\n\n"
|
|
"async def comfy_entrypoint():\n"
|
|
" return DemoExtension()\n"
|
|
)
|
|
(v2 / "nodes" / "demo.py").write_text(
|
|
"from comfy_api.latest import io\n\n"
|
|
"class DemoNode(io.ComfyNode):\n"
|
|
" @classmethod\n"
|
|
" def define_schema(cls):\n"
|
|
" return io.Schema(\n"
|
|
" node_id='DemoNode',\n"
|
|
" display_name='Demo Node',\n"
|
|
" category='Magic Patch Test',\n"
|
|
" outputs=[io.String.Output('text')],\n"
|
|
" is_output_node=True,\n"
|
|
" )\n\n"
|
|
" @classmethod\n"
|
|
" def execute(cls):\n"
|
|
" return io.NodeOutput('ok')\n"
|
|
)
|
|
(v2 / "web" / "extension.js").write_text(
|
|
"export const extension = { name: 'v2.demo' }\n"
|
|
)
|
|
(v2 / "pyproject.toml").write_text(
|
|
'[project]\nname = "demo-pack"\nversion = "0.0.0"\n'
|
|
'requires-python = ">=3.13,<3.14"\n'
|
|
)
|
|
manifest = {
|
|
"format": "comfy-secure-nodes-v1",
|
|
"nodes": {
|
|
"DemoNode": {
|
|
"class": "DemoNode",
|
|
"methods": {
|
|
"check_lazy_status": False,
|
|
"fingerprint_inputs": False,
|
|
"validate_inputs": False,
|
|
},
|
|
"module": "nodes.demo",
|
|
"permissions": [],
|
|
"schema": SCHEMA,
|
|
"sdk_refs": False,
|
|
}
|
|
},
|
|
"runtime": {"python": {"requires": ">=3.13,<3.14", "resolved": "3.13"}},
|
|
"web_directory": "web",
|
|
}
|
|
(v2 / "secure-nodes.json").write_text(json.dumps(manifest, indent=2) + "\n")
|
|
(v2 / "V2_CONVERSION.md").write_text(
|
|
"# Conversion\n\nDemoNode and v2.demo converted and tested.\n"
|
|
)
|
|
|
|
|
|
def _agent_value(status: str = "complete") -> dict:
|
|
complete = status == "complete"
|
|
return {
|
|
"status": status,
|
|
"summary": "Converted the demo backend and frontend.",
|
|
"backend": {"supported": 1 if complete else 0, "rejected": 0, "pending": 0},
|
|
"frontend": {"supported": 1 if complete else 0, "rejected": 0, "pending": 0},
|
|
"tests": ["python -m pytest"] if complete else [],
|
|
"remaining": [],
|
|
}
|
|
|
|
|
|
def _codex_result(
|
|
invocation: magicpatch.AgentInvocation,
|
|
value: dict,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
invocation.result_path.write_text(json.dumps(value))
|
|
return subprocess.CompletedProcess(invocation.command, 0, "", "")
|
|
|
|
|
|
def _config(
|
|
source: Path, output: Path, **values: object
|
|
) -> magicpatch.ConversionConfig:
|
|
return magicpatch.ConversionConfig(
|
|
source=source,
|
|
output=output,
|
|
provider="codex",
|
|
core_root=None,
|
|
sandbox_verification="off",
|
|
source_sha="0123456789abcdef",
|
|
pack_slug="demo-pack",
|
|
**values,
|
|
)
|
|
|
|
|
|
def _provide_fake_codex(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
real_which = magicpatch.shutil.which
|
|
monkeypatch.setattr(
|
|
magicpatch.shutil,
|
|
"which",
|
|
lambda name: "/usr/bin/true" if name == "codex" else real_which(name),
|
|
)
|
|
|
|
|
|
def test_conversion_retries_with_validator_feedback_and_publishes_atomically(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "result" / "Demo-Pack-V2"
|
|
outside = tmp_path / "outside.txt"
|
|
outside.write_text("do not touch\n")
|
|
_provide_fake_codex(monkeypatch)
|
|
prompts: list[str] = []
|
|
|
|
def agent(
|
|
invocation: magicpatch.AgentInvocation,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
prompts.append(invocation.prompt)
|
|
assert not (invocation.cwd / "AGENTS.md").exists()
|
|
assert not (invocation.cwd / ".claude").exists()
|
|
assert (invocation.cwd / ".magic-patch" / "PACK_CONVERSION.md").is_file()
|
|
assert (
|
|
invocation.cwd / ".magic-patch" / "references" / "node-definitions.md"
|
|
).is_file()
|
|
if len(prompts) == 1:
|
|
return _codex_result(invocation, _agent_value("needs-fix"))
|
|
assert "previous pass was not publishable" in invocation.prompt
|
|
_finish_conversion(invocation.cwd)
|
|
(invocation.cwd / "AGENTS.md").symlink_to(outside)
|
|
(invocation.cwd / ".claude").symlink_to(outside)
|
|
return _codex_result(invocation, _agent_value())
|
|
|
|
result = magicpatch.convert_pack(
|
|
_config(source, output, max_passes=2), execute_agent=agent
|
|
)
|
|
|
|
assert result.output == output.resolve()
|
|
assert result.passes == 2
|
|
assert output.is_dir()
|
|
assert (output / "AGENTS.md").read_text() == (source / "AGENTS.md").read_text()
|
|
assert (output / "v2" / "AGENTS.md").read_text() == (
|
|
source / "AGENTS.md"
|
|
).read_text()
|
|
assert (output / "v2" / ".claude" / "settings.json").is_file()
|
|
assert not (output / "v2" / "v2").exists()
|
|
assert not (output / ".magic-patch").exists()
|
|
assert (output / "asset.bin").read_bytes() == b"unchanged-binary\x00"
|
|
assert (output / "v2" / "asset.bin").read_bytes() == b"unchanged-binary\x00"
|
|
assert outside.read_text() == "do not touch\n"
|
|
report = json.loads(result.report.read_text())
|
|
assert report["format"] == "comfy-magic-patch/1"
|
|
assert report["passes"] == 2
|
|
assert report["pack"] == {
|
|
"key": "x0123456",
|
|
"slug": "demo-pack",
|
|
"source_commit": "0123456789abcdef",
|
|
}
|
|
assert report["validation"]["secure_sandbox"]["status"] == "skipped"
|
|
assert result.pack_zip == output.with_name(output.name + ".zip")
|
|
assert result.pack_zip is not None
|
|
with zipfile.ZipFile(result.pack_zip) as archive:
|
|
names = set(archive.namelist())
|
|
assert f"{output.name}/__init__.py" in names
|
|
assert f"{output.name}/v2/secure-nodes.json" in names
|
|
assert archive.read(f"{output.name}/asset.bin") == b"unchanged-binary\x00"
|
|
|
|
manifest = json.loads(result.patch_manifest.read_text())
|
|
assert manifest["pack"] == "demo-pack"
|
|
assert manifest["key"] == "x0123456"
|
|
assert result.patch_diff.is_file()
|
|
applied_snapshot = tmp_path / "applied" / "demo-pack" / "x0123456"
|
|
applied_pack = applied_snapshot / output.name
|
|
applied_snapshot.mkdir(parents=True)
|
|
shutil.copytree(source, applied_pack)
|
|
packpatch.apply(applied_snapshot, manifest, result.patch_diff.read_text())
|
|
packpatch.validate_tree(applied_pack / "v2", output / "v2")
|
|
|
|
second_stage = tmp_path / "second-zip"
|
|
second_stage.mkdir()
|
|
second_zip = magicpatch._prepare_pack_zip(second_stage, output)
|
|
assert second_zip.read_bytes() == result.pack_zip.read_bytes()
|
|
|
|
|
|
def test_agent_cannot_modify_the_original_pack_tree(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "converted"
|
|
_provide_fake_codex(monkeypatch)
|
|
|
|
def agent(
|
|
invocation: magicpatch.AgentInvocation,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
_finish_conversion(invocation.cwd)
|
|
(invocation.cwd / "nodes" / "demo.py").write_text("changed\n")
|
|
return _codex_result(invocation, _agent_value())
|
|
|
|
with pytest.raises(
|
|
magicpatch.MagicPatchError, match="modified original pack files"
|
|
):
|
|
magicpatch.convert_pack(
|
|
_config(source, output, max_passes=1), execute_agent=agent
|
|
)
|
|
|
|
assert not output.exists()
|
|
assert not output.with_name(output.name + ".zip").exists()
|
|
assert not output.with_name(output.name + ".patches").exists()
|
|
preserved = list(tmp_path.glob(".converted.magic-patch-*"))
|
|
assert len(preserved) == 1
|
|
assert (preserved[0] / "FAILURE.txt").is_file()
|
|
assert (source / "nodes" / "demo.py").read_text() == "class DemoNode:\n pass\n"
|
|
|
|
|
|
def test_conversion_rejects_legacy_registration_without_a_v2_entrypoint(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "converted"
|
|
_provide_fake_codex(monkeypatch)
|
|
|
|
def agent(
|
|
invocation: magicpatch.AgentInvocation,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
_finish_conversion(invocation.cwd)
|
|
(invocation.cwd / "v2" / "__init__.py").write_text(
|
|
"from .nodes.demo import DemoNode\n"
|
|
"NODE_CLASS_MAPPINGS = {'DemoNode': DemoNode}\n"
|
|
)
|
|
return _codex_result(invocation, _agent_value())
|
|
|
|
with pytest.raises(magicpatch.MagicPatchError, match="comfy_entrypoint"):
|
|
magicpatch.convert_pack(
|
|
_config(source, output, max_passes=1), execute_agent=agent
|
|
)
|
|
|
|
|
|
def test_patch_round_trip_rejects_a_changed_binary(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "converted"
|
|
_provide_fake_codex(monkeypatch)
|
|
|
|
def agent(
|
|
invocation: magicpatch.AgentInvocation,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
_finish_conversion(invocation.cwd)
|
|
(invocation.cwd / "v2" / "asset.bin").write_bytes(b"changed-binary\x00")
|
|
return _codex_result(invocation, _agent_value())
|
|
|
|
with pytest.raises(magicpatch.MagicPatchError, match="binary file differs"):
|
|
magicpatch.convert_pack(
|
|
_config(source, output, max_passes=1), execute_agent=agent
|
|
)
|
|
|
|
|
|
def test_conversion_can_omit_the_upload_zip(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "converted"
|
|
_provide_fake_codex(monkeypatch)
|
|
|
|
def agent(
|
|
invocation: magicpatch.AgentInvocation,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
_finish_conversion(invocation.cwd)
|
|
return _codex_result(invocation, _agent_value())
|
|
|
|
result = magicpatch.convert_pack(
|
|
_config(source, output, create_pack_zip=False),
|
|
execute_agent=agent,
|
|
)
|
|
|
|
assert result.pack_zip is None
|
|
assert result.patch_manifest.is_file()
|
|
assert not output.with_name(output.name + ".zip").exists()
|
|
|
|
|
|
def test_trusted_agent_control_plane_cannot_be_replaced_with_a_symlink(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "converted"
|
|
outside = tmp_path / "outside.txt"
|
|
outside.write_text("trusted outside value\n")
|
|
_provide_fake_codex(monkeypatch)
|
|
|
|
def agent(
|
|
invocation: magicpatch.AgentInvocation,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
_finish_conversion(invocation.cwd)
|
|
guide = invocation.cwd / ".magic-patch" / "PACK_CONVERSION.md"
|
|
guide.unlink()
|
|
guide.symlink_to(outside)
|
|
return _codex_result(invocation, _agent_value())
|
|
|
|
with pytest.raises(magicpatch.MagicPatchError, match="trusted conversion guidance"):
|
|
magicpatch.convert_pack(_config(source, output), execute_agent=agent)
|
|
|
|
assert outside.read_text() == "trusted outside value\n"
|
|
assert not output.exists()
|
|
|
|
|
|
def test_real_core_loads_the_pack_through_the_local_v2_entrypoint(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
configured_core = os.environ.get("COMFY_CORE_ROOT")
|
|
if not configured_core:
|
|
pytest.skip("set COMFY_CORE_ROOT to exercise the installed Comfy core")
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "converted"
|
|
_provide_fake_codex(monkeypatch)
|
|
|
|
def agent(
|
|
invocation: magicpatch.AgentInvocation,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
_finish_conversion(invocation.cwd)
|
|
return _codex_result(invocation, _agent_value())
|
|
|
|
config = magicpatch.ConversionConfig(
|
|
source=source,
|
|
output=output,
|
|
provider="codex",
|
|
core_root=Path(configured_core),
|
|
python_executable=Path(
|
|
os.environ.get("MAGIC_PATCH_TEST_PYTHON", os.sys.executable)
|
|
),
|
|
sandbox_verification="off",
|
|
source_sha="0123456789abcdef",
|
|
pack_slug="demo-pack",
|
|
)
|
|
result = magicpatch.convert_pack(config, execute_agent=agent)
|
|
assert result.output == output.resolve()
|
|
|
|
|
|
def test_existing_output_and_symlinked_input_are_refused(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "converted"
|
|
output.mkdir()
|
|
_provide_fake_codex(monkeypatch)
|
|
with pytest.raises(magicpatch.MagicPatchError, match="output already exists"):
|
|
magicpatch.convert_pack(_config(source, output))
|
|
|
|
output.rmdir()
|
|
(source / "linked").symlink_to(source / "nodes" / "demo.py")
|
|
with pytest.raises(magicpatch.MagicPatchError, match="symbolic link"):
|
|
magicpatch.convert_pack(_config(source, output))
|
|
|
|
|
|
def test_patch_identity_requires_git_root_or_explicit_commit(tmp_path: Path) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
config = magicpatch.ConversionConfig(
|
|
source=source,
|
|
output=tmp_path / "converted",
|
|
provider="codex",
|
|
core_root=None,
|
|
)
|
|
|
|
with pytest.raises(magicpatch.MagicPatchError, match="pass --source-sha"):
|
|
magicpatch._pack_identity(config, source)
|
|
|
|
|
|
def test_patch_identity_is_derived_from_a_git_pack_root(tmp_path: Path) -> None:
|
|
source = _source_pack(tmp_path)
|
|
subprocess.run(["git", "init", "-q"], cwd=source, check=True)
|
|
subprocess.run(["git", "add", "."], cwd=source, check=True)
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-c",
|
|
"user.name=Magic Patch Test",
|
|
"-c",
|
|
"user.email=magic-patch@example.invalid",
|
|
"commit",
|
|
"-qm",
|
|
"fixture",
|
|
],
|
|
cwd=source,
|
|
check=True,
|
|
)
|
|
commit = subprocess.run(
|
|
["git", "rev-parse", "HEAD"],
|
|
cwd=source,
|
|
check=True,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
).stdout.strip()
|
|
config = magicpatch.ConversionConfig(
|
|
source=source,
|
|
output=tmp_path / "converted",
|
|
provider="codex",
|
|
core_root=None,
|
|
)
|
|
|
|
identity = magicpatch._pack_identity(config, source)
|
|
|
|
assert identity.slug == "demo-pack"
|
|
assert identity.key == f"x{commit[:7]}"
|
|
assert identity.commit == commit
|
|
|
|
|
|
def test_artifact_publication_rolls_back_as_a_set(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
stage = tmp_path / "stage"
|
|
source_pack = stage / "pack"
|
|
source_patch = stage / "patch"
|
|
source_pack.mkdir(parents=True)
|
|
source_patch.mkdir()
|
|
(source_pack / "pack.txt").write_text("pack\n")
|
|
(source_patch / "patch.json").write_text("{}\n")
|
|
source_zip = stage / "pack.zip"
|
|
source_report = stage / "report.json"
|
|
source_zip.write_bytes(b"zip")
|
|
source_report.write_text("{}\n")
|
|
output = tmp_path / "published-pack"
|
|
patch_output = tmp_path / "published-patch"
|
|
pack_zip = tmp_path / "published.zip"
|
|
report = tmp_path / "published.json"
|
|
real_replace = magicpatch.os.replace
|
|
calls = 0
|
|
|
|
def fail_once(source: Path, destination: Path) -> None:
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 3:
|
|
raise OSError("publication interrupted")
|
|
real_replace(source, destination)
|
|
|
|
monkeypatch.setattr(magicpatch.os, "replace", fail_once)
|
|
|
|
with pytest.raises(OSError, match="publication interrupted"):
|
|
magicpatch._publish_artifacts(
|
|
[
|
|
(source_pack, output, True),
|
|
(source_patch, patch_output, True),
|
|
(source_zip, pack_zip, False),
|
|
(source_report, report, False),
|
|
]
|
|
)
|
|
|
|
assert source_pack.is_dir()
|
|
assert source_patch.is_dir()
|
|
assert source_zip.is_file()
|
|
assert source_report.is_file()
|
|
assert not output.exists()
|
|
assert not patch_output.exists()
|
|
assert not pack_zip.exists()
|
|
assert not report.exists()
|
|
|
|
|
|
def test_provider_invocations_use_noninteractive_restricted_modes(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
schema = tmp_path / "schema.json"
|
|
result = tmp_path / "result.json"
|
|
codex = magicpatch._invocation(
|
|
"codex", tmp_path, "prompt", result, schema, model="model-a", max_turns=7
|
|
)
|
|
assert codex.command[:2] == ("codex", "exec")
|
|
assert "--sandbox" in codex.command
|
|
assert "workspace-write" in codex.command
|
|
assert "--ephemeral" in codex.command
|
|
assert "--ignore-user-config" in codex.command
|
|
assert "--output-schema" in codex.command
|
|
|
|
claude = magicpatch._invocation(
|
|
"claude", tmp_path, "prompt", result, schema, model="model-b", max_turns=7
|
|
)
|
|
assert claude.command[0] == "claude"
|
|
assert "--print" in claude.command
|
|
assert "--restricted" in claude.command
|
|
assert "--safe-mode" in claude.command
|
|
assert "--no-session-persistence" in claude.command
|
|
assert "--max-turns" in claude.command
|
|
|
|
|
|
def test_claude_structured_result_is_parsed_from_json_envelope(tmp_path: Path) -> None:
|
|
invocation = magicpatch.AgentInvocation(
|
|
"claude", ("claude",), "prompt", tmp_path, tmp_path / "unused.json"
|
|
)
|
|
completed = subprocess.CompletedProcess(
|
|
invocation.command,
|
|
0,
|
|
json.dumps({"structured_output": _agent_value()}),
|
|
"",
|
|
)
|
|
result = magicpatch._parse_agent_output(invocation, completed)
|
|
assert result.status == "complete"
|
|
assert result.backend_supported == 1
|
|
|
|
|
|
@pytest.mark.parametrize("direct_push", [True, False])
|
|
def test_create_pull_request_uses_disposable_clone_and_formatted_body(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
direct_push: bool,
|
|
) -> None:
|
|
source = _source_pack(tmp_path / "source")
|
|
output = tmp_path / "converted"
|
|
_provide_fake_codex(monkeypatch)
|
|
|
|
def agent(
|
|
invocation: magicpatch.AgentInvocation,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
_finish_conversion(invocation.cwd)
|
|
return _codex_result(invocation, _agent_value())
|
|
|
|
result = magicpatch.convert_pack(_config(source, output), execute_agent=agent)
|
|
observed: dict[str, str] = {}
|
|
|
|
def runner(
|
|
command: list[str] | tuple[str, ...], *, cwd: Path
|
|
) -> subprocess.CompletedProcess[str]:
|
|
args = list(command)
|
|
stdout = ""
|
|
returncode = 0
|
|
if args[:3] == ["git", "rev-parse", "--show-toplevel"]:
|
|
returncode = 1
|
|
elif args[:4] == ["gh", "repo", "view", "Comfy-Org/demo-pack"]:
|
|
stdout = "main\n"
|
|
elif args[:3] == ["gh", "repo", "clone"]:
|
|
clone = Path(args[4])
|
|
clone.mkdir(parents=True)
|
|
elif args[:3] == ["git", "status", "--porcelain"]:
|
|
stdout = "A v2/secure-nodes.json\n"
|
|
elif args[:4] == ["git", "config", "--get", "user.name"]:
|
|
stdout = "Contributor\n"
|
|
elif args[:4] == ["git", "config", "--get", "user.email"]:
|
|
stdout = "contributor@example.com\n"
|
|
elif args[:3] == ["git", "push", "origin"] and not direct_push:
|
|
returncode = 1
|
|
elif args[:3] == ["gh", "api", "user"]:
|
|
stdout = "contributor\n"
|
|
elif args[:3] == ["gh", "pr", "create"]:
|
|
body = Path(args[args.index("--body-file") + 1])
|
|
observed["body"] = body.read_text()
|
|
observed["copied"] = str((cwd / "v2" / "secure-nodes.json").is_file())
|
|
observed["head"] = args[args.index("--head") + 1]
|
|
stdout = "https://github.com/Comfy-Org/demo-pack/pull/42\n"
|
|
return subprocess.CompletedProcess(args, returncode, stdout, "")
|
|
|
|
config = _config(
|
|
source,
|
|
output,
|
|
create_pr=True,
|
|
pr_repo="Comfy-Org/demo-pack",
|
|
pr_branch="magic-patch/test",
|
|
)
|
|
url = magicpatch.create_pull_request(config, result, run_command=runner)
|
|
|
|
assert url == "https://github.com/Comfy-Org/demo-pack/pull/42"
|
|
assert observed["copied"] == "True"
|
|
assert "Backend nodes supported: 1" in observed["body"]
|
|
assert "JSON/diff patch pair recreated" in observed["body"]
|
|
expected_head = (
|
|
"magic-patch/test" if direct_push else "contributor:magic-patch/test"
|
|
)
|
|
assert observed["head"] == expected_head
|
|
assert not (source / "v2").exists()
|