235 lines
8.6 KiB
Python
235 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from core.software_contracts import SoftwareContractError, get_contract
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ADAPTER = ROOT / "windows-node" / "adapters" / "freecad.model.author@v1"
|
|
sys.path.insert(0, str(ADAPTER))
|
|
SPEC = importlib.util.spec_from_file_location("freecad_launcher", ADAPTER / "worker.py")
|
|
assert SPEC and SPEC.loader
|
|
launcher = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(launcher)
|
|
from recipe import RecipeError, normalize_model
|
|
|
|
|
|
def model() -> dict[str, object]:
|
|
return {
|
|
"type": "recipe",
|
|
"recipe_version": 1,
|
|
"title": "受控模型",
|
|
"unit": "mm",
|
|
"parameters": {"length": 20},
|
|
"steps": [
|
|
{
|
|
"id": "base",
|
|
"op": "box",
|
|
"length": {"parameter": "length"},
|
|
"width": 10,
|
|
"height": 5,
|
|
},
|
|
{"id": "hole", "op": "cylinder", "radius": 2, "height": 5},
|
|
{"id": "result", "op": "cut", "target": "base", "tools": ["hole"]},
|
|
],
|
|
"result": "result",
|
|
}
|
|
|
|
|
|
class FreeCadContractTests(unittest.TestCase):
|
|
def test_contract_accepts_parameterized_minimal_recipe(self) -> None:
|
|
request = {
|
|
"schema_version": 1,
|
|
"inputs": [],
|
|
"operation": {"model": model()},
|
|
"outputs": [],
|
|
}
|
|
normalized, digest = get_contract("freecad.model.author@v1").normalize_request(
|
|
request
|
|
)
|
|
self.assertEqual(normalized["operation"]["model"]["result"], "result")
|
|
self.assertEqual(len(digest), 64)
|
|
contract = get_contract("freecad.model.author@v1")
|
|
self.assertEqual(contract.workspace.state_output, "project")
|
|
self.assertEqual(contract.workspace.preview_outputs, ("preview",))
|
|
self.assertEqual(
|
|
set(contract.expected_outputs(normalized)),
|
|
{
|
|
"project",
|
|
"geometry_step",
|
|
"preview",
|
|
"model_recipe",
|
|
"model_manifest",
|
|
"provenance",
|
|
},
|
|
)
|
|
|
|
def test_contract_rejects_raw_expression_strings_and_fcstd_inputs(self) -> None:
|
|
request = {
|
|
"schema_version": 1,
|
|
"inputs": [],
|
|
"operation": {"model": model()},
|
|
"outputs": [],
|
|
}
|
|
request["operation"]["model"]["steps"][0]["length"] = "length * 2"
|
|
with self.assertRaises(SoftwareContractError):
|
|
get_contract("freecad.model.author@v1").normalize_request(request)
|
|
self.assertEqual(
|
|
get_contract("freecad.model.author@v1").input_policy["suffixes"],
|
|
[".step", ".stp"],
|
|
)
|
|
|
|
|
|
class FreeCadRecipeTests(unittest.TestCase):
|
|
def test_parameter_references_are_resolved_to_numbers(self) -> None:
|
|
resolved = normalize_model(model(), input_bound=False)
|
|
self.assertEqual(resolved["steps"][0]["length"], 20.0)
|
|
self.assertNotIn("Spreadsheet", json.dumps(resolved))
|
|
|
|
def test_forward_reference_and_unbound_step_are_rejected(self) -> None:
|
|
value = model()
|
|
value["steps"][0] = {
|
|
"id": "moved",
|
|
"op": "translate",
|
|
"target": "later",
|
|
"vector": [1, 0, 0],
|
|
}
|
|
with self.assertRaisesRegex(RecipeError, "moved"):
|
|
normalize_model(value, input_bound=False)
|
|
imported = {
|
|
"type": "recipe",
|
|
"recipe_version": 1,
|
|
"title": "import",
|
|
"unit": "mm",
|
|
"steps": [{"id": "part", "op": "import_step", "input": "geometry"}],
|
|
"result": "part",
|
|
}
|
|
with self.assertRaises(RecipeError) as caught:
|
|
normalize_model(imported, input_bound=False)
|
|
self.assertEqual(caught.exception.code, "FREECAD_STEP_INPUT_NOT_BOUND")
|
|
|
|
|
|
class FreeCadLauncherTests(unittest.TestCase):
|
|
def test_version_gate_accepts_only_stable_113_to_before_120(self) -> None:
|
|
executable = Path(r"C:\Program Files\FreeCAD 1.1\bin\FreeCADCmd.exe")
|
|
for output in ("FreeCAD 1.1.3", "FreeCAD 1.1.9 Revision: 12345"):
|
|
with patch.object(
|
|
subprocess,
|
|
"run",
|
|
return_value=subprocess.CompletedProcess([], 0, output, ""),
|
|
):
|
|
self.assertTrue(launcher._version(executable).startswith("1.1."))
|
|
for output in (
|
|
"FreeCAD 1.1.2",
|
|
"FreeCAD 1.2.0",
|
|
"FreeCAD 1.1.3 RC1",
|
|
"FreeCAD 1.1.4 weekly",
|
|
"FreeCAD 1.1.4, Libs: 1.1.4devR123",
|
|
"FreeCAD 1.1.4, Libs: 1.1.4dev.123",
|
|
):
|
|
with (
|
|
patch.object(
|
|
subprocess,
|
|
"run",
|
|
return_value=subprocess.CompletedProcess([], 0, output, ""),
|
|
),
|
|
self.assertRaises(launcher.AdapterError),
|
|
):
|
|
launcher._version(executable)
|
|
|
|
def test_managed_launcher_uses_fake_freecadcmd_with_fixed_inputs(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
job_dir = Path(directory) / "含空格 job"
|
|
request = {
|
|
"schema_version": 1,
|
|
"inputs": [],
|
|
"operation": {"model": model()},
|
|
"outputs": [],
|
|
}
|
|
record = {
|
|
"job_id": "job",
|
|
"lease_id": "lease",
|
|
"request_digest": "digest",
|
|
"request": request,
|
|
}
|
|
request_path = job_dir / "request" / "request.json"
|
|
request_path.parent.mkdir(parents=True)
|
|
request_path.write_text(json.dumps(record), encoding="utf-8")
|
|
fake = Path(r"C:\Program Files\FreeCAD 1.1\bin\FreeCADCmd.exe")
|
|
captured: dict[str, object] = {}
|
|
|
|
def execute(command, **kwargs):
|
|
captured["command"] = command
|
|
captured["environment"] = kwargs["env"]
|
|
output = job_dir / "output"
|
|
metadata = output / ".meta"
|
|
for name in (
|
|
"model.FCStd",
|
|
"model.step",
|
|
"model-recipe.json",
|
|
"model-manifest.json",
|
|
):
|
|
(output / name).write_bytes(b"fake-output")
|
|
(metadata / "provenance.json").write_bytes(b"{}")
|
|
(metadata / "preview-mesh.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"vertices": [[0, 0, 0], [1, 0, 0], [0, 1, 0]],
|
|
"triangles": [[0, 1, 2]],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(metadata / "engine-result.json").write_text(
|
|
'{"status":"succeeded"}', encoding="utf-8"
|
|
)
|
|
return subprocess.CompletedProcess(command, 0)
|
|
|
|
with (
|
|
patch.object(launcher, "_find_freecad", return_value=(fake, "1.1.3")),
|
|
patch.object(launcher.subprocess, "run", side_effect=execute),
|
|
):
|
|
artifacts = launcher.run(job_dir)
|
|
command = captured["command"]
|
|
environment = captured["environment"]
|
|
self.assertEqual(command[0], str(fake))
|
|
self.assertIn("--safe-mode", command)
|
|
self.assertEqual(Path(command[-1]).name, "freecad_worker.py")
|
|
self.assertEqual(
|
|
environment["ZCBOT_FREECAD_JOB_DIR"], str(job_dir.resolve())
|
|
)
|
|
self.assertNotIn("PYTHONPATH", {key.upper() for key in environment})
|
|
self.assertEqual(len(artifacts), 6)
|
|
|
|
def test_launcher_uses_fixed_script_safe_mode_and_isolated_configuration(
|
|
self,
|
|
) -> None:
|
|
source = (ADAPTER / "worker.py").read_text(encoding="utf-8")
|
|
for marker in (
|
|
"--safe-mode",
|
|
"--user-cfg",
|
|
"--system-cfg",
|
|
"FREECAD_USER_HOME",
|
|
"FREECAD_USER_DATA",
|
|
"FREECAD_USER_TEMP",
|
|
"ZCBOT_FREECAD_JOB_DIR",
|
|
):
|
|
self.assertIn(marker, source)
|
|
for forbidden in ("shell=True", "eval(", "exec(", "requests."):
|
|
self.assertNotIn(forbidden, source)
|
|
engine = (ADAPTER / "freecad_worker.py").read_text(encoding="utf-8")
|
|
self.assertIn("do not rely on __main__", engine)
|
|
self.assertNotIn("FreeCADGui", engine)
|
|
self.assertNotIn("Spreadsheet", engine)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|