refactor: add DesignIR distillation pipeline

This commit is contained in:
Jerry
2026-07-27 17:24:04 +08:00
parent 0adaac2b2a
commit 550b339bcc
1128 changed files with 10248 additions and 426666 deletions
@@ -1,982 +0,0 @@
#!/usr/bin/env python3
from __future__ import annotations
import json
import hashlib
import os
import stat
import tempfile
import textwrap
import unittest
import zipfile
from pathlib import Path
from unittest import mock
from tests.python.support.paths import add_repo_path, repo_path
add_repo_path("skills/bambu-labs/scripts")
import bambu_lan_print as bambu
def write_gcode(path: Path, body: str = "M104 S220\nM140 S65\nG1 X1 Y1 Z0.2 E0.1\n") -> None:
path.write_text(body, encoding="utf-8")
def write_sliced_3mf(path: Path) -> None:
with zipfile.ZipFile(path, "w") as archive:
archive.writestr("[Content_Types].xml", "<Types/>")
archive.writestr("Metadata/plate_1.gcode", "; plate 1\nG1 X1\n")
def write_fake_bambox(path: Path) -> None:
script = textwrap.dedent(
"""\
#!/usr/bin/env python3
import json
import sys
import zipfile
from pathlib import Path
args = sys.argv[1:]
if args[0] == "pack":
output = Path(args[args.index("-o") + 1])
with zipfile.ZipFile(output, "w") as archive:
archive.writestr("[Content_Types].xml", "<Types/>")
archive.writestr("Metadata/plate_1.gcode", "G1 X1 E0.1\\n")
print(json.dumps({"packed": str(output)}))
raise SystemExit(0)
if args[0] == "validate":
print(json.dumps({"ok": True}))
raise SystemExit(0)
raise SystemExit(2)
"""
)
path.write_text(script, encoding="utf-8")
path.chmod(path.stat().st_mode | stat.S_IXUSR)
class BambuLanPrintTests(unittest.TestCase):
def test_new_printer_onboarding_reference_is_linked_and_specific(self) -> None:
skill_root = repo_path("skills/bambu-labs")
skill_text = (skill_root / "SKILL.md").read_text(encoding="utf-8")
onboarding_path = skill_root / "references" / "new-printer-onboarding.md"
onboarding = onboarding_path.read_text(encoding="utf-8")
self.assertIn("references/new-printer-onboarding.md", skill_text)
for required in [
"Enable LAN Only",
"Enable Developer Mode",
"A1 / A1 Mini",
"P1P / P1S",
"X1 / X1C / X1E",
"H2D / Newer Bambu Printers",
"printer IP address and LAN access code",
]:
with self.subTest(required=required):
self.assertIn(required, onboarding)
def test_explicit_print_request_policy_is_documented(self) -> None:
skill_root = repo_path("skills/bambu-labs")
skill_text = (skill_root / "SKILL.md").read_text(encoding="utf-8")
checklist = (skill_root / "references" / "real-printer-checklist.md").read_text(encoding="utf-8")
self.assertIn("explicit user request to print or start", skill_text)
self.assertIn("do not pause for a second confirmation", skill_text)
self.assertIn("live-start authorization for this checklist", checklist)
def test_default_config_uses_workspace_root_not_skill_root(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
workspace_root = Path(tmp) / "workspace"
workspace_root.mkdir()
with mock.patch.dict(os.environ, {"INIT_CWD": str(workspace_root)}):
self.assertEqual(workspace_root.resolve(), bambu.default_workspace_root())
skill_root = repo_path("skills/bambu-labs")
self.assertNotEqual(skill_root.resolve(), bambu.DEFAULT_CONFIG_PATH.parent)
def test_inspects_plain_gcode(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
job = Path(tmp) / "job.gcode"
write_gcode(job)
inspection = bambu.inspect_gcode_file(job)
self.assertEqual(inspection.size_bytes, job.stat().st_size)
self.assertRegex(inspection.md5, r"^[0-9a-f]{32}$")
self.assertEqual(inspection.path, str(job))
def test_rejects_non_plain_gcode_inputs(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
missing = root / "missing.gcode"
empty = root / "empty.gcode"
empty.write_text("", encoding="utf-8")
sliced = root / "job.gcode.3mf"
write_sliced_3mf(sliced)
mesh = root / "part.stl"
mesh.write_text("solid part\nendsolid part\n", encoding="utf-8")
cases = [
(missing, "does not exist"),
(empty, "empty"),
(sliced, "Expected plain .gcode"),
(mesh, "Expected a plain .gcode"),
]
for path, message in cases:
with self.subTest(path=path):
with self.assertRaisesRegex(bambu.BambuPrintError, message):
bambu.inspect_gcode_file(path)
def test_plain_send_plan_uses_gcode_file(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
job = Path(tmp) / "job 01.gcode"
write_gcode(job)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--action",
"upload-start",
"--sequence-id",
"seq-1",
"--remote-name",
"job 01.gcode",
]
)
plan = bambu.build_send_plan(args)
self.assertTrue(plan["dry_run"])
self.assertEqual(plan["handoff"], "plain")
self.assertTrue(plan["ftps"]["will_upload"])
self.assertTrue(plan["mqtt"]["will_publish"])
self.assertEqual(plan["ftps"]["remote_path"], "cache/job_01.gcode")
self.assertEqual(plan["mqtt"]["topic"], "device/00M00A000000000/request")
payload = plan["mqtt"]["payload"]["print"]
self.assertEqual(payload["command"], "gcode_file")
self.assertEqual(payload["param"], "cache/job_01.gcode")
self.assertEqual(payload["sequence_id"], "seq-1")
def test_gcode_param_override_is_allowed_for_plain_handoff(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
job = Path(tmp) / "job.gcode"
write_gcode(job)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--gcode-param",
"/cache/job.gcode",
]
)
plan = bambu.build_send_plan(args)
self.assertEqual(plan["mqtt"]["payload"]["print"]["param"], "/cache/job.gcode")
def test_config_set_writes_printer_json_without_echoing_access_code(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
config = root / "printers.json"
args = bambu.parse_args(
[
"config",
"--config",
str(config),
"set",
"--printer",
"a1-mini",
"--host",
"192.168.1.34",
"--access-code",
"12345678",
"--model",
"a1-mini",
"--serial",
"0309CA4C0901107",
]
)
with mock.patch("builtins.print") as mocked_print:
code = bambu.config_set_main(args)
self.assertEqual(code, 0)
data = json.loads(config.read_text(encoding="utf-8"))
self.assertEqual(data["printers"]["a1-mini"]["access_code"], "12345678")
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["entry"]["host"], "192.168.1.34")
self.assertNotIn("access_code", payload["entry"])
def test_send_parser_loads_printer_defaults_from_json_config(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
job = root / "job.gcode"
config = root / "printers.json"
write_gcode(job)
config.write_text(
json.dumps(
{
"printers": {
"a1-mini": {
"host": "192.168.1.34",
"access_code": "12345678",
"serial": "0309CA4C0901107",
"model": "a1-mini",
}
}
}
),
encoding="utf-8",
)
args = bambu.parse_args(["send", "--config", str(config), "--printer", "a1-mini", "--gcode", str(job)])
plan = bambu.build_send_plan(args)
self.assertEqual(args.host, "192.168.1.34")
self.assertEqual(args.serial, "0309CA4C0901107")
self.assertEqual(bambu.access_code_from_args(args), "12345678")
self.assertEqual(plan["mqtt"]["topic"], "device/0309CA4C0901107/request")
def test_dry_run_send_plan_does_not_discover_serial_when_omitted(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
job = root / "job.gcode"
config = root / "printers.json"
write_gcode(job)
config.write_text(
json.dumps({"printers": {"a1-mini": {"host": "192.168.1.34", "access_code": "12345678"}}}),
encoding="utf-8",
)
args = bambu.parse_args(
[
"send",
"--config",
str(config),
"--printer",
"a1-mini",
"--gcode",
str(job),
"--action",
"upload-start",
]
)
with mock.patch.object(
bambu,
"discover_printer_serial",
side_effect=AssertionError("dry-run plans must not perform network discovery"),
) as mocked_discover:
plan = bambu.build_send_plan(args)
mocked_discover.assert_not_called()
self.assertEqual(args.serial, "")
self.assertIsNone(plan["mqtt"]["topic"])
cached = json.loads(config.read_text(encoding="utf-8"))
self.assertNotIn("serial", cached["printers"]["a1-mini"])
def test_execute_send_discovers_serial_from_printer_certificate_when_omitted(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
job = root / "job.gcode"
config = root / "printers.json"
write_gcode(job)
config.write_text(
json.dumps({"printers": {"a1-mini": {"host": "192.168.1.34", "access_code": "12345678"}}}),
encoding="utf-8",
)
args = bambu.parse_args(
[
"send",
"--config",
str(config),
"--printer",
"a1-mini",
"--gcode",
str(job),
"--action",
"upload-start",
"--execute",
"--confirm-start-print",
]
)
with (
mock.patch.object(bambu, "discover_printer_serial", return_value="0309CA4C0901107") as mocked_discover,
mock.patch.object(bambu, "upload_ftps") as mocked_upload,
mock.patch.object(bambu, "publish_mqtt") as mocked_publish,
mock.patch("builtins.print") as mocked_print,
):
code = bambu.send_main(args)
self.assertEqual(code, 0)
mocked_discover.assert_called_once()
mocked_upload.assert_called_once()
mocked_publish.assert_called_once()
self.assertEqual(args.serial, "0309CA4C0901107")
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["mqtt"]["topic"], "device/0309CA4C0901107/request")
cached = json.loads(config.read_text(encoding="utf-8"))
self.assertEqual(cached["printers"]["a1-mini"]["serial"], "0309CA4C0901107")
def test_serial_command_reports_tls_certificate_serial(self) -> None:
args = bambu.parse_args(["serial", "--host", "192.168.1.34", "--json"])
with mock.patch.object(bambu, "discover_printer_serial", return_value="0309CA4C0901107"), mock.patch(
"builtins.print"
) as mocked_print:
code = bambu.serial_main(args)
self.assertEqual(code, 0)
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["serial"], "0309CA4C0901107")
self.assertEqual(payload["source"], "printer_tls_certificate_common_name")
def test_bambox_project_refuses_unsupported_a1_mini_profile(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
job = Path(tmp) / "job.gcode"
write_gcode(job)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--handoff",
"bambox-project",
"--bambox-profile",
"a1-mini-0.4",
"--filament",
"PLA",
]
)
with self.assertRaisesRegex(bambu.BambuPrintError, "a1-mini-0.4 is not enabled"):
bambu.build_send_plan(args)
def test_bambox_project_builds_pack_command_without_running_bambox(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
job = Path(tmp) / "job.gcode"
write_gcode(job)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--handoff",
"bambox-project",
"--bambox-profile",
"p1s-0.4",
"--filament",
"PLA",
"--bambox-bin",
"/usr/local/bin/bambox",
"--action",
"upload-start",
]
)
plan = bambu.build_send_plan(args)
command = plan["input"]["bambox"]["pack_command"]
self.assertEqual(command[:3], ["/usr/local/bin/bambox", "pack", str(job.resolve())])
self.assertIn("-m", command)
self.assertIn("p1s", command)
self.assertIn("--nozzle-diameter", command)
self.assertIn("0.4", command)
self.assertIn("-f", command)
self.assertIn("PLA", command)
self.assertEqual(plan["mqtt"]["payload"]["print"]["command"], "project_file")
self.assertIsNone(plan["mqtt"]["payload"]["print"]["md5"])
self.assertEqual(plan["mqtt"]["payload"]["print"]["subtask_name"], "job")
self.assertTrue(plan["mqtt"]["payload"]["print"]["bed_levelling"])
self.assertFalse(plan["mqtt"]["payload"]["print"]["use_ams"])
self.assertEqual(plan["ftps"]["remote_path"], "job.gcode.3mf")
self.assertEqual(plan["ftps"]["url"], "ftp:///job.gcode.3mf")
def test_package_execute_uses_fake_bambox_and_validates_archive(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
job = root / "job.gcode"
output = root / "job.gcode.3mf"
fake_bambox = root / "bambox"
write_gcode(job)
write_fake_bambox(fake_bambox)
args = bambu.parse_args(
[
"package",
"--gcode",
str(job),
"--output",
str(output),
"--bambox-profile",
"p1s-0.4",
"--filament",
"PLA",
"--bambox-bin",
str(fake_bambox),
"--execute",
]
)
with mock.patch("builtins.print") as mocked_print:
code = bambu.package_main(args)
self.assertEqual(code, 0)
self.assertTrue(output.exists())
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["executed"], ["bambox_pack", "bambox_validate"])
self.assertEqual(payload["result"]["project"]["plates"][0]["path"], "Metadata/plate_1.gcode")
def test_template_project_pack_replaces_plate_gcode_and_md5(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
job = root / "job.gcode"
template = root / "template.gcode.3mf"
output = root / "job.gcode.3mf"
write_gcode(job, "M104 S220\nM140 S65\nG1 X2 Y2 Z0.2 E0.2\n")
write_sliced_3mf(template)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--handoff",
"template-project",
"--template-project",
str(template),
]
)
result = bambu.package_with_template_project(args, output)
self.assertEqual(result["pack"]["replaced_plate"], "Metadata/plate_1.gcode")
with zipfile.ZipFile(output) as archive:
self.assertEqual(archive.read("Metadata/plate_1.gcode"), job.read_bytes())
self.assertEqual(
archive.read("Metadata/plate_1.gcode.md5").decode("ascii"),
hashlib.md5(job.read_bytes()).hexdigest().upper(),
)
def test_template_project_upload_start_execute_uses_root_project_file(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
job = root / "job.gcode"
template = root / "template.gcode.3mf"
write_gcode(job)
write_sliced_3mf(template)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--handoff",
"template-project",
"--template-project",
str(template),
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--access-code",
"12345678",
"--action",
"upload-start",
"--execute",
"--confirm-start-print",
]
)
with (
mock.patch.object(bambu, "upload_ftps") as mocked_upload,
mock.patch.object(bambu, "publish_mqtt") as mocked_publish,
mock.patch("builtins.print") as mocked_print,
):
code = bambu.send_main(args)
self.assertEqual(code, 0)
mocked_upload.assert_called_once()
self.assertEqual(mocked_upload.call_args.args[2], "job.gcode.3mf")
mocked_publish.assert_called_once()
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["handoff"], "template-project")
self.assertEqual(payload["ftps"]["remote_path"], "job.gcode.3mf")
self.assertEqual(payload["ftps"]["url"], "ftp:///job.gcode.3mf")
self.assertEqual(payload["mqtt"]["payload"]["print"]["command"], "project_file")
self.assertEqual(payload["mqtt"]["payload"]["print"]["url"], "ftp:///job.gcode.3mf")
self.assertEqual(payload["mqtt"]["payload"]["print"]["subtask_name"], "job")
self.assertTrue(payload["mqtt"]["payload"]["print"]["flow_cali"])
self.assertRegex(payload["mqtt"]["payload"]["print"]["md5"], r"^[0-9A-F]{32}$")
def test_bambox_upload_start_execute_packages_then_uses_project_file(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
job = root / "job.gcode"
fake_bambox = root / "bambox"
write_gcode(job)
write_fake_bambox(fake_bambox)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--handoff",
"bambox-project",
"--bambox-profile",
"p1s-0.4",
"--filament",
"PLA",
"--bambox-bin",
str(fake_bambox),
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--access-code",
"12345678",
"--action",
"upload-start",
"--execute",
"--confirm-start-print",
]
)
with (
mock.patch.object(bambu, "upload_ftps") as mocked_upload,
mock.patch.object(bambu, "publish_mqtt") as mocked_publish,
mock.patch("builtins.print") as mocked_print,
):
code = bambu.send_main(args)
self.assertEqual(code, 0)
mocked_upload.assert_called_once()
mocked_publish.assert_called_once()
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["executed"], ["upload", "publish_start_request"])
self.assertEqual(payload["mqtt"]["payload"]["print"]["command"], "project_file")
self.assertEqual(payload["mqtt"]["payload"]["print"]["param"], "Metadata/plate_1.gcode")
self.assertEqual(payload["mqtt"]["payload"]["print"]["url"], "ftp:///job.gcode.3mf")
self.assertEqual(payload["mqtt"]["payload"]["print"]["subtask_name"], "job")
self.assertRegex(payload["mqtt"]["payload"]["print"]["md5"], r"^[0-9A-F]{32}$")
def test_implicit_ftps_reuses_control_tls_session_for_data_connection(self) -> None:
context = mock.Mock()
control_socket = mock.Mock()
control_socket.session = "session-token"
data_socket = mock.Mock()
wrapped_data_socket = mock.Mock()
context.wrap_socket.return_value = wrapped_data_socket
ftp = bambu.ImplicitFTP_TLS(context=context)
ftp.sock = control_socket
ftp.host = "printer.local"
ftp._prot_p = True
with mock.patch.object(bambu.ftplib.FTP, "ntransfercmd", return_value=(data_socket, 123)):
conn, size = ftp.ntransfercmd("STOR job.gcode")
self.assertIs(conn, wrapped_data_socket)
self.assertEqual(size, 123)
context.wrap_socket.assert_called_once_with(
data_socket,
server_hostname="printer.local",
session="session-token",
)
def test_ftps_upload_accepts_timeout_when_control_channel_confirms_completion(self) -> None:
ftp = mock.Mock()
ftp.storbinary.side_effect = bambu.socket.timeout("data channel shutdown timed out")
ftp.voidresp.return_value = "226 Transfer complete"
bambu.storbinary_with_bambu_timeout_tolerance(ftp, "STOR job.gcode", mock.Mock())
ftp.storbinary.assert_called_once()
ftp.voidresp.assert_called_once()
def test_start_execute_requires_confirmation_before_network_use(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
job = Path(tmp) / "job.gcode"
write_gcode(job)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--action",
"start",
"--execute",
]
)
with self.assertRaisesRegex(bambu.BambuPrintError, "confirm-start-print"):
bambu.send_main(args)
def test_validate_local_host_accepts_hostname_resolving_to_private_address(self) -> None:
with mock.patch.object(
bambu.socket,
"getaddrinfo",
return_value=[
(
bambu.socket.AF_INET,
bambu.socket.SOCK_STREAM,
0,
"",
("192.168.1.50", 0),
)
],
):
bambu.validate_local_host("printer.local", allow_nonprivate=False)
def test_validate_local_host_rejects_hostname_resolving_to_public_address(self) -> None:
with mock.patch.object(
bambu.socket,
"getaddrinfo",
return_value=[
(
bambu.socket.AF_INET,
bambu.socket.SOCK_STREAM,
0,
"",
("8.8.8.8", 0),
)
],
):
with self.assertRaisesRegex(bambu.BambuPrintError, "non-private"):
bambu.validate_local_host("printer.example.com", allow_nonprivate=False)
def test_validate_local_host_does_not_resolve_when_nonprivate_hosts_are_allowed(self) -> None:
with mock.patch.object(bambu.socket, "getaddrinfo") as mocked_getaddrinfo:
bambu.validate_local_host("printer.example.com", allow_nonprivate=True)
mocked_getaddrinfo.assert_not_called()
def test_validate_local_host_rejects_literal_public_address(self) -> None:
with self.assertRaisesRegex(bambu.BambuPrintError, "non-private"):
bambu.validate_local_host("8.8.8.8", allow_nonprivate=False)
def test_upload_start_reports_published_start_request_not_confirmed_start(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
job = Path(tmp) / "job.gcode"
write_gcode(job)
args = bambu.parse_args(
[
"send",
"--gcode",
str(job),
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--access-code",
"12345678",
"--action",
"upload-start",
"--execute",
"--confirm-start-print",
]
)
with (
mock.patch.object(bambu, "upload_ftps") as mocked_upload,
mock.patch.object(bambu, "publish_mqtt") as mocked_publish,
mock.patch("builtins.print") as mocked_print,
):
code = bambu.send_main(args)
self.assertEqual(code, 0)
mocked_upload.assert_called_once()
mocked_publish.assert_called_once()
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["executed"], ["upload", "publish_start_request"])
self.assertEqual(payload["mqtt"]["payload"]["print"]["command"], "gcode_file")
self.assertIn("does not confirm printer acceptance", payload["execution_notes"][0])
def test_mqtt_client_id_defaults_to_serial(self) -> None:
args = mock.Mock()
args.access_code = "12345678"
args.host = "192.168.1.50"
args.mqtt_port = 8883
args.timeout = 20.0
args.tls_verify = False
args.allow_nonprivate_host = True
args.client_id = ""
args.serial = "00M00A000000000"
sent = bytearray()
connack = bytearray(b"\x20\x02\x00\x00")
class FakeSocket:
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def sendall(self, data):
sent.extend(data)
def recv(self, size):
chunk = bytes(connack[:size])
del connack[:size]
return chunk
fake_context = mock.Mock()
fake_context.wrap_socket.return_value = FakeSocket()
with (
mock.patch.object(bambu.ssl, "create_default_context", return_value=fake_context),
mock.patch.object(bambu.socket, "create_connection", return_value=FakeSocket()),
):
bambu.publish_mqtt(args, "device/00M00A000000000/request", {"print": {}})
self.assertIn(b"00M00A000000000", bytes(sent))
def test_mqtt_qos1_publish_waits_for_puback(self) -> None:
args = mock.Mock()
args.access_code = "12345678"
args.host = "192.168.1.50"
args.mqtt_port = 8883
args.timeout = 20.0
args.tls_verify = False
args.allow_nonprivate_host = True
args.client_id = ""
args.serial = "00M00A000000000"
args.mqtt_qos = 1
args.wait_after_publish = 0.0
sent = bytearray()
incoming = bytearray(b"\x20\x02\x00\x00\x40\x02\x00\x02")
class FakeSocket:
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def sendall(self, data):
sent.extend(data)
def recv(self, size):
chunk = bytes(incoming[:size])
del incoming[:size]
return chunk
def settimeout(self, timeout):
self.timeout = timeout
fake_context = mock.Mock()
fake_context.wrap_socket.return_value = FakeSocket()
with (
mock.patch.object(bambu.ssl, "create_default_context", return_value=fake_context),
mock.patch.object(bambu.socket, "create_connection", return_value=FakeSocket()),
):
result = bambu.publish_mqtt(args, "device/00M00A000000000/request", {"print": {}})
self.assertTrue(result["puback_received"])
self.assertIn(b"\x32", bytes(sent))
def test_mqtt_packet_encoding(self) -> None:
self.assertEqual(bambu.encode_remaining_length(0), b"\x00")
self.assertEqual(bambu.encode_remaining_length(321), b"\xc1\x02")
packet = bambu.mqtt_publish_packet("device/abc/request", json.dumps({"print": {}}).encode())
self.assertEqual(packet[0], 0x30)
self.assertIn(b"device/abc/request", packet)
qos_packet = bambu.mqtt_publish_packet(
"device/abc/request",
json.dumps({"print": {}}).encode(),
qos=1,
packet_id=7,
)
self.assertEqual(qos_packet[0], 0x32)
self.assertIn(b"\x00\x07", qos_packet)
subscribe = bambu.mqtt_subscribe_packet(1, "device/abc/report")
self.assertEqual(subscribe[0], 0x82)
self.assertIn(b"device/abc/report", subscribe)
def test_status_command_uses_configured_printer_report_topic(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
config = root / "printers.json"
config.write_text(
json.dumps(
{
"printers": {
"a1-mini": {
"host": "192.168.1.34",
"access_code": "12345678",
"serial": "0309CA4C0901107",
}
}
}
),
encoding="utf-8",
)
args = bambu.parse_args(["status", "--config", str(config), "--printer", "a1-mini", "--wait-seconds", "0.1"])
with mock.patch.object(
bambu,
"subscribe_mqtt_reports",
return_value=[{"topic": "device/0309CA4C0901107/report", "json": {"print": {"gcode_state": "RUNNING"}}}],
) as mocked_status, mock.patch("builtins.print") as mocked_print:
code = bambu.status_main(args)
self.assertEqual(code, 0)
mocked_status.assert_called_once()
self.assertEqual(mocked_status.call_args.args[1], "device/0309CA4C0901107/report")
payload = json.loads(mocked_print.call_args.args[0])
self.assertTrue(payload["ok"])
def test_clear_error_dry_run_builds_clean_print_error_payload(self) -> None:
args = bambu.parse_args(
[
"clear-error",
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--sequence-id",
"seq-1",
]
)
plan = bambu.build_clear_error_plan(args)
self.assertTrue(plan["dry_run"])
self.assertEqual(plan["mqtt"]["topic"], "device/00M00A000000000/request")
self.assertEqual(plan["mqtt"]["payload"]["print"]["command"], "clean_print_error")
self.assertEqual(plan["mqtt"]["payload"]["print"]["sequence_id"], "seq-1")
def test_clear_error_execute_publishes_without_start_confirmation(self) -> None:
args = bambu.parse_args(
[
"clear-error",
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--access-code",
"12345678",
"--execute",
]
)
with mock.patch.object(bambu, "publish_mqtt", return_value={"puback_received": True}) as mocked_publish, mock.patch(
"builtins.print"
) as mocked_print:
code = bambu.clear_error_main(args)
self.assertEqual(code, 0)
mocked_publish.assert_called_once()
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["executed"], ["publish_clean_print_error"])
self.assertEqual(payload["mqtt"]["payload"]["print"]["command"], "clean_print_error")
def test_pause_dry_run_builds_pause_payload(self) -> None:
args = bambu.parse_args(
[
"pause",
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--sequence-id",
"seq-1",
]
)
plan = bambu.build_print_control_plan(args)
self.assertTrue(plan["dry_run"])
self.assertEqual(plan["action"], "pause")
self.assertEqual(plan["mqtt"]["topic"], "device/00M00A000000000/request")
self.assertFalse(plan["mqtt"]["will_publish"])
payload = plan["mqtt"]["payload"]["print"]
self.assertEqual(payload["command"], "pause")
self.assertEqual(payload["sequence_id"], "seq-1")
def test_cancel_dry_run_maps_to_bambu_stop_payload(self) -> None:
args = bambu.parse_args(
[
"cancel",
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--sequence-id",
"seq-2",
]
)
plan = bambu.build_print_control_plan(args)
self.assertTrue(plan["dry_run"])
self.assertEqual(plan["action"], "cancel")
payload = plan["mqtt"]["payload"]["print"]
self.assertEqual(payload["command"], "stop")
self.assertEqual(payload["param"], "")
self.assertEqual(payload["sequence_id"], "seq-2")
def test_cancel_execute_requires_cancel_confirmation(self) -> None:
args = bambu.parse_args(
[
"cancel",
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--access-code",
"12345678",
"--execute",
]
)
with self.assertRaisesRegex(bambu.BambuPrintError, "--confirm-cancel-print"):
bambu.print_control_main(args)
def test_cancel_execute_publishes_stop_after_confirmation(self) -> None:
args = bambu.parse_args(
[
"cancel",
"--host",
"192.168.1.50",
"--serial",
"00M00A000000000",
"--access-code",
"12345678",
"--execute",
"--confirm-cancel-print",
]
)
with mock.patch.object(bambu, "publish_mqtt", return_value={"puback_received": True}) as mocked_publish, mock.patch(
"builtins.print"
) as mocked_print:
code = bambu.print_control_main(args)
self.assertEqual(code, 0)
mocked_publish.assert_called_once()
payload = json.loads(mocked_print.call_args.args[0])
self.assertEqual(payload["executed"], ["publish_cancel_request"])
self.assertEqual(payload["mqtt"]["payload"]["print"]["command"], "stop")
def test_formats_hms_codes_from_status_payload(self) -> None:
message = {"json": {"print": {"hms": [{"attr": 83887360, "code": 65543, "action": 0}]}}}
bambu.annotate_hms_codes(message)
self.assertEqual(message["hms"][0]["hms_code"], "0500-0500-0001-0007")
if __name__ == "__main__":
unittest.main()
@@ -21,42 +21,33 @@ class CadRouterTests(unittest.TestCase):
return route_module.build_parser().parse_args(list(args))
def test_routes_gear_to_simplecadapi(self) -> None:
result = route_module.route(self.parse("生成一个模数 2 的人字齿轮", "--output", "step"))
result = route_module.route(self.parse("生成一个模数 2 的人字齿轮"))
self.assertEqual("simplecadapi", result["selected_backend"])
def test_routes_connecting_rod_to_build123d(self) -> None:
result = route_module.route(
self.parse("生成中心距 120mm 的发动机连杆", "--output", "step", "--manufacturing", "machining")
)
result = route_module.route(self.parse("生成中心距 120mm 的发动机连杆"))
self.assertEqual("build123d", result["selected_backend"])
def test_routes_printable_vase_to_cadam(self) -> None:
result = route_module.route(
self.parse("生成一个带蜂窝纹理的可调参数 3D 打印花瓶", "--output", "stl", "--browser-controls")
)
self.assertEqual("cadam", result["selected_backend"])
self.assertEqual("CADAM", result["project"])
self.assertEqual("scripts/cadam_compile.mjs", result["adapter"])
def test_existing_step_stays_step_first(self) -> None:
result = route_module.route(self.parse("修改现有零件的孔径", "--existing-model", "--output", "step"))
result = route_module.route(self.parse("修改现有零件的孔径", "--existing-model"))
self.assertEqual("build123d", result["selected_backend"])
def test_scad_output_is_hard_signal(self) -> None:
result = route_module.route(self.parse("做一个参数化旋钮", "--output", "scad"))
self.assertEqual("cadam", result["selected_backend"])
def test_every_route_reports_only_runtime_project_roles(self) -> None:
result = route_module.route(self.parse("生成一个 40mm 法兰", "--output", "step"))
def test_every_route_has_fixed_step_artifact_contract(self) -> None:
result = route_module.route(self.parse("做一个参数化旋钮"))
self.assertEqual(
{"text-to-cad", "SimpleCADAPI", "CADAM"},
set(result["project_contributions"]),
{
"primary_format": "step",
"editable_contract": "DesignIR 2.0",
},
result["artifact_contract"],
)
def test_cadam_probe_finds_sibling_runtime(self) -> None:
expected_root = Path(__file__).resolve().parents[5] / "CADAM"
if expected_root.is_dir():
self.assertTrue(route_module.cadam_runtime_available(expected_root))
def test_every_route_reports_only_runtime_project_roles(self) -> None:
result = route_module.route(self.parse("生成一个 40mm 法兰"))
self.assertEqual(
{"text-to-cad", "SimpleCADAPI"},
set(result["project_contributions"]),
)
def test_explicit_override_is_respected(self) -> None:
result = route_module.route(self.parse("生成一个齿轮", "--backend", "build123d"))
@@ -94,8 +85,6 @@ class CadRouterTests(unittest.TestCase):
result = route_module.route(
self.parse(
"根据经验生成一个法兰轴套",
"--output",
"step",
"--experience-library",
str(library),
"--experience-family",
@@ -209,8 +198,6 @@ class CadRouterTests(unittest.TestCase):
result = route_module.route(
self.parse(
"生成一个带法兰轴套和中心通孔",
"--output",
"step",
"--experience-library",
str(library),
)
@@ -273,8 +260,6 @@ class CadRouterTests(unittest.TestCase):
result = route_module.route(
self.parse(
"生成一个带法兰轴套",
"--output",
"step",
"--experience-library",
str(library),
)
@@ -318,8 +303,6 @@ class CadRouterTests(unittest.TestCase):
result = route_module.route(
self.parse(
"Generate a mounting plate with four corner holes",
"--output",
"step",
"--experience-library",
str(library),
)
@@ -336,25 +319,26 @@ class CadRouterTests(unittest.TestCase):
[item["id"] for item in context["experiences"]],
)
def test_step_edit_uses_source_and_library_not_private_case(self) -> None:
def test_step_edit_uses_private_evidence_and_isolated_designir(self) -> None:
result = route_module.route(
self.parse(
"把中心孔径调整大一些",
"--edit-source",
"/tmp/source.step",
"--output",
"step",
)
)
plan = result["design_plan"]
self.assertEqual("modify", plan["operation"])
self.assertEqual(
"direct_step_feature_rebuild",
"private_evidence_to_isolated_designir_reconstruction",
plan["edit_context"]["modification_mode"],
)
forbidden = plan["edit_context"]["forbidden_knowledge_sources"]
self.assertIn("cad-experience-plugin/parser/output", forbidden)
self.assertNotIn("private_case", plan["edit_context"])
self.assertIn("teacher STEP during reconstruction", forbidden)
self.assertIn(
"deterministically extracted private teacher evidence",
plan["edit_context"]["allowed_knowledge_sources"],
)
self.assertEqual(
"pending_source_classification", plan["experience_status"]
)
@@ -368,8 +352,6 @@ class CadRouterTests(unittest.TestCase):
"把孔径改为 12mm",
"--edit-source",
"/tmp/model.py",
"--output",
"step",
)
)
self.assertEqual(
@@ -377,22 +359,21 @@ class CadRouterTests(unittest.TestCase):
result["design_plan"]["edit_context"]["modification_mode"],
)
def test_parser_staging_files_are_rejected_as_edit_sources(self) -> None:
parser_source = (
def test_teacher_vault_is_rejected_as_reconstruction_source(self) -> None:
teacher_source = (
Path(__file__).resolve().parents[5]
/ "cad-experience-plugin"
/ "parser"
/ "input"
/ "designir-pipeline"
/ "workspace"
/ "teacher"
/ "inbox"
/ "part.step"
)
with self.assertRaisesRegex(ValueError, "cannot use"):
with self.assertRaisesRegex(ValueError, "cannot read the teacher"):
route_module.route(
self.parse(
"修改这个零件",
"--edit-source",
str(parser_source),
"--output",
"step",
str(teacher_source),
)
)
@@ -1,78 +0,0 @@
import subprocess
import sys
import unittest
from unittest import mock
from tests.python.support.paths import add_repo_path, repo_path
add_repo_path("skills/dxf/scripts")
from dxf import cli as dxf
class DxfCliTests(unittest.TestCase):
def test_requires_explicit_target(self) -> None:
with self.assertRaises(SystemExit) as cm:
dxf.main([])
self.assertEqual(2, cm.exception.code)
def test_passes_targets_in_order(self) -> None:
with mock.patch.object(dxf, "generate_dxf_targets", return_value=0) as generate:
self.assertEqual(0, dxf.main(["drawings/second.py", "drawings/first.py"]))
generate.assert_called_once_with(["drawings/second.py", "drawings/first.py"], output=None, verbose=False)
def test_passes_verbose_flag(self) -> None:
with mock.patch.object(dxf, "generate_dxf_targets", return_value=0) as generate:
self.assertEqual(0, dxf.main(["drawings/part.py", "--verbose"]))
generate.assert_called_once_with(["drawings/part.py"], output=None, verbose=True)
def test_passes_output_flag(self) -> None:
with mock.patch.object(dxf, "generate_dxf_targets", return_value=0) as generate:
self.assertEqual(0, dxf.main(["drawings/part.py", "-o", "DXF/part.dxf"]))
generate.assert_called_once_with(["drawings/part.py"], output="DXF/part.dxf", verbose=False)
def test_output_flag_rejects_multiple_targets(self) -> None:
with self.assertRaises(SystemExit) as cm:
dxf.main(["drawings/first.py", "drawings/second.py", "-o", "DXF/first.dxf"])
self.assertEqual(2, cm.exception.code)
def test_scripts_dxf_directory_invokes_cli(self) -> None:
skill_root = repo_path("skills/dxf")
result = subprocess.run(
[sys.executable, "scripts/dxf", "--help"],
cwd=skill_root,
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
self.assertEqual("", result.stderr)
self.assertEqual(0, result.returncode)
self.assertIn("usage: dxf", result.stdout)
self.assertIn("--output", result.stdout)
def test_cli_import_does_not_import_heavy_cad_modules(self) -> None:
skill_root = repo_path("skills/dxf")
code = (
"import sys; sys.path.insert(0, 'scripts'); import dxf.cli; "
"print('OCP.OCP' in sys.modules); "
"print('cadpy.step_scene' in sys.modules)"
)
result = subprocess.run(
[sys.executable, "-c", code],
cwd=skill_root,
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
self.assertEqual("", result.stderr)
self.assertEqual(0, result.returncode)
self.assertEqual(["False", "False"], result.stdout.strip().splitlines())
if __name__ == "__main__":
unittest.main()
@@ -1,37 +0,0 @@
import unittest
from pathlib import Path
import ezdxf
from tests.python.support.paths import add_repo_path
from tests.python.support.tmp_root import temporary_directory
add_repo_path("skills/dxf/scripts")
from dxf.render_payload import build_dxf_render_payload
class CadpyDxfTests(unittest.TestCase):
def test_build_dxf_render_payload_supports_straight_lwpolyline(self) -> None:
with temporary_directory(prefix="tmp-cad-dxf-") as tmpdir:
dxf_path = Path(tmpdir) / "outline.dxf"
doc = ezdxf.new("R2010")
modelspace = doc.modelspace()
modelspace.add_lwpolyline(
[(0.0, 0.0), (10.0, 0.0), (10.0, 5.0), (0.0, 5.0)],
close=True,
dxfattribs={"layer": "CUT"},
)
doc.saveas(dxf_path)
payload = build_dxf_render_payload(dxf_path, file_ref="test/outline.dxf")
self.assertEqual(4, payload["counts"]["paths"])
self.assertEqual(0, payload["counts"]["circles"])
self.assertEqual(4, len(payload["geometry"]["lines"]))
self.assertEqual(10.0, payload["bounds"]["width"])
self.assertEqual(5.0, payload["bounds"]["height"])
if __name__ == "__main__":
unittest.main()
@@ -1,77 +0,0 @@
import textwrap
import unittest
from pathlib import Path
from cadpy import catalog as cad_catalog
from cadpy import generation as cad_generation
from cadpy.metadata import parse_generator_metadata
from tests.python.support.tmp_root import temporary_directory
STANDALONE_DXF_SOURCE = textwrap.dedent(
'''
"""Standalone DXF drafting source."""
import ezdxf
def gen_dxf():
doc = ezdxf.new()
msp = doc.modelspace()
msp.add_lwpolyline([(0, 0), (40, 0), (40, 20), (0, 20)], close=True)
return doc
'''
).strip()
def _write_standalone_source(root: Path, stem: str = "outline") -> Path:
script_path = root / f"{stem}.py"
script_path.write_text(STANDALONE_DXF_SOURCE + "\n")
return script_path
class StandaloneDxfSourceTests(unittest.TestCase):
def test_metadata_allows_gen_dxf_without_gen_step(self) -> None:
with temporary_directory(prefix="dxf-skill") as root:
script_path = _write_standalone_source(Path(root))
metadata = parse_generator_metadata(script_path)
assert metadata is not None
self.assertTrue(metadata.has_gen_dxf)
self.assertFalse(metadata.has_gen_step)
self.assertIsNone(metadata.kind)
def test_metadata_still_requires_gen_step_for_urdf_and_sdf(self) -> None:
with temporary_directory(prefix="dxf-skill") as root:
script_path = Path(root) / "robot.py"
script_path.write_text("def gen_urdf():\n return '<robot/>'\n")
with self.assertRaisesRegex(ValueError, "require gen_step"):
parse_generator_metadata(script_path)
def test_explicit_target_resolves_dxf_only_source(self) -> None:
with temporary_directory(prefix="dxf-skill") as root:
script_path = _write_standalone_source(Path(root))
source = cad_catalog.source_from_path(script_path)
assert source is not None
self.assertEqual("dxf", source.kind)
self.assertIsNone(source.step_path)
self.assertEqual(script_path.with_suffix(".dxf"), source.dxf_path)
def test_directory_catalog_skips_dxf_only_source(self) -> None:
with temporary_directory(prefix="dxf-skill") as root:
_write_standalone_source(Path(root))
self.assertEqual((), cad_catalog.iter_cad_sources(Path(root)))
def test_generate_dxf_targets_writes_sibling_output(self) -> None:
with temporary_directory(prefix="dxf-skill") as root:
script_path = _write_standalone_source(Path(root))
self.assertEqual(0, cad_generation.generate_dxf_targets([str(script_path)]))
output_path = script_path.with_suffix(".dxf")
self.assertTrue(output_path.exists())
self.assertGreater(output_path.stat().st_size, 0)
if __name__ == "__main__":
unittest.main()
@@ -1,301 +0,0 @@
#!/usr/bin/env python3
from __future__ import annotations
import json
import os
import tempfile
import unittest
import zipfile
from pathlib import Path
from tests.python.support.paths import add_repo_path
add_repo_path("skills/gcode/scripts")
import gcode_tool as gcode
def make_executable(path: Path) -> None:
path.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
path.chmod(0o755)
def write_profile(tmp: Path, backend: str = "orcaslicer") -> Path:
native_config = tmp / f"{backend}.ini"
native_config.write_text("# slicer profile\n", encoding="utf-8")
profile = tmp / "profile.json"
profile.write_text(
json.dumps(
{
"backend": backend,
"native_config": str(native_config),
"machine": {
"name": "Test Printer",
"bed_size_mm": [180, 180],
"z_height_mm": 180,
},
"filament": {
"type": "PLA",
"nozzle_temp_c": 220,
"bed_temp_c": 65,
},
}
),
encoding="utf-8",
)
return profile
def write_profile_with_motion_bounds(tmp: Path) -> Path:
native_config = tmp / "orcaslicer.ini"
native_config.write_text("# slicer profile\n", encoding="utf-8")
profile = tmp / "profile_with_motion_bounds.json"
profile.write_text(
json.dumps(
{
"backend": "orcaslicer",
"native_config": str(native_config),
"machine": {
"name": "Test Printer",
"bed_size_mm": [180, 180],
"z_height_mm": 180,
"motion_bounds_mm": {
"x": [-14, 181],
"y": [-4, 185],
"z": [-1.1, 180],
},
},
"filament": {
"type": "PLA",
"nozzle_temp_c": 220,
"bed_temp_c": 65,
},
}
),
encoding="utf-8",
)
return profile
class GCodeToolTests(unittest.TestCase):
def test_discovers_fake_preferred_backend_paths(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
bin_dir = Path(tmp) / "bin"
bin_dir.mkdir()
make_executable(bin_dir / "OrcaSlicer")
make_executable(bin_dir / "prusa-slicer")
report = gcode.discovery_report(search_path=str(bin_dir))
backends = {item["id"]: item for item in report["backends"]}
self.assertTrue(backends["orcaslicer"]["available"])
self.assertTrue(backends["prusa-slicer"]["available"])
self.assertFalse(backends["curaengine"]["available"])
self.assertEqual(report["preferred_order"], ["orcaslicer", "prusa-slicer", "curaengine"])
def test_profile_validation_requires_backend_native_config_and_bed_limits(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
missing_backend = root / "missing_backend.json"
missing_backend.write_text("{}", encoding="utf-8")
with self.assertRaisesRegex(gcode.GCodeToolError, "backend"):
gcode.load_profile(missing_backend)
missing_native = root / "missing_native.json"
missing_native.write_text(
json.dumps(
{
"backend": "orcaslicer",
"machine": {"name": "Printer", "bed_size_mm": [180, 180], "z_height_mm": 180},
"filament": {"type": "PLA", "nozzle_temp_c": 220, "bed_temp_c": 65},
}
),
encoding="utf-8",
)
with self.assertRaisesRegex(gcode.GCodeToolError, "native_config"):
gcode.load_profile(missing_native)
bad_bed = root / "bad_bed.json"
native_config = root / "profile.ini"
native_config.write_text("# config\n", encoding="utf-8")
bad_bed.write_text(
json.dumps(
{
"backend": "orcaslicer",
"native_config": str(native_config),
"machine": {"name": "Printer", "bed_size_mm": [180], "z_height_mm": 180},
"filament": {"type": "PLA", "nozzle_temp_c": 220, "bed_temp_c": 65},
}
),
encoding="utf-8",
)
with self.assertRaisesRegex(gcode.GCodeToolError, "bed_size_mm"):
gcode.load_profile(bad_bed)
def test_input_classification_for_supported_rejected_and_sliced_bambu_files(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
stl = root / "part.stl"
stl.write_text("solid part\nendsolid part\n", encoding="utf-8")
stl_info = gcode.inspect_input(stl)
self.assertTrue(stl_info.direct_to_slicer)
self.assertFalse(stl_info.needs_stl_conversion)
glb = root / "part.glb"
glb.write_bytes(b"glTF")
glb_info = gcode.inspect_input(glb)
self.assertTrue(glb_info.needs_stl_conversion)
sliced = root / "job.gcode.3mf"
with zipfile.ZipFile(sliced, "w") as archive:
archive.writestr("[Content_Types].xml", "<Types/>")
archive.writestr("Metadata/plate_1.gcode", "G1 X1\n")
sliced_info = gcode.inspect_input(sliced)
self.assertTrue(sliced_info.already_sliced_bambu)
self.assertEqual(sliced_info.status, "already_sliced_bambu_3mf")
step = root / "part.step"
step.write_text("ISO-10303-21;", encoding="utf-8")
with self.assertRaisesRegex(gcode.GCodeToolError, "out of scope"):
gcode.inspect_input(step)
def test_dry_run_command_construction_for_each_backend(self) -> None:
cases = [
("orcaslicer", "OrcaSlicer", ["--load-settings", "--outputdir", "--slice"]),
("prusa-slicer", "prusa-slicer", ["--load", "--export-gcode", "--output"]),
("curaengine", "CuraEngine", ["slice", "-j", "-l", "-o"]),
]
for backend, executable, expected_parts in cases:
with self.subTest(backend=backend), tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
bin_dir = root / "bin"
bin_dir.mkdir()
make_executable(bin_dir / executable)
profile = write_profile(root, backend)
model = root / "part.obj"
model.write_text("o part\n", encoding="utf-8")
output = root / "part.gcode"
args = gcode.build_parser().parse_args(
[
"slice",
"--input",
str(model),
"--output",
str(output),
"--profile",
str(profile),
"--backend",
"auto",
"--dry-run",
]
)
plan = gcode.build_slice_plan(args, search_path=str(bin_dir))
command = plan["command"]
self.assertEqual(Path(command[0]).name, executable)
for part in expected_parts:
self.assertIn(part, command)
self.assertEqual(plan["backend"], backend)
self.assertFalse(plan["conversion"]["required"])
def test_refuses_to_slice_already_sliced_bambu_3mf(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
bin_dir = root / "bin"
bin_dir.mkdir()
make_executable(bin_dir / "OrcaSlicer")
profile = write_profile(root)
sliced = root / "job.gcode.3mf"
with zipfile.ZipFile(sliced, "w") as archive:
archive.writestr("Metadata/plate_1.gcode", "G1 X1\n")
args = gcode.build_parser().parse_args(
[
"slice",
"--input",
str(sliced),
"--output",
str(root / "job.gcode"),
"--profile",
str(profile),
"--dry-run",
]
)
with self.assertRaisesRegex(gcode.GCodeToolError, "already a sliced Bambu"):
gcode.build_slice_plan(args, search_path=str(bin_dir))
def test_gcode_validation_passes_valid_simple_file(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = gcode.load_profile(write_profile(root))
toolpath = root / "valid.gcode"
toolpath.write_text(
"\n".join(
[
"M104 S220",
"M140 S65",
"G90",
"G1 X10 Y10 Z0.2 F1800",
"G1 X20 Y10 E0.4 F1200",
]
),
encoding="utf-8",
)
result = gcode.validate_gcode_file(toolpath, profile)
self.assertTrue(result["ok"])
self.assertEqual(result["errors"], [])
self.assertGreaterEqual(result["stats"]["extrusion_moves"], 1)
def test_gcode_validation_reports_empty_no_extrusion_and_out_of_bounds(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = gcode.load_profile(write_profile(root))
empty = root / "empty.gcode"
empty.write_text("", encoding="utf-8")
empty_result = gcode.validate_gcode_file(empty, profile)
self.assertFalse(empty_result["ok"])
self.assertIn("G-code file is empty.", empty_result["errors"])
no_extrusion = root / "no_extrusion.gcode"
no_extrusion.write_text("M104 S220\nG1 X10 Y10 Z0.2\n", encoding="utf-8")
no_extrusion_result = gcode.validate_gcode_file(no_extrusion, profile)
self.assertFalse(no_extrusion_result["ok"])
self.assertIn("No extrusion moves found.", no_extrusion_result["errors"])
out_of_bounds = root / "out_of_bounds.gcode"
out_of_bounds.write_text("M104 S220\nG1 X999 Y10 Z0.2 E0.1\n", encoding="utf-8")
out_of_bounds_result = gcode.validate_gcode_file(out_of_bounds, profile)
self.assertFalse(out_of_bounds_result["ok"])
self.assertTrue(any("X=999.0" in error for error in out_of_bounds_result["errors"]))
def test_gcode_validation_uses_optional_motion_bounds(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = gcode.load_profile(write_profile_with_motion_bounds(root))
toolpath = root / "native_start_positions.gcode"
toolpath.write_text("M104 S220\nM140 S65\nG90\nG1 X-13.5 Y-4 Z-1 E0.1\n", encoding="utf-8")
result = gcode.validate_gcode_file(toolpath, profile)
self.assertTrue(result["ok"])
def test_gcode_validation_warns_for_unknown_commands(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = gcode.load_profile(write_profile(root))
toolpath = root / "unknown.gcode"
toolpath.write_text("M104 S220\nM999\nG1 X10 Y10 Z0.2 E0.1\n", encoding="utf-8")
result = gcode.validate_gcode_file(toolpath, profile)
self.assertTrue(result["ok"])
self.assertTrue(any("M999" in warning for warning in result["warnings"]))
if __name__ == "__main__":
unittest.main()
@@ -1,51 +0,0 @@
import unittest
import xml.etree.ElementTree as ET
from tests.python.support.paths import add_repo_path
add_repo_path("skills/sdf/scripts")
from sdf import builder
class SdfBuilderTests(unittest.TestCase):
def test_builds_minimal_model_with_box_geometry(self) -> None:
root = builder.sdf_root()
model = builder.model(root, "fixture", static=False)
base = builder.link(model, "base_link")
visual = builder.visual(base, "base_visual")
builder.box(visual, (0.4, 0.3, 0.1))
collision = builder.collision(base, "base_collision")
builder.box(collision, (0.4, 0.3, 0.1))
xml = ET.tostring(root, encoding="unicode")
self.assertIn('<sdf version="1.12"', xml)
self.assertIn('<model name="fixture"', xml)
self.assertIn("<size>0.4 0.3 0.1</size>", xml)
def test_rejects_invalid_numeric_values(self) -> None:
root = builder.sdf_root()
model = builder.model(root, "fixture")
base = builder.link(model, "base_link")
visual = builder.visual(base, "base_visual")
with self.assertRaisesRegex(ValueError, "positive"):
builder.box(visual, (1, 0, 1))
with self.assertRaisesRegex(ValueError, "nonzero"):
builder.axis(model, (0, 0, 0))
def test_builds_static_plugin_metadata(self) -> None:
root = builder.sdf_root()
model = builder.model(root, "fixture")
builder.plugin(model, "controller", "gz-sim-joint-controller-system", topic="/cmd")
xml = ET.tostring(root, encoding="unicode")
self.assertIn('name="controller"', xml)
self.assertIn('filename="gz-sim-joint-controller-system"', xml)
self.assertIn("<topic>/cmd</topic>", xml)
if __name__ == "__main__":
unittest.main()
@@ -1,297 +0,0 @@
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from tests.python.support.paths import add_repo_path
add_repo_path("skills/sdf/scripts")
from sdf import cli
from sdf.source import SdfSourceError
def _write_sdf_source(path: Path, body: str) -> None:
path.write_text(
"\n".join(
[
"def gen_sdf():",
*[f" {line}" for line in body.splitlines()],
"",
]
),
encoding="utf-8",
)
def _strip_cadpy_metadata_comments(text: str) -> str:
return "\n".join(
line for line in text.splitlines() if not line.startswith("<!-- cadpy:")
) + "\n"
class SdfCliTests(unittest.TestCase):
def test_requires_explicit_target(self) -> None:
with self.assertRaises(SystemExit) as cm:
cli.main([])
self.assertEqual(2, cm.exception.code)
def test_rejects_summary_option(self) -> None:
with self.assertRaises(SystemExit) as cm:
cli.main(["sample_robot.py", "--summary"])
self.assertEqual(2, cm.exception.code)
def test_passes_targets_and_output(self) -> None:
with mock.patch.object(cli, "generate_sdf_targets", return_value=0) as generate:
self.assertEqual(0, cli.main(["sample_robot.py", "-o", "sample_robot.sdf"]))
generate.assert_called_once_with(["sample_robot.py"], output="sample_robot.sdf", gz_check="auto", strict=False)
def test_passes_gz_check_and_strict_options(self) -> None:
with mock.patch.object(cli, "generate_sdf_targets", return_value=0) as generate:
self.assertEqual(0, cli.main(["sample_robot.py", "--gz-check", "required", "--strict"]))
generate.assert_called_once_with(["sample_robot.py"], output=None, gz_check="required", strict=True)
def test_rejects_output_with_pair_target(self) -> None:
with self.assertRaises(SystemExit) as cm:
cli.main(["sample_robot.py=sample_robot.sdf", "-o", "other.sdf"])
self.assertEqual(2, cm.exception.code)
def test_generates_default_sibling_output_from_xml_string(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_sdf_source(
source_path,
"return '<sdf version=\"1.12\"><model name=\"sample\"><link name=\"base_link\" /></model></sdf>'",
)
self.assertEqual(0, cli.generate_sdf_targets([str(source_path)]))
output_text = source_path.with_suffix(".sdf").read_text(encoding="utf-8")
self.assertIn("<!-- cadpy:sourcePath=", output_text)
self.assertEqual(
'<sdf version="1.12"><model name="sample"><link name="base_link" /></model></sdf>\n',
_strip_cadpy_metadata_comments(output_text),
)
def test_validates_generated_sdf_before_writing_output(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
output_path = source_path.with_suffix(".sdf")
output_path.write_text("original output\n", encoding="utf-8")
_write_sdf_source(
source_path,
"return '<robot name=\"not_sdf\" />'",
)
with self.assertRaisesRegex(SdfSourceError, "root element must be <sdf>"):
cli.generate_sdf_targets([str(source_path)])
self.assertEqual("original output\n", output_path.read_text(encoding="utf-8"))
def test_generates_default_sibling_output_from_element_root(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
source_path.write_text(
"\n".join(
[
"import xml.etree.ElementTree as ET",
"",
"def gen_sdf():",
" sdf = ET.Element('sdf', {'version': '1.12'})",
" model = ET.SubElement(sdf, 'model', {'name': 'sample'})",
" ET.SubElement(model, 'link', {'name': 'base_link'})",
" return sdf",
"",
]
),
encoding="utf-8",
)
self.assertEqual(0, cli.generate_sdf_targets([str(source_path)]))
output_text = source_path.with_suffix(".sdf").read_text(encoding="utf-8")
self.assertIn("<!-- cadpy:sourcePath=", output_text)
self.assertEqual(
'<?xml version="1.0"?>\n'
'<sdf version="1.12">\n'
' <model name="sample">\n'
' <link name="base_link" />\n'
' </model>\n'
'</sdf>\n',
_strip_cadpy_metadata_comments(output_text),
)
def test_generates_envelope_output_from_element_root(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
source_path.write_text(
"\n".join(
[
"import xml.etree.ElementTree as ET",
"",
"def gen_sdf():",
" sdf = ET.Element('sdf', {'version': '1.12'})",
" model = ET.SubElement(sdf, 'model', {'name': 'sample'})",
" ET.SubElement(model, 'link', {'name': 'base_link'})",
" return {'xml': sdf}",
"",
]
),
encoding="utf-8",
)
self.assertEqual(0, cli.generate_sdf_targets([str(source_path)]))
self.assertIn(
'<link name="base_link" />',
source_path.with_suffix(".sdf").read_text(encoding="utf-8"),
)
def test_generates_output_override_for_single_target(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
output_path = Path(tempdir) / "custom" / "robot.sdf"
_write_sdf_source(
source_path,
"return '<sdf version=\"1.12\"><model name=\"sample\"><link name=\"base_link\" /></model></sdf>'",
)
self.assertEqual(0, cli.generate_sdf_targets([str(source_path)], output=str(output_path)))
self.assertTrue(output_path.exists())
self.assertFalse(source_path.with_suffix(".sdf").exists())
def test_generates_mixed_plain_and_paired_targets(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
root = Path(tempdir)
first_path = root / "first.py"
second_path = root / "second.py"
second_output = root / "custom" / "second.sdf"
_write_sdf_source(first_path, "return '<sdf version=\"1.12\"><model name=\"first\"><link name=\"base\" /></model></sdf>'")
_write_sdf_source(second_path, "return '<sdf version=\"1.12\"><model name=\"second\"><link name=\"base\" /></model></sdf>'")
self.assertEqual(0, cli.generate_sdf_targets([str(first_path), f"{second_path}={second_output}"]))
self.assertTrue(first_path.with_suffix(".sdf").exists())
self.assertTrue(second_output.exists())
self.assertFalse(second_path.with_suffix(".sdf").exists())
def test_rejects_legacy_sdf_output_field(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_sdf_source(
source_path,
"\n".join(
[
"return {",
" 'xml': '<sdf version=\"1.12\"><model name=\"sample\"><link name=\"base_link\" /></model></sdf>',",
" 'sdf_output': 'legacy/ignored.sdf',",
"}",
]
),
)
with self.assertRaisesRegex(TypeError, "unsupported field\\(s\\): sdf_output"):
cli.generate_sdf_targets([str(source_path)])
self.assertFalse(source_path.with_suffix(".sdf").exists())
self.assertFalse((Path(tempdir) / "legacy" / "ignored.sdf").exists())
def test_envelope_prints_assumptions_and_warnings(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_sdf_source(
source_path,
"\n".join(
[
"return {",
" 'xml': '<sdf version=\"1.12\"><model name=\"sample\"><link name=\"base_link\" /></model></sdf>',",
" 'metadata': {'target_consumer': 'CAD Viewer'},",
" 'assumptions': [{'code': 'mesh_units', 'message': 'Assumed mesh units are meters'}],",
" 'warnings': ['Plugin startup was not smoke-tested'],",
"}",
]
),
)
with mock.patch("builtins.print") as print_mock:
self.assertEqual(0, cli.generate_sdf_targets([str(source_path)], gz_check="never"))
printed = "\n".join(" ".join(str(arg) for arg in call.args) for call in print_mock.call_args_list)
self.assertIn("Assumption [mesh_units]", printed)
self.assertIn("generator_warning", printed)
def test_rejects_unknown_envelope_fields(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_sdf_source(
source_path,
"\n".join(
[
"return {",
" 'xml': '<sdf version=\"1.12\"><model name=\"sample\"><link name=\"base_link\" /></model></sdf>',",
" 'unexpected': True,",
"}",
]
),
)
with self.assertRaisesRegex(TypeError, "unsupported field"):
cli.generate_sdf_targets([str(source_path)], gz_check="never")
def test_strict_rejects_generator_warnings_before_writing(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_sdf_source(
source_path,
"\n".join(
[
"return {",
" 'xml': '<sdf version=\"1.12\"><model name=\"sample\"><link name=\"base_link\" /></model></sdf>',",
" 'warnings': ['Unverified plugin'],",
"}",
]
),
)
with self.assertRaisesRegex(SdfSourceError, "generator_warning"):
cli.generate_sdf_targets([str(source_path)], gz_check="never", strict=True)
self.assertFalse(source_path.with_suffix(".sdf").exists())
def test_gz_check_required_failure_prevents_write(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_sdf_source(
source_path,
"return '<sdf version=\"1.12\"><model name=\"sample\"><link name=\"base_link\" /></model></sdf>'",
)
with mock.patch("sdf.external.shutil.which", return_value=None):
with self.assertRaisesRegex(SdfSourceError, "gz_check_unavailable"):
cli.generate_sdf_targets([str(source_path)], gz_check="required")
self.assertFalse(source_path.with_suffix(".sdf").exists())
def test_rejects_invalid_output_suffix(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_sdf_source(source_path, "return '<sdf version=\"1.12\"><model name=\"sample\" /></sdf>'")
with self.assertRaisesRegex(ValueError, "must end in .sdf"):
cli.generate_sdf_targets([f"{source_path}={Path(tempdir) / 'sample.xml'}"])
def test_rejects_duplicate_output_paths(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-sdf-") as tempdir:
first_path = Path(tempdir) / "first.py"
second_path = Path(tempdir) / "second.py"
output_path = Path(tempdir) / "same.sdf"
_write_sdf_source(first_path, "return '<sdf version=\"1.12\"><model name=\"first\"><link name=\"base\" /></model></sdf>'")
_write_sdf_source(second_path, "return '<sdf version=\"1.12\"><model name=\"second\"><link name=\"base\" /></model></sdf>'")
with self.assertRaisesRegex(ValueError, "used more than once"):
cli.generate_sdf_targets([f"{first_path}={output_path}", f"{second_path}={output_path}"])
if __name__ == "__main__":
unittest.main()
@@ -1,294 +0,0 @@
import tempfile
import unittest
from pathlib import Path
from sdf.source import SdfSourceError, parse_sdf_xml, read_sdf_source
class SdfSourceTests(unittest.TestCase):
def setUp(self) -> None:
self._tempdir = tempfile.TemporaryDirectory(prefix="tmp-sdf-source-")
self.temp_root = Path(self._tempdir.name)
def tearDown(self) -> None:
self._tempdir.cleanup()
def _file_ref(self, name: str) -> str:
return (self.temp_root / f"{name}.sdf").resolve().as_posix()
def _write_mesh(self, name: str) -> Path:
mesh_path = self.temp_root / "meshes" / name
mesh_path.parent.mkdir(parents=True, exist_ok=True)
mesh_path.write_text("solid empty\nendsolid empty\n", encoding="utf-8")
return mesh_path
def _write_sdf(self, name: str, body: str) -> Path:
sdf_path = self.temp_root / f"{name}.sdf"
sdf_path.write_text(body.strip() + "\n", encoding="utf-8")
return sdf_path
def test_read_sdf_source_accepts_minimal_model(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
</model>
</sdf>
""",
)
source = read_sdf_source(source_path)
self.assertEqual(self._file_ref("robot"), source.file_ref)
self.assertEqual("1.12", source.version)
self.assertEqual(("sample",), source.model_names)
self.assertEqual((), source.world_names)
self.assertEqual(("base_link",), source.links)
self.assertEqual(0, len(source.joints))
def test_parse_sdf_xml_validates_without_existing_output_file(self) -> None:
output_path = self.temp_root / "generated" / "robot.sdf"
source = parse_sdf_xml(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
</model>
</sdf>
""",
source_path=output_path,
base_dir=output_path.parent,
)
self.assertEqual(output_path.resolve().as_posix(), source.file_ref)
self.assertEqual(("sample",), source.model_names)
self.assertFalse(output_path.exists())
def test_read_sdf_source_accepts_local_mesh_references(self) -> None:
mesh_path = self._write_mesh("base.stl")
source_path = self._write_sdf(
"robot",
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<visual name="base_visual">
<geometry>
<mesh>
<uri>meshes/base.stl</uri>
</mesh>
</geometry>
</visual>
<collision name="base_collision">
<geometry>
<mesh>
<uri>meshes/base.stl</uri>
</mesh>
</geometry>
</collision>
</link>
</model>
</sdf>
""",
)
source = read_sdf_source(source_path)
self.assertEqual((mesh_path.resolve(), mesh_path.resolve()), source.mesh_paths)
self.assertEqual((mesh_path.resolve(),), source.visual_mesh_paths)
self.assertEqual((mesh_path.resolve(),), source.collision_mesh_paths)
def test_read_sdf_source_accepts_external_mesh_uri_without_resolution(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<visual name="base_visual">
<geometry>
<mesh>
<uri>model://sample/meshes/base.dae</uri>
</mesh>
</geometry>
</visual>
</link>
</model>
</sdf>
""",
)
source = read_sdf_source(source_path)
self.assertEqual((), source.mesh_paths)
def test_read_sdf_source_accepts_world_models(self) -> None:
source_path = self._write_sdf(
"world",
"""
<sdf version="1.12">
<world name="sample_world">
<model name="sample">
<link name="base_link" />
</model>
</world>
</sdf>
""",
)
source = read_sdf_source(source_path)
self.assertEqual(("sample_world",), source.world_names)
self.assertEqual(("sample",), source.model_names)
def test_read_sdf_source_accepts_world_only_scene(self) -> None:
source_path = self._write_sdf(
"world_only",
"""
<sdf version="1.12">
<world name="sample_world">
<include>
<uri>model://sun</uri>
</include>
</world>
</sdf>
""",
)
source = read_sdf_source(source_path)
self.assertEqual(("sample_world",), source.world_names)
self.assertEqual((), source.model_names)
def test_read_sdf_source_rejects_missing_root(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<model name="sample">
<link name="base_link" />
</model>
""",
)
with self.assertRaisesRegex(SdfSourceError, "root element must be <sdf>"):
read_sdf_source(source_path)
def test_read_sdf_source_rejects_missing_version(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<sdf>
<model name="sample">
<link name="base_link" />
</model>
</sdf>
""",
)
with self.assertRaisesRegex(SdfSourceError, "version is required"):
read_sdf_source(source_path)
def test_read_sdf_source_rejects_missing_model_name(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<sdf version="1.12">
<model>
<link name="base_link" />
</model>
</sdf>
""",
)
with self.assertRaisesRegex(SdfSourceError, "model name is required"):
read_sdf_source(source_path)
def test_read_sdf_source_rejects_duplicate_links(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
<link name="base_link" />
</model>
</sdf>
""",
)
with self.assertRaisesRegex(SdfSourceError, "duplicates"):
read_sdf_source(source_path)
def test_read_sdf_source_rejects_missing_joint_child_link(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
<link name="arm_link" />
<joint name="base_to_arm" type="revolute">
<parent>base_link</parent>
<child>missing_link</child>
</joint>
</model>
</sdf>
""",
)
with self.assertRaisesRegex(SdfSourceError, "missing link"):
read_sdf_source(source_path)
def test_read_sdf_source_rejects_duplicate_joints(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
<link name="arm_link" />
<joint name="base_to_arm" type="fixed">
<parent>base_link</parent>
<child>arm_link</child>
</joint>
<joint name="base_to_arm" type="fixed">
<parent>base_link</parent>
<child>arm_link</child>
</joint>
</model>
</sdf>
""",
)
with self.assertRaisesRegex(SdfSourceError, "duplicates"):
read_sdf_source(source_path)
def test_read_sdf_source_rejects_missing_local_mesh(self) -> None:
source_path = self._write_sdf(
"robot",
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<visual name="base_visual">
<geometry>
<mesh>
<uri>meshes/missing.stl</uri>
</mesh>
</geometry>
</visual>
</link>
</model>
</sdf>
""",
)
with self.assertRaisesRegex(SdfSourceError, "missing mesh file"):
read_sdf_source(source_path)
if __name__ == "__main__":
unittest.main()
@@ -1,338 +0,0 @@
import tempfile
import unittest
from pathlib import Path
from sdf.validation import validate_sdf_xml
class SdfValidationTests(unittest.TestCase):
def setUp(self) -> None:
self._tempdir = tempfile.TemporaryDirectory(prefix="tmp-sdf-validation-")
self.temp_root = Path(self._tempdir.name)
def tearDown(self) -> None:
self._tempdir.cleanup()
def _validate(self, body: str):
return validate_sdf_xml(
body.strip(),
source_path=self.temp_root / "generated.sdf",
base_dir=self.temp_root,
)
def _error_codes(self, body: str) -> set[str]:
return {finding.code for finding in self._validate(body).errors}
def _warning_codes(self, body: str) -> set[str]:
return {finding.code for finding in self._validate(body).warnings}
def test_valid_minimal_model_passes(self) -> None:
result = self._validate(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
</model>
</sdf>
"""
)
self.assertEqual([], result.errors)
def test_valid_world_only_scene_passes(self) -> None:
result = self._validate(
"""
<sdf version="1.12">
<world name="sample_world">
<include><uri>model://sun</uri></include>
<light name="key" type="directional" />
</world>
</sdf>
"""
)
self.assertEqual([], result.errors)
def test_malformed_pose_length_fails(self) -> None:
self.assertIn(
"invalid_numeric_vector",
self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<pose>1 2 3</pose>
<link name="base_link" />
</model>
</sdf>
"""
),
)
def test_quaternion_pose_zero_norm_fails_and_non_unit_warns(self) -> None:
self.assertIn(
"zero_quaternion",
self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<pose rotation_format="quat_xyzw">0 0 0 0 0 0 0</pose>
<link name="base_link" />
</model>
</sdf>
"""
),
)
self.assertIn(
"non_unit_quaternion",
self._warning_codes(
"""
<sdf version="1.12">
<model name="sample">
<pose rotation_format="quat_xyzw">0 0 0 0 0 0 2</pose>
<link name="base_link" />
</model>
</sdf>
"""
),
)
def test_nontrivial_pose_without_relative_to_warns(self) -> None:
self.assertIn(
"pose_missing_relative_to",
self._warning_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<pose>1 0 0 0 0 0</pose>
</link>
</model>
</sdf>
"""
),
)
def test_frame_cycle_fails(self) -> None:
self.assertIn(
"frame_cycle",
self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<frame name="a" attached_to="b" />
<frame name="b" attached_to="a" />
<link name="base_link" />
</model>
</sdf>
"""
),
)
def test_joint_type_and_world_child_are_checked(self) -> None:
illegal_type_errors = self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
<link name="arm_link" />
<joint name="bad" type="hinge">
<parent>base_link</parent>
<child>arm_link</child>
</joint>
</model>
</sdf>
"""
)
self.assertIn("unknown_joint_type", illegal_type_errors)
world_child_errors = self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
<joint name="bad" type="fixed">
<parent>base_link</parent>
<child>world</child>
</joint>
</model>
</sdf>
"""
)
self.assertIn("invalid_joint_child", world_child_errors)
def test_axis_zero_fails_and_non_unit_warns(self) -> None:
zero_errors = self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
<link name="arm_link" />
<joint name="bad" type="revolute">
<parent>base_link</parent>
<child>arm_link</child>
<axis><xyz>0 0 0</xyz></axis>
</joint>
</model>
</sdf>
"""
)
self.assertIn("zero_axis", zero_errors)
warnings = self._warning_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
<link name="arm_link" />
<joint name="warns" type="revolute">
<parent>base_link</parent>
<child>arm_link</child>
<axis><xyz>0 0 2</xyz></axis>
</joint>
</model>
</sdf>
"""
)
self.assertIn("non_unit_axis", warnings)
def test_invalid_primitive_dimensions_fail(self) -> None:
errors = self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<visual name="base_visual">
<geometry><box><size>1 0 1</size></box></geometry>
</visual>
</link>
</model>
</sdf>
"""
)
self.assertIn("invalid_dimension", errors)
def test_visual_without_geometry_fails(self) -> None:
self.assertIn(
"invalid_geometry_count",
self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<visual name="base_visual" />
</link>
</model>
</sdf>
"""
),
)
def test_mesh_uri_resolution_distinguishes_local_and_external(self) -> None:
local_errors = self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<visual name="base_visual">
<geometry><mesh><uri>meshes/missing.stl</uri></mesh></geometry>
</visual>
</link>
</model>
</sdf>
"""
)
self.assertIn("missing_mesh_file", local_errors)
external_result = self._validate(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<visual name="base_visual">
<geometry><mesh><uri>package://sample/meshes/base.dae</uri></mesh></geometry>
</visual>
</link>
</model>
</sdf>
"""
)
self.assertEqual([], external_result.errors)
def test_invalid_inertial_values_fail(self) -> None:
mass_errors = self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<inertial><mass>0</mass></inertial>
</link>
</model>
</sdf>
"""
)
self.assertIn("invalid_mass", mass_errors)
inertia_errors = self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<inertial>
<mass>1</mass>
<inertia>
<ixx>1</ixx><iyy>1</iyy><izz>-1</izz>
<ixy>0</ixy><ixz>0</ixz><iyz>0</iyz>
</inertia>
</inertial>
</link>
</model>
</sdf>
"""
)
self.assertIn("invalid_inertia_matrix", inertia_errors)
def test_sensor_plugin_and_include_required_fields(self) -> None:
errors = self._error_codes(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link">
<sensor name="camera" />
</link>
<plugin name="controller" />
<include />
</model>
</sdf>
"""
)
self.assertIn("missing_sensor_type", errors)
self.assertIn("missing_plugin_filename", errors)
self.assertIn("missing_child_text", errors)
def test_plugin_contents_are_static_metadata_not_motion_contracts(self) -> None:
result = self._validate(
"""
<sdf version="1.12">
<model name="sample">
<link name="base_link" />
<link name="door_link" />
<joint name="door_joint" type="fixed">
<parent>base_link</parent>
<child>door_link</child>
</joint>
<plugin name="controller" filename="gz-sim-custom-controller-system">
<custom_parameter>opaque</custom_parameter>
<nested_block invalid_for_validator="but_plugin_owned" />
</plugin>
</model>
</sdf>
"""
)
self.assertEqual([], result.errors)
if __name__ == "__main__":
unittest.main()
@@ -1,269 +0,0 @@
from __future__ import annotations
from pathlib import Path
import tempfile
import unittest
from tests.python.support.paths import add_repo_path
add_repo_path("skills/srdf/scripts")
from srdf import cli
SAMPLE_URDF = """\
<robot name="sample">
<link name="base"/>
<link name="shoulder_link"/>
<link name="wrist"/>
<link name="tool"/>
<joint name="shoulder" type="revolute">
<parent link="base"/>
<child link="shoulder_link"/>
<limit lower="-1" upper="1" effort="1" velocity="1"/>
</joint>
<joint name="elbow" type="revolute">
<parent link="shoulder_link"/>
<child link="wrist"/>
<limit lower="-1" upper="1" effort="1" velocity="1"/>
</joint>
<joint name="wrist_tool" type="fixed">
<parent link="wrist"/>
<child link="tool"/>
</joint>
</robot>
"""
SAMPLE_SRDF = """\
<robot name="sample" xmlns:tcad="https://text-to-cad.dev/srdf">
<tcad:urdf path="robot.urdf"/>
<group name="arm">
<joint name="shoulder"/>
<joint name="elbow"/>
</group>
<group name="gripper">
<link name="tool"/>
</group>
<end_effector name="tcp" parent_link="wrist" group="gripper" parent_group="arm"/>
<group_state name="home" group="arm">
<joint name="shoulder" value="0"/>
<joint name="elbow" value="0"/>
</group_state>
<disable_collisions link1="wrist" link2="tool" reason="Adjacent"/>
</robot>
"""
def write_source(path: Path, payload: object | None = None, *, function_name: str = "gen_srdf") -> None:
if payload is None:
payload = {"xml": SAMPLE_SRDF, "urdf": "robot.urdf"}
path.write_text(
f"def {function_name}():\n"
f" return {payload!r}\n",
encoding="utf-8",
)
class SrdfCliTests(unittest.TestCase):
def test_writes_sibling_srdf_without_hidden_artifact(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "robot_srdf.py"
write_source(source_path)
self.assertEqual(0, cli.generate_srdf_targets([str(source_path)]))
output_path = root / "robot_srdf.srdf"
self.assertTrue(output_path.is_file())
output_text = output_path.read_text(encoding="utf-8")
self.assertIn("tcad:urdf", output_text)
self.assertIn("path=\"robot.urdf\"", output_text)
self.assertIn("<group name=\"arm\">", output_text)
self.assertFalse((root / ".robot.urdf").exists())
def test_injects_tcad_urdf_link(self) -> None:
srdf = """\
<robot name="sample">
<group name="arm">
<chain base_link="base" tip_link="tool"/>
</group>
</robot>
"""
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "robot_srdf.py"
write_source(source_path, {"xml": srdf, "urdf": "robot.urdf"})
self.assertEqual(0, cli.generate_srdf_targets([str(source_path)]))
output_text = (root / "robot_srdf.srdf").read_text(encoding="utf-8")
self.assertIn("tcad:urdf", output_text)
self.assertIn("path=\"robot.urdf\"", output_text)
def test_updates_legacy_explorer_urdf_link(self) -> None:
legacy_srdf = SAMPLE_SRDF.replace(
'xmlns:tcad="https://text-to-cad.dev/srdf"',
'xmlns:explorer="https://text-to-cad.dev/explorer"',
).replace("tcad:urdf", "explorer:urdf")
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "robot_srdf.py"
write_source(source_path, {"xml": legacy_srdf, "urdf": "robot.urdf"})
self.assertEqual(0, cli.generate_srdf_targets([str(source_path)]))
output_text = (root / "robot_srdf.srdf").read_text(encoding="utf-8")
self.assertIn("tcad:urdf", output_text)
self.assertNotIn("explorer:urdf", output_text)
self.assertIn("path=\"robot.urdf\"", output_text)
def test_writes_srdf_from_element_root(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "robot_srdf.py"
source_path.write_text(
"\n".join(
[
"import xml.etree.ElementTree as ET",
"",
"def gen_srdf():",
" robot = ET.Element('robot', {'name': 'sample'})",
" group = ET.SubElement(robot, 'group', {'name': 'arm'})",
" ET.SubElement(group, 'joint', {'name': 'shoulder'})",
" ET.SubElement(group, 'joint', {'name': 'elbow'})",
" return {'xml': robot, 'urdf': 'robot.urdf'}",
"",
]
),
encoding="utf-8",
)
self.assertEqual(0, cli.generate_srdf_targets([str(source_path)]))
output_text = (root / "robot_srdf.srdf").read_text(encoding="utf-8")
self.assertTrue(output_text.startswith('<?xml version="1.0"?>\n'))
self.assertIn("tcad:urdf", output_text)
self.assertIn("<group name=\"arm\">", output_text)
self.assertIn("<joint name=\"shoulder\" />", output_text)
def test_supports_output_option(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "source.py"
output_path = root / "out" / "robot.srdf"
write_source(source_path)
self.assertEqual(0, cli.generate_srdf_targets([str(source_path)], output=str(output_path)))
self.assertTrue(output_path.is_file())
def test_supports_source_output_pairs(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "source.py"
output_path = root / "out" / "robot.srdf"
write_source(source_path)
self.assertEqual(0, cli.generate_srdf_targets([f"{source_path}={output_path}"]))
self.assertTrue(output_path.is_file())
def test_rejects_missing_gen_srdf(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "source.py"
write_source(source_path, function_name="gen_urdf")
with self.assertRaisesRegex(RuntimeError, "gen_srdf"):
cli.generate_srdf_targets([str(source_path)])
def test_rejects_missing_xml_or_urdf(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
source_path = root / "source.py"
write_source(source_path, {"xml": SAMPLE_SRDF})
with self.assertRaisesRegex(TypeError, "urdf"):
cli.generate_srdf_targets([str(source_path)])
write_source(source_path, {"urdf": "robot.urdf"})
with self.assertRaisesRegex(TypeError, "xml"):
cli.generate_srdf_targets([str(source_path)])
def test_rejects_invalid_xml_or_urdf_reference(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "source.py"
write_source(source_path, {"xml": "<sdf/>", "urdf": "robot.urdf"})
with self.assertRaisesRegex(Exception, "root element must be <robot>"):
cli.generate_srdf_targets([str(source_path)])
write_source(source_path, {"xml": SAMPLE_SRDF, "urdf": "missing.urdf"})
with self.assertRaisesRegex(FileNotFoundError, "urdf file does not exist"):
cli.generate_srdf_targets([str(source_path)])
def test_rejects_group_state_values_outside_group_or_limits(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "source.py"
write_source(
source_path,
{
"xml": SAMPLE_SRDF.replace('<joint name="shoulder" value="0"/>', '<joint name="shoulder" value="2"/>'),
"urdf": "robot.urdf",
},
)
with self.assertRaisesRegex(Exception, "above its URDF upper limit"):
cli.generate_srdf_targets([str(source_path)])
write_source(
source_path,
{
"xml": SAMPLE_SRDF.replace('<joint name="shoulder" value="0"/>', '<joint name="wrist_tool" value="0"/>'),
"urdf": "robot.urdf",
},
)
with self.assertRaisesRegex(Exception, "not in group"):
cli.generate_srdf_targets([str(source_path)])
def test_rejects_overlapping_end_effector_group(self) -> None:
srdf = SAMPLE_SRDF.replace('<link name="tool"/>', '<link name="wrist"/>')
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "source.py"
write_source(source_path, {"xml": srdf, "urdf": "robot.urdf"})
with self.assertRaisesRegex(Exception, "shares link"):
cli.generate_srdf_targets([str(source_path)])
def test_rejects_disabled_collision_without_reason(self) -> None:
srdf = SAMPLE_SRDF.replace(' reason="Adjacent"', "")
with tempfile.TemporaryDirectory(prefix="tmp-gen-srdf-") as tempdir:
root = Path(tempdir)
(root / "robot.urdf").write_text(SAMPLE_URDF, encoding="utf-8")
source_path = root / "source.py"
write_source(source_path, {"xml": srdf, "urdf": "robot.urdf"})
with self.assertRaisesRegex(Exception, "requires a reason"):
cli.generate_srdf_targets([str(source_path)])
def test_output_must_be_srdf(self) -> None:
with self.assertRaisesRegex(ValueError, "must end in .srdf"):
cli.generate_srdf_targets(["source.py=out.xml"])
if __name__ == "__main__":
unittest.main()
@@ -1,70 +0,0 @@
from __future__ import annotations
from pathlib import Path
import tempfile
import unittest
from srdf.source import SrdfSourceError, read_srdf_source
SAMPLE_SRDF = """\
<robot name="sample" xmlns:tcad="https://text-to-cad.dev/srdf">
<tcad:urdf path="robot.urdf"/>
<group name="arm">
<joint name="shoulder"/>
<joint name="elbow"/>
</group>
<group name="gripper">
<link name="tool"/>
</group>
<end_effector name="tcp" parent_link="wrist" group="gripper" parent_group="arm"/>
<group_state name="home" group="arm">
<joint name="shoulder" value="0"/>
<joint name="elbow" value="1.57"/>
</group_state>
<disable_collisions link1="base" link2="shoulder_link" reason="Adjacent"/>
</robot>
"""
LEGACY_SAMPLE_SRDF = SAMPLE_SRDF.replace(
'xmlns:tcad="https://text-to-cad.dev/srdf"',
'xmlns:explorer="https://text-to-cad.dev/explorer"',
).replace("tcad:urdf", "explorer:urdf")
class SrdfSourceTests(unittest.TestCase):
def test_reads_moveit2_srdf_inventory(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-srdf-source-") as tempdir:
srdf_path = Path(tempdir) / "robot.srdf"
srdf_path.write_text(SAMPLE_SRDF, encoding="utf-8")
source = read_srdf_source(srdf_path)
self.assertEqual(source.robot_name, "sample")
self.assertEqual(source.urdf_ref, "robot.urdf")
self.assertEqual(source.planning_groups[0].name, "arm")
self.assertEqual(source.planning_groups[0].joint_names, ("shoulder", "elbow"))
self.assertEqual(source.end_effectors[0].name, "tcp")
self.assertEqual(source.group_states[0].joint_values_by_name_rad["elbow"], 1.57)
self.assertEqual(source.disabled_collision_pairs[0].reason, "Adjacent")
def test_reads_legacy_explorer_urdf_metadata(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-srdf-source-") as tempdir:
srdf_path = Path(tempdir) / "robot.srdf"
srdf_path.write_text(LEGACY_SAMPLE_SRDF, encoding="utf-8")
source = read_srdf_source(srdf_path)
self.assertEqual(source.urdf_ref, "robot.urdf")
def test_rejects_non_robot_root(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-srdf-source-") as tempdir:
srdf_path = Path(tempdir) / "robot.srdf"
srdf_path.write_text("<sdf/>", encoding="utf-8")
with self.assertRaisesRegex(SrdfSourceError, "root element must be <robot>"):
read_srdf_source(srdf_path)
if __name__ == "__main__":
unittest.main()
@@ -1,264 +0,0 @@
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from tests.python.support.paths import add_repo_path
add_repo_path("skills/urdf/scripts")
from urdf import cli
from urdf.source import UrdfSourceError
def _write_urdf_source(path: Path, body: str) -> None:
path.write_text(
"\n".join(
[
"def gen_urdf():",
*[f" {line}" for line in body.splitlines()],
"",
]
),
encoding="utf-8",
)
def _strip_cadpy_metadata_comments(text: str) -> str:
return "\n".join(
line for line in text.splitlines() if not line.startswith("<!-- cadpy:")
) + "\n"
class UrdfCliTests(unittest.TestCase):
def test_requires_explicit_target(self) -> None:
with self.assertRaises(SystemExit) as cm:
cli.main([])
self.assertEqual(2, cm.exception.code)
def test_rejects_summary_option(self) -> None:
with self.assertRaises(SystemExit) as cm:
cli.main(["sample_robot.py", "--summary"])
self.assertEqual(2, cm.exception.code)
def test_rejects_validate_options(self) -> None:
for option in ("--validate", "--no-validate"):
with self.subTest(option=option):
with self.assertRaises(SystemExit) as cm:
cli.main(["sample_robot.py", option])
self.assertEqual(2, cm.exception.code)
def test_passes_targets_and_output(self) -> None:
with mock.patch.object(cli, "generate_urdf_targets", return_value=0) as generate:
self.assertEqual(0, cli.main(["sample_robot.py", "-o", "sample_robot.urdf"]))
generate.assert_called_once_with(["sample_robot.py"], output="sample_robot.urdf")
def test_rejects_output_with_pair_target(self) -> None:
with self.assertRaises(SystemExit) as cm:
cli.main(["sample_robot.py=sample_robot.urdf", "-o", "other.urdf"])
self.assertEqual(2, cm.exception.code)
def test_generates_default_sibling_output_from_xml_string(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_urdf_source(
source_path,
"return '<robot name=\"sample\"><link name=\"base_link\" /></robot>'",
)
self.assertEqual(0, cli.generate_urdf_targets([str(source_path)]))
output_text = source_path.with_suffix(".urdf").read_text(encoding="utf-8")
self.assertIn("<!-- cadpy:sourcePath=", output_text)
self.assertEqual(
'<robot name="sample"><link name="base_link" /></robot>\n',
_strip_cadpy_metadata_comments(output_text),
)
def test_generates_default_sibling_output_from_element_root(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
source_path.write_text(
"\n".join(
[
"import xml.etree.ElementTree as ET",
"",
"def gen_urdf():",
" robot = ET.Element('robot', {'name': 'sample'})",
" ET.SubElement(robot, 'link', {'name': 'base_link'})",
" return robot",
"",
]
),
encoding="utf-8",
)
self.assertEqual(0, cli.generate_urdf_targets([str(source_path)]))
output_text = source_path.with_suffix(".urdf").read_text(encoding="utf-8")
self.assertIn("<!-- cadpy:sourcePath=", output_text)
self.assertEqual(
'<?xml version="1.0"?>\n<robot name="sample">\n <link name="base_link" />\n</robot>\n',
_strip_cadpy_metadata_comments(output_text),
)
def test_generates_envelope_output_from_element_root(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
source_path.write_text(
"\n".join(
[
"import xml.etree.ElementTree as ET",
"",
"def gen_urdf():",
" robot = ET.Element('robot', {'name': 'sample'})",
" ET.SubElement(robot, 'link', {'name': 'base_link'})",
" return {'xml': robot}",
"",
]
),
encoding="utf-8",
)
self.assertEqual(0, cli.generate_urdf_targets([str(source_path)]))
self.assertIn(
'<link name="base_link" />',
source_path.with_suffix(".urdf").read_text(encoding="utf-8"),
)
def test_generates_output_override_for_single_target(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
output_path = Path(tempdir) / "custom" / "robot.urdf"
_write_urdf_source(
source_path,
"return '<robot name=\"sample\"><link name=\"base_link\" /></robot>'",
)
self.assertEqual(0, cli.generate_urdf_targets([str(source_path)], output=str(output_path)))
self.assertTrue(output_path.exists())
self.assertFalse(source_path.with_suffix(".urdf").exists())
def test_validates_generated_output(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
output_path = Path(tempdir) / "custom" / "robot.urdf"
_write_urdf_source(
source_path,
"return '<robot name=\"sample\"><link name=\"base_link\" /></robot>'",
)
with mock.patch.object(cli, "read_urdf_source") as read_urdf_source:
self.assertEqual(0, cli.generate_urdf_targets([str(source_path)], output=str(output_path)))
read_urdf_source.assert_called_once_with(output_path.resolve())
def test_rejects_invalid_generated_output(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
output_path = Path(tempdir) / "custom" / "robot.urdf"
_write_urdf_source(source_path, "return '<not-robot />'")
with self.assertRaisesRegex(UrdfSourceError, "root element must be <robot>"):
cli.generate_urdf_targets([str(source_path)], output=str(output_path))
def test_generates_mixed_plain_and_paired_targets(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
root = Path(tempdir)
first_path = root / "first.py"
second_path = root / "second.py"
second_output = root / "custom" / "second.urdf"
_write_urdf_source(first_path, "return '<robot name=\"first\"><link name=\"base\" /></robot>'")
_write_urdf_source(second_path, "return '<robot name=\"second\"><link name=\"base\" /></robot>'")
self.assertEqual(0, cli.generate_urdf_targets([str(first_path), f"{second_path}={second_output}"]))
self.assertTrue(first_path.with_suffix(".urdf").exists())
self.assertTrue(second_output.exists())
self.assertFalse(second_path.with_suffix(".urdf").exists())
def test_rejects_legacy_urdf_output_field(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_urdf_source(
source_path,
"\n".join(
[
"return {",
" 'xml': '<robot name=\"sample\"><link name=\"base_link\" /></robot>',",
" 'urdf_output': 'legacy/ignored.urdf',",
"}",
]
),
)
with self.assertRaisesRegex(TypeError, "unsupported field\\(s\\): urdf_output"):
cli.generate_urdf_targets([str(source_path)])
def test_rejects_legacy_validate_field(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_urdf_source(
source_path,
"\n".join(
[
"return {",
" 'xml': '<robot name=\"sample\"><link name=\"base_link\" /></robot>',",
" 'validate': False,",
"}",
]
),
)
with self.assertRaisesRegex(TypeError, "unsupported field\\(s\\): validate"):
cli.generate_urdf_targets([str(source_path)])
def test_rejects_explorer_metadata_field(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
output_path = Path(tempdir) / "custom" / "sample_robot.urdf"
_write_urdf_source(
source_path,
"\n".join(
[
"return {",
" 'xml': '<robot name=\"sample\"><link name=\"base_link\" /></robot>',",
" 'explorer_metadata': {",
" 'schemaVersion': 1,",
" 'kind': 'example-urdf-consumer',",
" 'defaultJoints': {'joint_1': 90},",
" },",
"}",
]
),
)
with self.assertRaisesRegex(TypeError, "unsupported field\\(s\\): explorer_metadata"):
cli.generate_urdf_targets([f"{source_path}={output_path}"])
self.assertFalse(output_path.exists())
def test_rejects_invalid_output_suffix(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
source_path = Path(tempdir) / "sample_robot.py"
_write_urdf_source(source_path, "return '<robot name=\"sample\"><link name=\"base\" /></robot>'")
with self.assertRaisesRegex(ValueError, "must end in .urdf"):
cli.generate_urdf_targets([f"{source_path}={Path(tempdir) / 'sample.xml'}"])
def test_rejects_duplicate_output_paths(self) -> None:
with tempfile.TemporaryDirectory(prefix="tmp-urdf-") as tempdir:
first_path = Path(tempdir) / "first.py"
second_path = Path(tempdir) / "second.py"
output_path = Path(tempdir) / "same.urdf"
_write_urdf_source(first_path, "return '<robot name=\"first\"><link name=\"base\" /></robot>'")
_write_urdf_source(second_path, "return '<robot name=\"second\"><link name=\"base\" /></robot>'")
with self.assertRaisesRegex(ValueError, "used more than once"):
cli.generate_urdf_targets([f"{first_path}={output_path}", f"{second_path}={output_path}"])
if __name__ == "__main__":
unittest.main()
@@ -1,593 +0,0 @@
import tempfile
import unittest
import warnings
from pathlib import Path
from urdf.source import (
MeshUriKind,
UrdfSourceError,
UrdfSourceWarning,
classify_mesh_uri,
read_urdf_source,
resolve_mesh_uri,
)
class UrdfSourceTests(unittest.TestCase):
def setUp(self) -> None:
self._tempdir = tempfile.TemporaryDirectory(prefix="tmp-urdf-source-")
self.temp_root = Path(self._tempdir.name)
def tearDown(self) -> None:
self._tempdir.cleanup()
def _file_ref(self, name: str) -> str:
return (self.temp_root / f"{name}.urdf").resolve().as_posix()
def _write_mesh(self, name: str) -> Path:
mesh_path = self.temp_root / name if Path(name).suffix else self.temp_root / f"{name}.stl"
mesh_path.parent.mkdir(parents=True, exist_ok=True)
mesh_path.write_text("solid empty\nendsolid empty\n", encoding="utf-8")
return mesh_path
def _write_urdf(self, name: str, body: str) -> Path:
urdf_path = self.temp_root / f"{name}.urdf"
urdf_path.write_text(body.strip() + "\n", encoding="utf-8")
script_path = self.temp_root / f"{name}.py"
if not script_path.exists():
script_path.write_text(
"\n".join(
[
"def gen_step():",
" return {'instances': []}",
"",
"def gen_urdf():",
" return {'xml': ''}",
"",
]
),
encoding="utf-8",
)
return urdf_path
def test_read_urdf_source_accepts_valid_mesh_robot(self) -> None:
mesh_path = self._write_mesh("base")
source_path = self._write_urdf(
"robot",
f"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="{mesh_path.name}" scale="0.001 0.001 0.001" />
</geometry>
</visual>
</link>
</robot>
""",
)
source = read_urdf_source(source_path)
self.assertEqual(self._file_ref("robot"), source.file_ref)
self.assertEqual("sample-robot", source.robot_name)
self.assertEqual("base_link", source.root_link)
self.assertEqual(("base_link",), source.links)
self.assertEqual(0, len(source.joints))
self.assertEqual((mesh_path.resolve(),), source.mesh_paths)
self.assertEqual((mesh_path.resolve(),), source.visual_mesh_paths)
self.assertEqual((), source.collision_mesh_paths)
def test_read_urdf_source_accepts_collision_meshes(self) -> None:
visual_mesh_path = self._write_mesh("visual")
collision_mesh_path = self._write_mesh("collision")
source_path = self._write_urdf(
"robot",
f"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="{visual_mesh_path.name}" />
</geometry>
</visual>
<collision>
<origin xyz="0 0 0" rpy="0 0 0" />
<geometry>
<mesh filename="{collision_mesh_path.name}" />
</geometry>
</collision>
</link>
</robot>
""",
)
source = read_urdf_source(source_path)
self.assertEqual(
(visual_mesh_path.resolve(), collision_mesh_path.resolve()),
source.mesh_paths,
)
self.assertEqual((visual_mesh_path.resolve(),), source.visual_mesh_paths)
self.assertEqual((collision_mesh_path.resolve(),), source.collision_mesh_paths)
def test_read_urdf_source_accepts_primitive_collision_geometry(self) -> None:
visual_mesh_path = self._write_mesh("visual")
source_path = self._write_urdf(
"robot",
f"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="{visual_mesh_path.name}" />
</geometry>
</visual>
<collision>
<geometry>
<box size="0.1 0.2 0.3" />
</geometry>
</collision>
</link>
</robot>
""",
)
source = read_urdf_source(source_path)
self.assertEqual((visual_mesh_path.resolve(),), source.mesh_paths)
self.assertEqual((), source.collision_mesh_paths)
def test_read_urdf_source_accepts_primitive_visual_geometry(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<box size="0.1 0.2 0.3" />
</geometry>
</visual>
</link>
</robot>
""",
)
source = read_urdf_source(source_path)
self.assertEqual(("base_link",), source.links)
self.assertEqual((), source.mesh_paths)
def test_read_urdf_source_accepts_non_stl_mesh_formats(self) -> None:
visual_mesh_path = self._write_mesh("meshes/base.dae")
collision_mesh_path = self._write_mesh("meshes/base_collision.obj")
source_path = self._write_urdf(
"robot",
f"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="{visual_mesh_path.relative_to(self.temp_root).as_posix()}" />
</geometry>
</visual>
<collision>
<geometry>
<mesh filename="{collision_mesh_path.relative_to(self.temp_root).as_posix()}" />
</geometry>
</collision>
</link>
</robot>
""",
)
source = read_urdf_source(source_path)
self.assertEqual((visual_mesh_path.resolve(), collision_mesh_path.resolve()), source.mesh_paths)
def test_classifies_mesh_uris(self) -> None:
package_ref = classify_mesh_uri("package://robot_description/meshes/base.stl")
self.assertEqual(MeshUriKind.PACKAGE, package_ref.kind)
self.assertEqual("robot_description", package_ref.package_name)
self.assertEqual("meshes/base.stl", package_ref.package_path.as_posix())
self.assertEqual(MeshUriKind.REMOTE, classify_mesh_uri("https://example.com/base.stl").kind)
self.assertEqual(MeshUriKind.LOCAL_ABSOLUTE, classify_mesh_uri("/tmp/base.stl").kind)
self.assertEqual(MeshUriKind.LOCAL_RELATIVE, classify_mesh_uri("meshes/base.stl").kind)
def test_resolves_package_mesh_uri_with_package_map(self) -> None:
package_root = self.temp_root / "robot_description"
mesh_path = package_root / "meshes" / "base.stl"
mesh_path.parent.mkdir(parents=True)
mesh_path.write_text("solid empty\nendsolid empty\n", encoding="utf-8")
self.assertEqual(
mesh_path.resolve(),
resolve_mesh_uri(
"package://robot_description/meshes/base.stl",
package_map={"robot_description": package_root},
),
)
def test_read_urdf_source_accepts_unresolved_package_mesh_with_warning(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="package://robot_description/meshes/base.stl" />
</geometry>
</visual>
</link>
</robot>
""",
)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always", UrdfSourceWarning)
source = read_urdf_source(source_path)
self.assertEqual((), source.mesh_paths)
warning_messages = [str(warning.message) for warning in caught]
self.assertTrue(any("syntax is valid but was not resolved" in message for message in warning_messages))
def test_read_urdf_source_accepts_unresolved_remote_mesh_with_warning(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="https://example.com/meshes/base.glb" />
</geometry>
</visual>
</link>
</robot>
""",
)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always", UrdfSourceWarning)
source = read_urdf_source(source_path)
self.assertEqual((), source.mesh_paths)
warning_messages = [str(warning.message) for warning in caught]
self.assertTrue(any("is not a local mesh URI and was not resolved" in message for message in warning_messages))
def test_read_urdf_source_resolves_package_mesh_with_package_map(self) -> None:
package_root = self.temp_root / "robot_description"
mesh_path = package_root / "meshes" / "base.stl"
mesh_path.parent.mkdir(parents=True)
mesh_path.write_text("solid empty\nendsolid empty\n", encoding="utf-8")
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="package://robot_description/meshes/base.stl" />
</geometry>
</visual>
</link>
</robot>
""",
)
source = read_urdf_source(source_path, package_map={"robot_description": package_root})
self.assertEqual((mesh_path.resolve(),), source.mesh_paths)
def test_read_urdf_source_accepts_valid_inertial(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<inertial>
<origin xyz="0 0 0" rpy="0 0 0" />
<mass value="1.2" />
<inertia ixx="0.1" ixy="0" ixz="0" iyy="0.1" iyz="0" izz="0.1" />
</inertial>
</link>
</robot>
""",
)
source = read_urdf_source(source_path)
self.assertEqual(("base_link",), source.links)
def test_read_urdf_source_rejects_nonpositive_inertial_mass(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<inertial>
<mass value="0" />
<inertia ixx="0.1" ixy="0" ixz="0" iyy="0.1" iyz="0" izz="0.1" />
</inertial>
</link>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "mass must be positive"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_invalid_inertia_triangle(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<inertial>
<mass value="1" />
<inertia ixx="0.01" ixy="0" ixz="0" iyy="0.01" iyz="0" izz="1" />
</inertial>
</link>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "triangle"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_invalid_origin_vector(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link" />
<link name="arm_link" />
<joint name="base_to_arm" type="fixed">
<origin xyz="0 0" />
<parent link="base_link" />
<child link="arm_link" />
</joint>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "origin xyz must have 3 values"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_zero_joint_axis(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link" />
<link name="arm_link" />
<joint name="base_to_arm" type="revolute">
<parent link="base_link" />
<child link="arm_link" />
<axis xyz="0 0 0" />
<limit lower="-1" upper="1" />
</joint>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "axis must be nonzero"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_invalid_mesh_scale(self) -> None:
mesh_path = self._write_mesh("base.obj")
source_path = self._write_urdf(
"robot",
f"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="{mesh_path.name}" scale="1 0 1" />
</geometry>
</visual>
</link>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "scale values must be positive"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_invalid_primitive_dimensions(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<collision>
<geometry>
<sphere radius="0" />
</geometry>
</collision>
</link>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "sphere radius must be positive"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_missing_geometry_element(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<visual />
</link>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "visual requires <geometry>"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_duplicate_links(self) -> None:
mesh_path = self._write_mesh("base")
source_path = self._write_urdf(
"robot",
f"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="{mesh_path.name}" />
</geometry>
</visual>
</link>
<link name="base_link" />
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "duplicates"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_missing_mesh(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="does-not-exist.stl" />
</geometry>
</visual>
</link>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "missing mesh file"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_missing_collision_mesh(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link">
<collision>
<geometry>
<mesh filename="does-not-exist.stl" />
</geometry>
</collision>
</link>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "missing mesh file"):
read_urdf_source(source_path)
def test_read_urdf_source_accepts_prismatic_joint_with_limits(self) -> None:
mesh_path = self._write_mesh("base")
source_path = self._write_urdf(
"robot",
f"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="{mesh_path.name}" />
</geometry>
</visual>
</link>
<link name="arm_link" />
<joint name="base_to_arm" type="prismatic">
<parent link="base_link" />
<child link="arm_link" />
<limit lower="0" upper="0.05" effort="1" velocity="1" />
</joint>
</robot>
""",
)
source = read_urdf_source(source_path)
self.assertEqual("prismatic", source.joints[0].joint_type)
self.assertEqual(0.0, source.joints[0].min_value_deg)
self.assertEqual(0.05, source.joints[0].max_value_deg)
def test_read_urdf_source_rejects_reversed_joint_limits(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link" />
<link name="arm_link" />
<joint name="base_to_arm" type="prismatic">
<parent link="base_link" />
<child link="arm_link" />
<limit lower="1" upper="0" />
</joint>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "lower limit exceeds upper limit"):
read_urdf_source(source_path)
def test_read_urdf_source_rejects_unsupported_joint_type(self) -> None:
mesh_path = self._write_mesh("base")
source_path = self._write_urdf(
"robot",
f"""
<robot name="sample-robot">
<link name="base_link">
<visual>
<geometry>
<mesh filename="{mesh_path.name}" />
</geometry>
</visual>
</link>
<link name="arm_link" />
<joint name="base_to_arm" type="planar">
<parent link="base_link" />
<child link="arm_link" />
</joint>
</robot>
""",
)
with self.assertRaisesRegex(UrdfSourceError, "unsupported type"):
read_urdf_source(source_path)
def test_file_ref_ignores_neighbor_step_toml(self) -> None:
source_path = self._write_urdf(
"robot",
"""
<robot name="sample-robot">
<link name="base_link" />
</robot>
""",
)
stale_path = self.temp_root / "robot.step.toml"
stale_path.write_text(
"\n".join(
[
'kind = "part"',
'source = "robot.urdf"',
]
)
+ "\n",
encoding="utf-8",
)
source = read_urdf_source(source_path)
self.assertEqual(source_path.resolve().as_posix(), source.file_ref)
if __name__ == "__main__":
unittest.main()