2163 lines
104 KiB
Python
2163 lines
104 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from engine.cdsl_engine.topology import (
|
|
TopologyBlendRelation, TopologyDelta, TopologyDeltaRelation, TopologyLineage, TopologyRecord, TopologyRegistry,
|
|
TopologySectionRelation,
|
|
)
|
|
|
|
|
|
def _intent(
|
|
family: str = "CAP_FACE",
|
|
version: str | None = "1511",
|
|
*,
|
|
allowed: list[str] | None = None,
|
|
multiplicity: str = "one",
|
|
) -> dict:
|
|
source_query = {
|
|
"ast": {"call": family},
|
|
"standard_library": "onshape/std/geometry.fs",
|
|
"standard_library_version": "1511.0",
|
|
}
|
|
if version is not None:
|
|
source_query["featurescript_version"] = version
|
|
return {
|
|
"version": "1.0",
|
|
"kind": "face",
|
|
"query_family": family,
|
|
"source_query": source_query,
|
|
"derivation_policy": {"allowed": allowed or ["continuation"], "multiplicity": multiplicity},
|
|
"evidence": "kernel_history",
|
|
}
|
|
|
|
|
|
def _intersection_intent() -> dict:
|
|
return {
|
|
"version": "1.0",
|
|
"kind": "edge",
|
|
"query_family": "INTERSECT",
|
|
"source_query": {
|
|
"ast": {"call": "makeQuery"}, "featurescript_version": "1511",
|
|
"standard_library": "onshape/std/geometry.fs",
|
|
"standard_library_version": "1511.0",
|
|
},
|
|
"derivation_policy": {"allowed": ["intersection"], "multiplicity": "one"},
|
|
"evidence": "kernel_history",
|
|
"intersection_sources": [
|
|
{"query_family": "CAP_FACE", "owner_feature_id": "f1", "output_role": "extrude.end"},
|
|
{
|
|
"query_family": "SWEPT_FACE", "owner_feature_id": "f2",
|
|
"source_entity": {"sketch_id": "S2", "entity_id": "E2"},
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
def _query_expr_leaf(label: str) -> dict:
|
|
return {
|
|
"node": "topology_query",
|
|
"owner": {"node": "literal", "value": "f1.opExtrude"},
|
|
"topology_type": {"node": "literal", "value": "SWEPT_FACE"},
|
|
"entity_type": {"node": "literal", "value": "FACE"},
|
|
"arguments": [{"node": "literal", "value": label}],
|
|
}
|
|
|
|
|
|
def _swept_face_operand(entity_id: str, *, all_fragments: bool = False) -> dict:
|
|
expression = _query_expr_leaf(entity_id)
|
|
allowed = ["boundary", "fragment"] if all_fragments else ["boundary"]
|
|
return {
|
|
"kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_FACE", allowed=allowed, multiplicity="all_fragments" if all_fragments else "one"),
|
|
"source_entity": {"sketch_id": "F0", "entity_id": entity_id},
|
|
"query_expr": {"version": "1.0", "root": expression},
|
|
},
|
|
}
|
|
|
|
|
|
def _proven_set(operands: list[dict], *, operator: str = "union", kind: str = "face") -> dict:
|
|
contracts = {
|
|
"union": "proven_operand_union",
|
|
"intersection": "proven_operand_intersection",
|
|
"subtraction": "proven_operand_subtraction",
|
|
}
|
|
roots = [item["selector_intent"]["query_expr"]["root"] for item in operands]
|
|
return {
|
|
"kind": kind,
|
|
"source": "runtime_snapshot",
|
|
"confidence": 1.0,
|
|
"query_operands": operands,
|
|
"selector_intent": {
|
|
**_intent("QUERY_SET", allowed=["boundary"], multiplicity="source_qualified"),
|
|
"query_expr": {
|
|
"version": "1.0",
|
|
"root": {"node": "set", "operator": operator, "operands": roots},
|
|
},
|
|
"query_set_contract": contracts[operator],
|
|
"set_kind": kind,
|
|
"body_scope": "active_member",
|
|
"empty_policy": "reject",
|
|
"multiple_policy": "all",
|
|
},
|
|
}
|
|
|
|
|
|
def _proven_union(operands: list[dict], *, kind: str = "face") -> dict:
|
|
return _proven_set(operands, kind=kind)
|
|
|
|
|
|
def _owner_body_selector(query_input: dict) -> dict:
|
|
input_root = query_input["selector_intent"]["query_expr"]["root"]
|
|
return {
|
|
"kind": "body", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"query_input": query_input,
|
|
"selector_intent": {
|
|
**_intent("OWNER_BODY", allowed=["boundary"]),
|
|
"kind": "body",
|
|
"query_expr": {
|
|
"version": "1.0",
|
|
"root": {
|
|
"node": "filter", "filter": "owner_body", "input": input_root,
|
|
"arguments": [],
|
|
},
|
|
},
|
|
"owner_body_contract": "exact_input_owner",
|
|
"body_scope": "active_member",
|
|
"empty_policy": "reject",
|
|
"multiple_policy": "one",
|
|
},
|
|
}
|
|
|
|
|
|
def _primary_cut_copy_cap_edge_selector() -> dict:
|
|
input_root = {
|
|
"node": "topology_query",
|
|
"owner": {"node": "literal", "value": "f1.opExtrude"},
|
|
"topology_type": {"node": "literal", "value": "CAP_EDGE"},
|
|
"entity_type": {"node": "literal", "value": "EDGE"},
|
|
"arguments": [],
|
|
}
|
|
query_input = {
|
|
"kind": "edge", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_EDGE", allowed=["boundary", "continuation"]),
|
|
"kind": "edge",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
"lineage_role": "extrude.end",
|
|
"query_expr": {"version": "1.0", "root": input_root},
|
|
},
|
|
}
|
|
return {
|
|
"kind": "edge", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"query_input": query_input,
|
|
"selector_intent": {
|
|
**_intent("COPY", allowed=["boundary", "continuation"]),
|
|
"kind": "edge",
|
|
"copy_contract": "primary_cut_cap_edge",
|
|
"query_expr": {
|
|
"version": "1.0",
|
|
"root": {
|
|
"node": "topology_query",
|
|
"owner": {"node": "literal", "value": "f1.boolean.opBoolean"},
|
|
"topology_type": {"node": "literal", "value": "COPY"},
|
|
"entity_type": {"node": "literal", "value": "EDGE"},
|
|
"arguments": [{
|
|
"node": "map",
|
|
"entries": [{"key": "derivedFrom", "value": input_root}],
|
|
}],
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def _primary_cut_copy_cap_face_selector() -> dict:
|
|
input_root = {
|
|
"node": "topology_query",
|
|
"owner": {"node": "literal", "value": "f1.opExtrude"},
|
|
"topology_type": {"node": "literal", "value": "CAP_FACE"},
|
|
"entity_type": {"node": "literal", "value": "FACE"},
|
|
"arguments": [],
|
|
}
|
|
query_input = {
|
|
"kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_FACE", allowed=["boundary", "continuation"]),
|
|
"kind": "face",
|
|
"source_entities": [{"sketch_id": "F0", "entity_id": "E0"}],
|
|
"lineage_role": "extrude.end",
|
|
"query_expr": {"version": "1.0", "root": input_root},
|
|
},
|
|
}
|
|
return {
|
|
"kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"query_input": query_input,
|
|
"selector_intent": {
|
|
**_intent("COPY", allowed=["boundary", "continuation"]),
|
|
"kind": "face",
|
|
"copy_contract": "primary_cut_cap_face_workplane",
|
|
"query_expr": {
|
|
"version": "1.0",
|
|
"root": {
|
|
"node": "topology_query",
|
|
"owner": {"node": "literal", "value": "f1.boolean.opBoolean"},
|
|
"topology_type": {"node": "literal", "value": "COPY"},
|
|
"entity_type": {"node": "literal", "value": "FACE"},
|
|
"arguments": [{
|
|
"node": "map",
|
|
"entries": [{"key": "derivedFrom", "value": input_root}],
|
|
}],
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
class SelectorProvenanceContractTests(unittest.TestCase):
|
|
def test_primary_cut_copy_cap_edge_requires_exact_two_link_lineage(self) -> None:
|
|
registry = TopologyRegistry()
|
|
source, transient, result = object(), object(), object()
|
|
registry.register(TopologyRecord(
|
|
"anchor:f1:edge:0", "edge", "f1", value=source, source_entity=("F0", "E0"),
|
|
))
|
|
registry.register(TopologyRecord(
|
|
"transient:f1:edge:0", "edge", "f1", "transient:f1", value=transient, transient=True,
|
|
))
|
|
registry.register(TopologyRecord(
|
|
"body:f1:edge:0", "edge", "f1", "body:f1", value=result,
|
|
))
|
|
registry._lineage.extend([
|
|
TopologyLineage(
|
|
("anchor:f1:edge:0",), ("transient:f1:edge:0",), "boundary", "kernel_history",
|
|
"complete", "proven", "extrude", "extrude.end", "f1", "edge", "edge",
|
|
),
|
|
TopologyLineage(
|
|
("transient:f1:edge:0",), ("body:f1:edge:0",), "continuation", "kernel_history",
|
|
"complete", "proven", "subtract", None, "f1", "edge", "edge",
|
|
),
|
|
])
|
|
selector = _primary_cut_copy_cap_edge_selector()
|
|
|
|
resolved = registry.resolve(selector, active_body_id="body:f1")
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "copy_lineage")
|
|
self.assertEqual(resolved.record.record_id, "body:f1:edge:0")
|
|
self.assertEqual(resolved.evidence["transient_records"], ["transient:f1:edge:0"])
|
|
self.assertEqual([item["operation"] for item in resolved.evidence["relations"]], ["extrude", "subtract"])
|
|
|
|
selector["selector_intent"]["query_expr"]["root"]["arguments"][0]["entries"][0]["value"] = {
|
|
"node": "literal", "value": "unrelated",
|
|
}
|
|
rejected = registry.resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_copy_invalid")
|
|
|
|
def test_primary_cut_copy_cap_face_requires_one_complete_active_successor(self) -> None:
|
|
selector = _primary_cut_copy_cap_face_selector()
|
|
|
|
def registry_for(
|
|
*,
|
|
continuation_count: int = 1,
|
|
continuation_coverage: str = "complete",
|
|
result_body_id: str = "body:f1",
|
|
) -> TopologyRegistry:
|
|
registry = TopologyRegistry()
|
|
registry.register(TopologyRecord("transient:f1:face:0", "face", "f1", "transient:f1", transient=True))
|
|
registry.register(TopologyRecord("body:f1:face:0", "face", "f1", result_body_id))
|
|
registry._lineage.append(TopologyLineage(
|
|
("source:f1:profile",), ("transient:f1:face:0",), "boundary", "kernel_history",
|
|
"complete", "proven", "extrude", "extrude.end", "f1", "face", "face",
|
|
))
|
|
for index in range(continuation_count):
|
|
registry._lineage.append(TopologyLineage(
|
|
("transient:f1:face:0",), ("body:f1:face:0",), "continuation", "kernel_history",
|
|
continuation_coverage, "proven", "subtract", None, "f1", "face", "face",
|
|
))
|
|
return registry
|
|
|
|
resolved = registry_for().resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "copy_lineage")
|
|
|
|
for label, registry in (
|
|
("missing", registry_for(continuation_count=0)),
|
|
("ambiguous", registry_for(continuation_count=2)),
|
|
("partial", registry_for(continuation_coverage="partial")),
|
|
):
|
|
with self.subTest(continuation=label):
|
|
rejected = registry.resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_copy_kernel_history_missing")
|
|
|
|
inactive = registry_for(result_body_id="body:other").resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(inactive.status, "not_found")
|
|
self.assertEqual(inactive.diagnostic.code, "selector_copy_inactive")
|
|
|
|
def test_primary_cut_copy_swept_face_requires_exact_source_edge_continuation(self) -> None:
|
|
selector = _primary_cut_copy_cap_face_selector()
|
|
selector["selector_intent"]["copy_contract"] = "primary_cut_swept_face_workplane"
|
|
query_input = selector["query_input"]
|
|
input_intent = query_input["selector_intent"]
|
|
input_intent["query_family"] = "SWEPT_FACE"
|
|
input_intent.pop("source_entities")
|
|
input_intent.pop("lineage_role")
|
|
input_intent["source_entity"] = {"sketch_id": "F0", "entity_id": "E0"}
|
|
input_root = input_intent["query_expr"]["root"]
|
|
input_root["topology_type"] = {"node": "literal", "value": "SWEPT_FACE"}
|
|
|
|
def registry_for(*, continuation_count: int = 1, result_body_id: str = "body:f1") -> TopologyRegistry:
|
|
registry = TopologyRegistry()
|
|
registry.register(TopologyRecord(
|
|
"anchor:f1:edge:0", "edge", "f1", value=object(), source_entity=("F0", "E0"), transient=True,
|
|
))
|
|
registry.register(TopologyRecord("transient:f1:face:0", "face", "f1", "transient:f1", transient=True))
|
|
registry.register(TopologyRecord("body:f1:face:0", "face", "f1", result_body_id))
|
|
registry._lineage.append(TopologyLineage(
|
|
("anchor:f1:edge:0",), ("transient:f1:face:0",), "boundary", "kernel_history",
|
|
"complete", "proven", "extrude", None, "f1", "edge", "face",
|
|
))
|
|
for _ in range(continuation_count):
|
|
registry._lineage.append(TopologyLineage(
|
|
("transient:f1:face:0",), ("body:f1:face:0",), "continuation", "kernel_history",
|
|
"complete", "proven", "subtract", None, "f1", "face", "face",
|
|
))
|
|
return registry
|
|
|
|
resolved = registry_for().resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "copy_lineage")
|
|
self.assertEqual(resolved.record.record_id, "body:f1:face:0")
|
|
|
|
for label, registry in (("missing", registry_for(continuation_count=0)), ("ambiguous", registry_for(continuation_count=2))):
|
|
with self.subTest(continuation=label):
|
|
rejected = registry.resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_copy_kernel_history_missing")
|
|
|
|
inactive = registry_for(result_body_id="body:other").resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(inactive.status, "not_found")
|
|
self.assertEqual(inactive.diagnostic.code, "selector_copy_inactive")
|
|
|
|
def _two_side_registry(
|
|
self,
|
|
*,
|
|
missing_right_result: bool = False,
|
|
split_left_result: bool = False,
|
|
right_shares_left_result: bool = False,
|
|
) -> TopologyRegistry:
|
|
registry = TopologyRegistry()
|
|
left_source, right_source = object(), object()
|
|
left_result, left_fragment, right_result = object(), object(), object()
|
|
if right_shares_left_result:
|
|
right_result = left_result
|
|
missing_result = object()
|
|
anchors = [
|
|
TopologyRecord("anchor:left", "edge", "f1", value=left_source, source_entity=("F0", "left")),
|
|
TopologyRecord("anchor:right", "edge", "f1", value=right_source, source_entity=("F0", "right")),
|
|
]
|
|
for anchor in anchors:
|
|
registry.register(anchor)
|
|
left_values = (left_result, left_fragment) if split_left_result else (left_result,)
|
|
right_values = (right_result, missing_result) if missing_right_result else (right_result,)
|
|
relations = [
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", left_source, left_values,
|
|
source_kind="edge", result_kind="face", derivation="boundary",
|
|
),
|
|
]
|
|
if not right_shares_left_result:
|
|
relations.append(TopologyDeltaRelation(
|
|
"generated", "edge", right_source, right_values,
|
|
source_kind="edge", result_kind="face", derivation="boundary",
|
|
))
|
|
registry.replace_body_topology(
|
|
"f1", "body:f1", [
|
|
TopologyRecord("body:f1:left", "face", "f1", "body:f1", value=left_result),
|
|
*([TopologyRecord("body:f1:left-fragment", "face", "f1", "body:f1", value=left_fragment)] if split_left_result else []),
|
|
*([] if right_shares_left_result else [
|
|
TopologyRecord("body:f1:right", "face", "f1", "body:f1", value=right_result),
|
|
]),
|
|
],
|
|
topology_delta=TopologyDelta("extrude", tuple(relations)),
|
|
additional_predecessors=anchors,
|
|
)
|
|
if right_shares_left_result:
|
|
# Set algebra can overlap only when separate complete operations
|
|
# prove the same active record. Keeping this fact in a different
|
|
# operation component avoids converting it into an artificial
|
|
# source merge, which the direct child policy must still reject.
|
|
registry._lineage.append(TopologyLineage(
|
|
source_record_ids=("anchor:right",),
|
|
result_record_ids=("body:f1:left",),
|
|
derivation="boundary",
|
|
evidence="kernel_history",
|
|
coverage="complete",
|
|
status="proven",
|
|
operation="independent_extrude",
|
|
feature_id="f1",
|
|
source_kind="edge",
|
|
result_kind="face",
|
|
))
|
|
return registry
|
|
|
|
def test_proven_q_union_preserves_operand_order_and_deduplicates_handles(self) -> None:
|
|
registry = self._two_side_registry()
|
|
left = _swept_face_operand("left")
|
|
right = _swept_face_operand("right")
|
|
|
|
resolved = registry.resolve(_proven_union([left, right, left]), active_body_id="body:f1")
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "query_set_union")
|
|
self.assertEqual([record.record_id for record in resolved.records], ["body:f1:left", "body:f1:right"])
|
|
self.assertEqual(
|
|
[item["result_records"] for item in resolved.evidence["operand_resolutions"]],
|
|
[["body:f1:left"], ["body:f1:right"], ["body:f1:left"]],
|
|
)
|
|
|
|
def test_proven_q_union_keeps_the_complete_fragment_set_per_operand(self) -> None:
|
|
registry = self._two_side_registry(split_left_result=True)
|
|
split_left = _swept_face_operand("left", all_fragments=True)
|
|
right = _swept_face_operand("right")
|
|
|
|
resolved = registry.resolve(_proven_union([split_left, right]), active_body_id="body:f1")
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(
|
|
[record.record_id for record in resolved.records],
|
|
["body:f1:left", "body:f1:left-fragment", "body:f1:right"],
|
|
)
|
|
self.assertEqual(
|
|
resolved.evidence["operand_resolutions"][0]["result_records"],
|
|
["body:f1:left", "body:f1:left-fragment"],
|
|
)
|
|
|
|
def test_proven_q_union_evaluates_nested_set_tree_without_flattening(self) -> None:
|
|
registry = self._two_side_registry()
|
|
left = _swept_face_operand("left")
|
|
right = _swept_face_operand("right")
|
|
nested = _proven_union([right, left])
|
|
|
|
resolved = registry.resolve(
|
|
_proven_union([left, nested]), active_body_id="body:f1",
|
|
)
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "query_set_union")
|
|
self.assertEqual(
|
|
[record.record_id for record in resolved.records],
|
|
["body:f1:left", "body:f1:right"],
|
|
)
|
|
self.assertEqual(
|
|
resolved.evidence["operand_resolutions"][1]["resolution_mode"],
|
|
"query_set_union",
|
|
)
|
|
|
|
def test_proven_q_intersection_keeps_first_operand_order_for_exact_shared_records(self) -> None:
|
|
registry = self._two_side_registry(split_left_result=True, right_shares_left_result=True)
|
|
left = _swept_face_operand("left", all_fragments=True)
|
|
right = _swept_face_operand("right")
|
|
|
|
resolved = registry.resolve(
|
|
_proven_set([left, right], operator="intersection"), active_body_id="body:f1",
|
|
)
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "query_set_intersection")
|
|
self.assertEqual([record.record_id for record in resolved.records], ["body:f1:left"])
|
|
self.assertEqual(resolved.evidence["set_operator"], "intersection")
|
|
|
|
def test_proven_q_subtraction_keeps_left_order_and_removes_exact_right_records(self) -> None:
|
|
registry = self._two_side_registry(split_left_result=True, right_shares_left_result=True)
|
|
left = _swept_face_operand("left", all_fragments=True)
|
|
right = _swept_face_operand("right")
|
|
|
|
resolved = registry.resolve(
|
|
_proven_set([left, right], operator="subtraction"), active_body_id="body:f1",
|
|
)
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "query_set_subtraction")
|
|
self.assertEqual([record.record_id for record in resolved.records], ["body:f1:left-fragment"])
|
|
self.assertEqual(resolved.evidence["set_operator"], "subtraction")
|
|
|
|
def test_proven_q_intersection_and_subtraction_reject_an_empty_final_set(self) -> None:
|
|
left = _swept_face_operand("left")
|
|
right = _swept_face_operand("right")
|
|
cases = (
|
|
(_proven_set([left, right], operator="intersection"), self._two_side_registry()),
|
|
(_proven_set([left, left], operator="subtraction"), self._two_side_registry()),
|
|
)
|
|
|
|
for selector, registry in cases:
|
|
with self.subTest(operator=selector["selector_intent"]["query_expr"]["root"]["operator"]):
|
|
resolved = registry.resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(resolved.status, "not_found")
|
|
self.assertEqual(resolved.diagnostic.code, "selector_query_empty")
|
|
|
|
def test_proven_q_subtraction_requires_exactly_two_source_operands(self) -> None:
|
|
left = _swept_face_operand("left")
|
|
selector = _proven_set([left, left, left], operator="subtraction")
|
|
|
|
resolved = TopologyRegistry().resolve(selector, active_body_id="body:f1")
|
|
|
|
self.assertEqual(resolved.status, "not_found")
|
|
self.assertEqual(resolved.diagnostic.code, "selector_query_set_invalid")
|
|
self.assertIn("qSubtraction", resolved.diagnostic.detail["reason"])
|
|
|
|
def test_proven_q_union_rejects_empty_or_kind_incompatible_contracts(self) -> None:
|
|
left = _swept_face_operand("left")
|
|
empty = _proven_union([left, left])
|
|
empty["query_operands"] = []
|
|
mixed = _proven_union([left, _swept_face_operand("right")])
|
|
mixed["query_operands"][1]["kind"] = "edge"
|
|
non_provenance = _proven_union([left, _swept_face_operand("right")])
|
|
non_provenance["query_operands"][1]["geometry"] = {"surface_type": "plane"}
|
|
|
|
for selector, expected in (
|
|
(empty, "operands do not match"),
|
|
(mixed, "parent set kind"),
|
|
(non_provenance, "unsupported non-provenance evidence"),
|
|
):
|
|
with self.subTest(expected=expected):
|
|
resolved = TopologyRegistry().resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(resolved.status, "not_found")
|
|
self.assertEqual(resolved.diagnostic.code, "selector_query_set_invalid")
|
|
self.assertIn(expected, resolved.diagnostic.detail["reason"])
|
|
|
|
def test_proven_q_union_rejects_inactive_or_partial_operand_lineage(self) -> None:
|
|
left = _swept_face_operand("left")
|
|
right = _swept_face_operand("right")
|
|
selector = _proven_union([left, right])
|
|
|
|
inactive = self._two_side_registry().resolve(selector, active_body_id="body:later")
|
|
self.assertEqual(inactive.status, "not_found")
|
|
self.assertEqual(inactive.diagnostic.code, "selector_body_member_inactive")
|
|
self.assertEqual(inactive.diagnostic.detail["operand_index"], 0)
|
|
|
|
partial = self._two_side_registry(missing_right_result=True).resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(partial.status, "not_found")
|
|
self.assertEqual(partial.diagnostic.code, "selector_kernel_history_missing")
|
|
self.assertEqual(partial.diagnostic.detail["operand_index"], 1)
|
|
|
|
def test_direct_swept_body_resolves_only_the_active_member(self) -> None:
|
|
"""A direct body query is body-graph evidence, never a stable-ID fallback."""
|
|
registry = TopologyRegistry()
|
|
body = object()
|
|
registry.register(TopologyRecord(
|
|
"body:f1", "body", "f1", "body:f1", value=body, owner_feature_ids=("f1",),
|
|
))
|
|
selector = {
|
|
"kind": "body", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_BODY", allowed=["boundary"]),
|
|
"kind": "body",
|
|
"evidence": "active_body_member",
|
|
"body_member_contract": "direct_new_body",
|
|
},
|
|
}
|
|
|
|
resolved = registry.resolve(selector, active_body_id="body:f1")
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertIs(resolved.record.value, body)
|
|
self.assertEqual(resolved.resolution_mode, "body_member")
|
|
self.assertEqual(resolved.evidence["body_member"], "body:f1")
|
|
|
|
inactive = registry.resolve(selector, active_body_id="body:later")
|
|
self.assertEqual(inactive.status, "not_found")
|
|
self.assertEqual(inactive.diagnostic.code, "selector_body_member_inactive")
|
|
|
|
mixed = {**selector, "stable_id": "body:f1"}
|
|
rejected = registry.resolve(mixed, active_body_id="body:f1")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_source_unavailable")
|
|
|
|
def test_q_owner_body_projects_one_proven_active_member_to_its_body_record(self) -> None:
|
|
registry = self._two_side_registry()
|
|
body = object()
|
|
registry.register(TopologyRecord(
|
|
"body:f1", "body", "f1", "body:f1", value=body, owner_feature_ids=("f1",),
|
|
))
|
|
|
|
resolved = registry.resolve(
|
|
_owner_body_selector(_swept_face_operand("left")), active_body_id="body:f1",
|
|
)
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "owner_body")
|
|
self.assertIs(resolved.record.value, body)
|
|
self.assertEqual(resolved.evidence["input_record_ids"], ["body:f1:left"])
|
|
self.assertEqual(resolved.evidence["body_member"], "body:f1")
|
|
|
|
def test_q_owner_body_rejects_ambiguous_or_missing_exact_body_projection(self) -> None:
|
|
ambiguous_registry = self._two_side_registry(split_left_result=True)
|
|
ambiguous_registry.register(TopologyRecord(
|
|
"body:f1", "body", "f1", "body:f1", value=object(), owner_feature_ids=("f1",),
|
|
))
|
|
ambiguous = ambiguous_registry.resolve(
|
|
_owner_body_selector(_swept_face_operand("left", all_fragments=True)), active_body_id="body:f1",
|
|
)
|
|
self.assertEqual(ambiguous.status, "not_found")
|
|
self.assertEqual(ambiguous.diagnostic.code, "selector_owner_body_ambiguous")
|
|
|
|
missing = self._two_side_registry().resolve(
|
|
_owner_body_selector(_swept_face_operand("left")), active_body_id="body:f1",
|
|
)
|
|
self.assertEqual(missing.status, "not_found")
|
|
self.assertEqual(missing.diagnostic.code, "selector_owner_body_not_found")
|
|
|
|
def test_q_owner_body_rejects_inactive_input_and_tampered_contract(self) -> None:
|
|
registry = self._two_side_registry()
|
|
registry.register(TopologyRecord(
|
|
"body:f1", "body", "f1", "body:f1", value=object(), owner_feature_ids=("f1",),
|
|
))
|
|
selector = _owner_body_selector(_swept_face_operand("left"))
|
|
|
|
inactive = registry.resolve(selector, active_body_id="body:later")
|
|
self.assertEqual(inactive.status, "not_found")
|
|
self.assertEqual(inactive.diagnostic.code, "selector_owner_body_input_unresolved")
|
|
self.assertEqual(inactive.diagnostic.detail["cause"]["code"], "selector_body_member_inactive")
|
|
|
|
tampered = {**selector, "geometry": {"surface_type": "plane"}}
|
|
rejected = registry.resolve(tampered, active_body_id="body:f1")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_owner_body_invalid")
|
|
|
|
query_set_child = _owner_body_selector(_swept_face_operand("left"))
|
|
query_set_child["query_input"] = _proven_union([
|
|
_swept_face_operand("left"), _swept_face_operand("left"),
|
|
])
|
|
query_set_child["selector_intent"]["query_expr"]["root"]["input"] = (
|
|
query_set_child["query_input"]["selector_intent"]["query_expr"]["root"]
|
|
)
|
|
rejected_set = registry.resolve(query_set_child, active_body_id="body:f1")
|
|
self.assertEqual(rejected_set.status, "not_found")
|
|
self.assertEqual(rejected_set.diagnostic.code, "selector_owner_body_invalid")
|
|
|
|
def test_direct_prism_swept_face_requires_proven_continuation_to_active_body(self) -> None:
|
|
"""A side-wall extent may cross only complete kernel continuations."""
|
|
registry = TopologyRegistry()
|
|
source_edge = object()
|
|
initial_side = object()
|
|
continued_side = object()
|
|
anchor = TopologyRecord(
|
|
"anchor:edge", "edge", "f1", value=source_edge,
|
|
source_entity=("F0", "E0"),
|
|
)
|
|
registry.register(anchor)
|
|
registry.replace_body_topology(
|
|
"f1", "body:f1", [
|
|
TopologyRecord("body:f1:face:0", "face", "f1", "body:f1", value=initial_side),
|
|
],
|
|
topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", source_edge, (initial_side,),
|
|
source_kind="edge", result_kind="face", derivation="boundary",
|
|
coverage="complete", status="proven",
|
|
),
|
|
)),
|
|
additional_predecessors=[anchor],
|
|
)
|
|
registry.replace_body_topology(
|
|
"cut", "body:cut", [
|
|
TopologyRecord("body:cut:face:0", "face", "cut", "body:cut", value=continued_side),
|
|
],
|
|
topology_delta=TopologyDelta("subtract", (
|
|
TopologyDeltaRelation(
|
|
"modified", "face", initial_side, (continued_side,),
|
|
derivation="continuation", coverage="complete", status="proven",
|
|
),
|
|
)),
|
|
)
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "f1",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_FACE", allowed=["boundary", "continuation"]),
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
},
|
|
}
|
|
|
|
resolved = registry.resolve(selector, active_body_id="body:cut")
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.record.record_id, "body:cut:face:0")
|
|
self.assertEqual(resolved.resolution_mode, "kernel_lineage")
|
|
self.assertEqual(
|
|
[relation["derivation"] for relation in resolved.evidence["relations"]],
|
|
["boundary", "continuation"],
|
|
)
|
|
|
|
boundary_only = {
|
|
**selector,
|
|
"selector_intent": {
|
|
**selector["selector_intent"],
|
|
"derivation_policy": {"allowed": ["boundary"], "multiplicity": "one"},
|
|
},
|
|
}
|
|
rejected = registry.resolve(boundary_only, active_body_id="body:cut")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
def test_direct_prism_source_anchors_resolve_cross_kind_kernel_lineage(self) -> None:
|
|
"""Direct profile entities resolve only through their prism history."""
|
|
from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter
|
|
|
|
adapter = Build123dGeometryAdapter()
|
|
outer = [
|
|
{"type": "line", "start_mm": [0, 0, 0], "end_mm": [10, 0, 0], "source_entity_id": "E0"},
|
|
{"type": "line", "start_mm": [10, 0, 0], "end_mm": [10, 5, 0], "source_entity_id": "E1"},
|
|
{"type": "line", "start_mm": [10, 5, 0], "end_mm": [0, 5, 0], "source_entity_id": "E2"},
|
|
{"type": "line", "start_mm": [0, 5, 0], "end_mm": [0, 0, 0], "source_entity_id": "E3"},
|
|
]
|
|
faces, anchor_specs = adapter.faces_for_sketch_with_source_anchors({
|
|
"id": "profile",
|
|
"source_sketch_id": "F0",
|
|
"profile": {"type": "analytic_contours", "contours": []},
|
|
"contour_regions_mm": [{"outer": outer, "holes": []}],
|
|
})
|
|
self.assertEqual(len(faces), 1)
|
|
self.assertEqual(
|
|
{(item["kind"], item.get("source_entity"), item.get("source_entities")) for item in anchor_specs},
|
|
{
|
|
("edge", ("F0", "E0"), None), ("edge", ("F0", "E1"), None),
|
|
("edge", ("F0", "E2"), None), ("edge", ("F0", "E3"), None),
|
|
("vertex", None, (("F0", "E0"), ("F0", "E1"))),
|
|
("vertex", None, (("F0", "E0"), ("F0", "E3"))),
|
|
("vertex", None, (("F0", "E1"), ("F0", "E2"))),
|
|
("vertex", None, (("F0", "E2"), ("F0", "E3"))),
|
|
},
|
|
)
|
|
solid, delta = adapter.extrude_with_topology_delta(faces[0], [0, 0, 8])
|
|
registry = TopologyRegistry()
|
|
anchors = [
|
|
TopologyRecord(
|
|
f"anchor:{index}", item["kind"], "extrude", value=item["value"],
|
|
source_entity=item.get("source_entity"),
|
|
source_entities=tuple(item.get("source_entities") or ()),
|
|
)
|
|
for index, item in enumerate(anchor_specs)
|
|
]
|
|
for anchor in anchors:
|
|
registry.register(anchor)
|
|
registry.replace_body_topology(
|
|
"extrude", "body:extrude",
|
|
adapter.topology_records(solid, "extrude", "body:extrude"),
|
|
topology_delta=delta,
|
|
additional_predecessors=anchors,
|
|
)
|
|
|
|
face_selector = {
|
|
"kind": "face", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_FACE", allowed=["boundary"]),
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
},
|
|
}
|
|
face_resolution = registry.resolve(face_selector, active_body_id="body:extrude")
|
|
self.assertEqual(face_resolution.status, "resolved")
|
|
self.assertEqual(face_resolution.resolution_mode, "kernel_lineage")
|
|
self.assertEqual(face_resolution.evidence["semantic_anchor"], {
|
|
"type": "source_entity", "sketch_id": "F0", "entity_id": "E0",
|
|
})
|
|
self.assertEqual(face_resolution.evidence["relations"][0]["source_kind"], "edge")
|
|
self.assertEqual(face_resolution.evidence["relations"][0]["result_kind"], "face")
|
|
|
|
edge_selector = {
|
|
"kind": "edge", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_EDGE", allowed=["boundary"]),
|
|
"source_entities": [
|
|
{"sketch_id": "F0", "entity_id": "E0"},
|
|
{"sketch_id": "F0", "entity_id": "E1"},
|
|
],
|
|
},
|
|
}
|
|
edge_resolution = registry.resolve(edge_selector, active_body_id="body:extrude")
|
|
self.assertEqual(edge_resolution.status, "resolved")
|
|
self.assertEqual(edge_resolution.evidence["relations"][0]["source_kind"], "vertex")
|
|
self.assertEqual(edge_resolution.evidence["relations"][0]["result_kind"], "edge")
|
|
|
|
# A later operation may materialize the same source sketch again. Its
|
|
# source anchor is not an alternative anchor for this producer query.
|
|
registry.register(TopologyRecord(
|
|
"anchor:later", "vertex", "later",
|
|
source_entities=(("F0", "E0"), ("F0", "E1")),
|
|
))
|
|
replayed_edge_resolution = registry.resolve(edge_selector, active_body_id="body:extrude")
|
|
self.assertEqual(replayed_edge_resolution.status, "resolved")
|
|
self.assertEqual(replayed_edge_resolution.record.record_id, edge_resolution.record.record_id)
|
|
|
|
cap_records = []
|
|
for role in ("extrude.start", "extrude.end"):
|
|
cap_resolution = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_EDGE", allowed=["boundary"]),
|
|
"kind": "edge",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
"lineage_role": role,
|
|
},
|
|
}, active_body_id="body:extrude")
|
|
self.assertEqual(cap_resolution.status, "resolved")
|
|
self.assertEqual(cap_resolution.resolution_mode, "kernel_lineage")
|
|
self.assertEqual(cap_resolution.evidence["relations"][0]["output_role"], role)
|
|
cap_records.append(cap_resolution.record.record_id)
|
|
self.assertEqual(len(set(cap_records)), 2)
|
|
|
|
cap_vertex_records = []
|
|
for role in ("extrude.start", "extrude.end"):
|
|
vertex_resolution = registry.resolve({
|
|
"kind": "vertex", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_VERTEX", allowed=["boundary"]),
|
|
"kind": "vertex",
|
|
"source_entities": [
|
|
{"sketch_id": "F0", "entity_id": "E0"},
|
|
{"sketch_id": "F0", "entity_id": "E1"},
|
|
],
|
|
"lineage_role": role,
|
|
},
|
|
}, active_body_id="body:extrude")
|
|
self.assertEqual(vertex_resolution.status, "resolved")
|
|
self.assertEqual(vertex_resolution.resolution_mode, "kernel_lineage")
|
|
self.assertEqual(vertex_resolution.evidence["relations"][0]["source_kind"], "vertex")
|
|
self.assertEqual(vertex_resolution.evidence["relations"][0]["result_kind"], "vertex")
|
|
self.assertEqual(vertex_resolution.evidence["relations"][0]["output_role"], role)
|
|
cap_vertex_records.append(vertex_resolution.record.record_id)
|
|
self.assertEqual(len(set(cap_vertex_records)), 2)
|
|
|
|
missing_role = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_EDGE", allowed=["boundary"]),
|
|
"kind": "edge",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
"lineage_role": "extrude.end",
|
|
},
|
|
}, active_body_id="body:other")
|
|
self.assertEqual(missing_role.status, "not_found")
|
|
self.assertEqual(missing_role.diagnostic.code, "selector_body_member_inactive")
|
|
|
|
def test_source_anchor_rejects_partial_or_missing_prism_history(self) -> None:
|
|
registry = TopologyRegistry()
|
|
source = object()
|
|
result = object()
|
|
missing_result = object()
|
|
anchor = TopologyRecord(
|
|
"anchor:edge", "edge", "extrude", value=source,
|
|
source_entity=("F0", "E0"),
|
|
)
|
|
registry.register(anchor)
|
|
registry.replace_body_topology("extrude", "body:extrude", [
|
|
TopologyRecord("extrude:side", "face", "extrude", "body:extrude", value=result),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", source, (result, missing_result),
|
|
source_kind="edge", result_kind="face", derivation="boundary",
|
|
coverage="complete", status="proven",
|
|
),
|
|
)), additional_predecessors=[anchor])
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_FACE", allowed=["boundary"]),
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
},
|
|
}
|
|
|
|
resolution = registry.resolve(selector, active_body_id="body:extrude")
|
|
|
|
# The adapter claimed complete history, but one result is absent from
|
|
# the final snapshot. The registry must downgrade the relation before
|
|
# a selector can consume it.
|
|
relation = registry.topology_deltas()[0]["relations"][0]
|
|
self.assertEqual(relation["coverage"], "partial")
|
|
self.assertEqual(relation["result_record_ids"], ["extrude:side"])
|
|
self.assertEqual(resolution.status, "not_found")
|
|
self.assertEqual(resolution.diagnostic.code, "selector_kernel_history_missing")
|
|
|
|
def test_exact_preserved_member_keeps_cap_vertex_lineage_across_compound_snapshot(self) -> None:
|
|
registry = TopologyRegistry()
|
|
source_vertex = object()
|
|
cap_vertex = object()
|
|
other_vertex = object()
|
|
anchor = TopologyRecord(
|
|
"anchor:vertex", "vertex", "f1", value=source_vertex,
|
|
source_entities=(("F0", "E0"), ("F0", "E1")),
|
|
)
|
|
registry.register(anchor)
|
|
registry.replace_body_topology(
|
|
"f1", "body:f1", [TopologyRecord("f1:cap", "vertex", "f1", "body:f1", value=cap_vertex)],
|
|
topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "vertex", source_vertex, (cap_vertex,), output_role="extrude.end",
|
|
source_kind="vertex", result_kind="vertex",
|
|
),
|
|
)), additional_predecessors=[anchor],
|
|
)
|
|
# F3 adds an independent member; F1's cap vertex is exactly the same
|
|
# B-rep item in its new compound member, while F3 itself is distinct.
|
|
registry.replace_body_topologies(
|
|
"f3", [
|
|
("body:f3:0", [TopologyRecord("f3:preserved-cap", "vertex", "f3", "body:f3:0", value=cap_vertex)]),
|
|
("body:f3:1", [TopologyRecord("f3:new", "vertex", "f3", "body:f3:1", value=other_vertex)]),
|
|
], active_body_id="body:f3", member_preservations=[("body:f1", "body:f3:0")],
|
|
)
|
|
selector = {
|
|
"kind": "vertex", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_VERTEX", allowed=["boundary", "continuation"]),
|
|
"kind": "vertex",
|
|
"source_entities": [
|
|
{"sketch_id": "F0", "entity_id": "E0"},
|
|
{"sketch_id": "F0", "entity_id": "E1"},
|
|
],
|
|
"lineage_role": "extrude.end",
|
|
},
|
|
}
|
|
resolution = registry.resolve(selector, active_body_id="body:f3")
|
|
self.assertEqual(resolution.status, "resolved")
|
|
self.assertEqual(resolution.record.record_id, "f3:preserved-cap")
|
|
self.assertEqual(
|
|
[relation["operation"] for relation in resolution.evidence["relations"]],
|
|
["extrude", "body_member_preserve"],
|
|
)
|
|
|
|
# A merely equal-looking replacement has a new opaque handle and must
|
|
# not inherit a member-preservation relation.
|
|
replacement = TopologyRegistry()
|
|
replacement.register(anchor)
|
|
replacement.replace_body_topology(
|
|
"f1", "body:f1", [TopologyRecord("f1:cap", "vertex", "f1", "body:f1", value=cap_vertex)],
|
|
topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation("generated", "vertex", source_vertex, (cap_vertex,), output_role="extrude.end", source_kind="vertex", result_kind="vertex"),
|
|
)), additional_predecessors=[anchor],
|
|
)
|
|
replacement.replace_body_topologies(
|
|
"f3", [("body:f3:0", [TopologyRecord("f3:replacement", "vertex", "f3", "body:f3:0", value=object())])],
|
|
active_body_id="body:f3", member_preservations=[("body:f1", "body:f3:0")],
|
|
)
|
|
rejected = replacement.resolve(selector, active_body_id="body:f3")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_body_member_inactive")
|
|
|
|
def test_cap_edge_continuation_requires_the_complete_active_chain(self) -> None:
|
|
"""A direct cap edge may cross only a proven one-to-one continuation."""
|
|
registry = TopologyRegistry()
|
|
source_edge = object()
|
|
initial_cap_edge = object()
|
|
continued_cap_edge = object()
|
|
anchor = TopologyRecord(
|
|
"anchor:edge", "edge", "extrude", value=source_edge,
|
|
source_entity=("F0", "E0"),
|
|
)
|
|
registry.register(anchor)
|
|
registry.replace_body_topology(
|
|
"extrude", "body:extrude", [
|
|
TopologyRecord("body:extrude:edge:0", "edge", "extrude", "body:extrude", value=initial_cap_edge),
|
|
],
|
|
topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", source_edge, (initial_cap_edge,),
|
|
source_kind="edge", result_kind="edge", derivation="boundary",
|
|
coverage="complete", status="proven", output_role="extrude.start",
|
|
),
|
|
)),
|
|
additional_predecessors=[anchor],
|
|
)
|
|
registry.replace_body_topology(
|
|
"cut", "body:cut", [
|
|
TopologyRecord("body:cut:edge:0", "edge", "cut", "body:cut", value=continued_cap_edge),
|
|
],
|
|
topology_delta=TopologyDelta("subtract", (
|
|
TopologyDeltaRelation(
|
|
"preserved", "edge", initial_cap_edge, (continued_cap_edge,),
|
|
source_kind="edge", result_kind="edge", derivation="continuation",
|
|
coverage="complete", status="proven",
|
|
),
|
|
)),
|
|
)
|
|
selector = {
|
|
"kind": "edge", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_EDGE", allowed=["boundary", "continuation"]),
|
|
"kind": "edge",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
"lineage_role": "extrude.start",
|
|
},
|
|
}
|
|
|
|
resolved = registry.resolve(selector, active_body_id="body:cut")
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.record.record_id, "body:cut:edge:0")
|
|
self.assertEqual(
|
|
[relation["derivation"] for relation in resolved.evidence["relations"]],
|
|
["boundary", "continuation"],
|
|
)
|
|
|
|
boundary_only = {
|
|
**selector,
|
|
"selector_intent": {
|
|
**selector["selector_intent"],
|
|
"derivation_policy": {"allowed": ["boundary"], "multiplicity": "one"},
|
|
},
|
|
}
|
|
rejected = registry.resolve(boundary_only, active_body_id="body:cut")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
partial_registry = TopologyRegistry()
|
|
partial_registry.register(anchor)
|
|
partial_registry.replace_body_topology(
|
|
"extrude", "body:extrude", [
|
|
TopologyRecord("body:extrude:edge:0", "edge", "extrude", "body:extrude", value=initial_cap_edge),
|
|
],
|
|
topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", source_edge, (initial_cap_edge,),
|
|
source_kind="edge", result_kind="edge", derivation="boundary",
|
|
coverage="complete", status="proven", output_role="extrude.start",
|
|
),
|
|
)),
|
|
additional_predecessors=[anchor],
|
|
)
|
|
partial_registry.replace_body_topology(
|
|
"cut", "body:cut", [
|
|
TopologyRecord("body:cut:edge:0", "edge", "cut", "body:cut", value=continued_cap_edge),
|
|
],
|
|
topology_delta=TopologyDelta("subtract", (
|
|
TopologyDeltaRelation(
|
|
"modified", "edge", initial_cap_edge, (continued_cap_edge,),
|
|
source_kind="edge", result_kind="edge", derivation="continuation",
|
|
coverage="partial", status="unknown",
|
|
),
|
|
)),
|
|
)
|
|
partial = partial_registry.resolve(selector, active_body_id="body:cut")
|
|
self.assertEqual(partial.status, "not_found")
|
|
self.assertEqual(partial.diagnostic.code, "selector_kernel_history_missing")
|
|
|
|
def test_direct_annulus_source_edges_resolve_each_prism_side_face(self) -> None:
|
|
"""Direct circular hole wires retain distinct source edge provenance."""
|
|
from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter
|
|
|
|
adapter = Build123dGeometryAdapter()
|
|
faces, anchor_specs = adapter.faces_for_sketch_with_source_anchors({
|
|
"id": "annulus",
|
|
"source_sketch_id": "F0",
|
|
"workplane": {
|
|
"origin_mm": [0, 0, 0], "x_dir": [1, 0, 0],
|
|
"y_dir": [0, 1, 0], "normal": [0, 0, 1],
|
|
},
|
|
"profile": {
|
|
"type": "analytic_contours",
|
|
"contours": [
|
|
{"role": "outer", "closed": True, "segments": [{
|
|
"type": "circle", "center": [0, 0], "radius_mm": 10,
|
|
"source_entity_id": "outer",
|
|
}]},
|
|
{"role": "inner", "closed": True, "segments": [{
|
|
"type": "circle", "center": [0, 0], "radius_mm": 5,
|
|
"source_entity_id": "inner",
|
|
}]},
|
|
],
|
|
},
|
|
})
|
|
self.assertEqual(len(faces), 1)
|
|
self.assertEqual(
|
|
{(item["kind"], item.get("source_entity")) for item in anchor_specs},
|
|
{("edge", ("F0", "outer")), ("edge", ("F0", "inner"))},
|
|
)
|
|
solid, delta = adapter.extrude_with_topology_delta(faces[0], [0, 0, 8])
|
|
registry = TopologyRegistry()
|
|
anchors = [
|
|
TopologyRecord(
|
|
f"anchor:{index}", item["kind"], "extrude", value=item["value"],
|
|
source_entity=item.get("source_entity"),
|
|
)
|
|
for index, item in enumerate(anchor_specs)
|
|
]
|
|
for anchor in anchors:
|
|
registry.register(anchor)
|
|
registry.replace_body_topology(
|
|
"extrude", "body:extrude",
|
|
adapter.topology_records(solid, "extrude", "body:extrude"),
|
|
topology_delta=delta,
|
|
additional_predecessors=anchors,
|
|
)
|
|
|
|
records = []
|
|
for entity_id in ("outer", "inner"):
|
|
resolution = registry.resolve({
|
|
"kind": "face", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_FACE", allowed=["boundary"]),
|
|
"source_entity": {"sketch_id": "F0", "entity_id": entity_id},
|
|
},
|
|
}, active_body_id="body:extrude")
|
|
self.assertEqual(resolution.status, "resolved")
|
|
self.assertEqual(resolution.resolution_mode, "kernel_lineage")
|
|
records.append(resolution.record.record_id)
|
|
self.assertEqual(len(set(records)), 2)
|
|
|
|
def test_direct_linear_profile_with_circle_hole_keeps_exact_prism_anchors(self) -> None:
|
|
"""A direct hole face retains outer vertex and inner edge provenance."""
|
|
from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter
|
|
|
|
adapter = Build123dGeometryAdapter()
|
|
outer = [
|
|
{"type": "line", "start_mm": [0, 0, 0], "end_mm": [12, 0, 0], "source_entity_id": "bottom"},
|
|
{"type": "line", "start_mm": [12, 0, 0], "end_mm": [12, 8, 0], "source_entity_id": "right"},
|
|
{"type": "line", "start_mm": [12, 8, 0], "end_mm": [0, 8, 0], "source_entity_id": "top"},
|
|
{"type": "line", "start_mm": [0, 8, 0], "end_mm": [0, 0, 0], "source_entity_id": "left"},
|
|
]
|
|
inner = [{
|
|
"type": "circle", "center_mm": [6, 4, 0], "radius_mm": 1.5,
|
|
"normal": [0, 0, 1], "x_dir_mm": [1, 0, 0], "source_entity_id": "hole",
|
|
}]
|
|
faces, anchor_specs = adapter.faces_for_sketch_with_source_anchors({
|
|
"id": "profile", "source_sketch_id": "F0",
|
|
"profile": {"type": "analytic_contours", "contours": []},
|
|
"contour_regions_mm": [{"outer": outer, "holes": [inner]}],
|
|
})
|
|
|
|
self.assertEqual(len(faces), 1)
|
|
self.assertEqual(
|
|
{(item["kind"], item.get("source_entity")) for item in anchor_specs if item["kind"] == "edge"},
|
|
{
|
|
("edge", ("F0", "bottom")), ("edge", ("F0", "right")),
|
|
("edge", ("F0", "top")), ("edge", ("F0", "left")),
|
|
("edge", ("F0", "hole")),
|
|
},
|
|
)
|
|
self.assertIn(
|
|
("vertex", None, (("F0", "right"), ("F0", "top"))),
|
|
{(item["kind"], item.get("source_entity"), item.get("source_entities")) for item in anchor_specs},
|
|
)
|
|
|
|
solid, delta = adapter.extrude_with_topology_delta(faces[0], [0, 0, 8])
|
|
anchors = [
|
|
TopologyRecord(
|
|
f"anchor:{index}", item["kind"], "extrude", value=item["value"],
|
|
source_entity=item.get("source_entity"),
|
|
source_entities=tuple(item.get("source_entities") or ()),
|
|
)
|
|
for index, item in enumerate(anchor_specs)
|
|
]
|
|
registry = TopologyRegistry()
|
|
for anchor in anchors:
|
|
registry.register(anchor)
|
|
registry.replace_body_topology(
|
|
"extrude", "body:extrude",
|
|
adapter.topology_records(solid, "extrude", "body:extrude"),
|
|
topology_delta=delta, additional_predecessors=anchors,
|
|
)
|
|
|
|
edge_resolution = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_EDGE", allowed=["boundary"]),
|
|
"source_entities": [
|
|
{"sketch_id": "F0", "entity_id": "right"},
|
|
{"sketch_id": "F0", "entity_id": "top"},
|
|
],
|
|
},
|
|
}, active_body_id="body:extrude")
|
|
self.assertEqual(edge_resolution.status, "resolved")
|
|
self.assertEqual(edge_resolution.resolution_mode, "kernel_lineage")
|
|
self.assertEqual(edge_resolution.evidence["relations"][0]["source_kind"], "vertex")
|
|
self.assertEqual(edge_resolution.evidence["relations"][0]["result_kind"], "edge")
|
|
|
|
face_resolution = registry.resolve({
|
|
"kind": "face", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_FACE", allowed=["boundary"]),
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "hole"},
|
|
},
|
|
}, active_body_id="body:extrude")
|
|
self.assertEqual(face_resolution.status, "resolved")
|
|
self.assertEqual(face_resolution.resolution_mode, "kernel_lineage")
|
|
|
|
cap_records = []
|
|
for role in ("extrude.start", "extrude.end"):
|
|
cap_resolution = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "extrude",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_EDGE", allowed=["boundary"]),
|
|
"kind": "edge",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "hole"},
|
|
"lineage_role": role,
|
|
},
|
|
}, active_body_id="body:extrude")
|
|
self.assertEqual(cap_resolution.status, "resolved")
|
|
self.assertEqual(cap_resolution.resolution_mode, "kernel_lineage")
|
|
self.assertEqual(cap_resolution.evidence["relations"][0]["source_kind"], "edge")
|
|
self.assertEqual(cap_resolution.evidence["relations"][0]["result_kind"], "edge")
|
|
self.assertEqual(cap_resolution.evidence["relations"][0]["output_role"], role)
|
|
cap_records.append(cap_resolution.record.record_id)
|
|
self.assertEqual(len(set(cap_records)), 2)
|
|
|
|
def test_partial_logical_circle_region_does_not_create_a_source_anchor(self) -> None:
|
|
"""A split circle needs every solver-proven segment before restoration."""
|
|
from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter
|
|
from engine.cdsl_engine.sketch_solver import resolve_profile
|
|
|
|
sketch = {
|
|
"id": "profile", "source_sketch_id": "F0",
|
|
"workplane": {
|
|
"origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1],
|
|
},
|
|
"profile": {"type": "analytic_contours", "contours": [
|
|
{"closed": True, "segments": [
|
|
{"type": "line", "start": [0, 0], "end": [12, 0], "source_entity_id": "bottom"},
|
|
{"type": "line", "start": [12, 0], "end": [12, 8], "source_entity_id": "right"},
|
|
{"type": "line", "start": [12, 8], "end": [0, 8], "source_entity_id": "top"},
|
|
{"type": "line", "start": [0, 8], "end": [0, 0], "source_entity_id": "left"},
|
|
]},
|
|
{"closed": True, "segments": [{
|
|
"type": "circle", "center": [6, 4], "radius_mm": 1.5, "source_entity_id": "hole",
|
|
}]},
|
|
]},
|
|
}
|
|
solved = resolve_profile(sketch)
|
|
solved["contour_regions_mm"][0]["holes"][0][-1].pop("logical_circle_source_entity_id")
|
|
_faces, anchors = Build123dGeometryAdapter().faces_for_sketch_with_source_anchors(solved)
|
|
|
|
self.assertNotIn(
|
|
("edge", ("F0", "hole")),
|
|
{(item["kind"], item.get("source_entity")) for item in anchors},
|
|
)
|
|
|
|
def test_direct_circle_profile_uses_only_its_declared_source_anchor(self) -> None:
|
|
"""A direct CDSL circle carries its source identity on the profile."""
|
|
from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter
|
|
|
|
sketch = {
|
|
"id": "profile", "source_sketch_id": "F0",
|
|
"workplane": {
|
|
"origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1],
|
|
},
|
|
"profile": {
|
|
"type": "circle", "center": [0, 0], "radius_mm": 5,
|
|
"source_entity_id": "E5",
|
|
},
|
|
# This generic entity entry deliberately disagrees. A direct
|
|
# profile must not look it up or infer the source by geometry.
|
|
"entities": [{"type": "circle", "center": [0, 0], "radius_mm": 5, "source_entity_id": "wrong"}],
|
|
}
|
|
faces, anchors = Build123dGeometryAdapter().faces_for_sketch_with_source_anchors(sketch)
|
|
|
|
self.assertEqual(len(faces), 1)
|
|
self.assertEqual(
|
|
{(item["kind"], item.get("source_entity")) for item in anchors},
|
|
{("edge", ("F0", "E5"))},
|
|
)
|
|
unlabelled = {**sketch, "profile": {**sketch["profile"]}}
|
|
unlabelled["profile"].pop("source_entity_id")
|
|
_faces, no_anchors = Build123dGeometryAdapter().faces_for_sketch_with_source_anchors(unlabelled)
|
|
self.assertFalse(no_anchors)
|
|
|
|
def test_direct_multiregion_prism_keeps_each_source_side_face(self) -> None:
|
|
"""Each independently built source region keeps its own final evidence."""
|
|
from engine.cdsl_engine.runtime import prepare_cdsl_execution
|
|
|
|
source_intent = _intent("SWEPT_FACE", allowed=["boundary"])
|
|
cdsl = {
|
|
"schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0",
|
|
"kind": "part", "part_id": "two-direct-circles", "meta": {"unit": "mm"},
|
|
"geometry": {"sketches": [{
|
|
"id": "sketch_F0", "source_sketch_id": "F0",
|
|
"workplane": {
|
|
"origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1],
|
|
},
|
|
"profile": {"type": "analytic_contours", "contours": [
|
|
{"closed": True, "segments": [{
|
|
"type": "circle", "center": [-20, 0], "radius_mm": 5,
|
|
"source_entity_id": "E0",
|
|
}]},
|
|
{"closed": True, "segments": [{
|
|
"type": "circle", "center": [20, 0], "radius_mm": 5,
|
|
"source_entity_id": "E1",
|
|
}]},
|
|
]},
|
|
}]},
|
|
"features": [{
|
|
"id": "f_F1", "atomic_id": "extrude_add_blind", "depends_on": [],
|
|
"sketch_id": "sketch_F0", "params": {"distance_mm": 8, "result_mode": "new_body"},
|
|
}],
|
|
}
|
|
|
|
execution = prepare_cdsl_execution(cdsl)
|
|
self.assertTrue(execution.analysis.runtime_eligible)
|
|
execution.execute_all()
|
|
self.assertEqual(len(execution.session.adapter.body_solids(execution.session.body)), 2)
|
|
|
|
resolutions = {}
|
|
for entity_id in ("E0", "E1"):
|
|
resolution = execution.session.topology.resolve({
|
|
"kind": "face", "owner_feature_id": "f_F1",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**source_intent,
|
|
"source_entity": {"sketch_id": "F0", "entity_id": entity_id},
|
|
},
|
|
}, active_body_id=execution.session.body_id)
|
|
self.assertEqual(resolution.status, "resolved")
|
|
self.assertEqual(resolution.resolution_mode, "kernel_lineage")
|
|
self.assertIsNotNone(resolution.record)
|
|
resolutions[entity_id] = resolution
|
|
|
|
self.assertNotEqual(
|
|
resolutions["E0"].record.record_id,
|
|
resolutions["E1"].record.record_id,
|
|
)
|
|
self.assertNotEqual(
|
|
resolutions["E0"].evidence["body_member"],
|
|
resolutions["E1"].evidence["body_member"],
|
|
)
|
|
active_records = {
|
|
record.record_id for record in execution.session.topology.records()
|
|
if record.body_id and record.body_id.startswith(f"{execution.session.body_id}:")
|
|
}
|
|
side_relations = [
|
|
relation for delta in execution.session.topology.topology_deltas()
|
|
for relation in delta["relations"]
|
|
if relation["source_kind"] == "edge" and relation["result_kind"] == "face"
|
|
]
|
|
self.assertEqual(len(side_relations), 2)
|
|
self.assertTrue(all(
|
|
relation["coverage"] == "complete"
|
|
and relation["lineage_status"] == "proven"
|
|
and len(relation["result_record_ids"]) == 1
|
|
and set(relation["result_record_ids"]).issubset(active_records)
|
|
for relation in side_relations
|
|
))
|
|
|
|
def test_multiregion_imprint_prism_fuse_uses_only_final_history(self) -> None:
|
|
"""A fused IMPRINT region keeps exact outer lineage and rejects a deleted seam."""
|
|
from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter
|
|
from engine.cdsl_engine.sketch_solver import resolve_all_sketches
|
|
|
|
profile = {
|
|
"type": "planar_imprint",
|
|
"source_entities": [
|
|
{"id": "bottom", "curve": {"type": "line", "start": [-5, -5], "end": [5, -5]}},
|
|
{"id": "right", "curve": {"type": "line", "start": [5, -5], "end": [5, 5]}},
|
|
{"id": "top", "curve": {"type": "line", "start": [5, 5], "end": [-5, 5]}},
|
|
{"id": "left", "curve": {"type": "line", "start": [-5, 5], "end": [-5, -5]}},
|
|
{"id": "divider", "curve": {"type": "line", "start": [0, -5], "end": [0, 5]}},
|
|
],
|
|
# A bare source set selects both bounded regions above the divider.
|
|
"selections": [{"source_entity_id": "top", "face_side": 1}],
|
|
}
|
|
resolved = resolve_all_sketches({"geometry": {"sketches": [{
|
|
"id": "imprint", "source_sketch_id": "F0",
|
|
"workplane": {"origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1]},
|
|
"profile": profile,
|
|
}]}})
|
|
adapter = Build123dGeometryAdapter()
|
|
faces, anchor_specs = adapter.faces_for_sketch_with_source_anchors(
|
|
resolved["geometry"]["sketches"][0],
|
|
)
|
|
self.assertEqual(len(faces), 2)
|
|
composed = adapter.extrude_faces_with_composed_topology_delta(faces, (0, 0, 2))
|
|
self.assertIsNotNone(composed)
|
|
body, topology_delta = composed
|
|
self.assertTrue(body.is_valid)
|
|
self.assertAlmostEqual(float(body.volume), 200.0, places=6)
|
|
self.assertEqual(topology_delta.history_status, "proven")
|
|
self.assertEqual(topology_delta.history_reason, "exact_prism_fuse_history")
|
|
self.assertTrue(all(
|
|
adapter._is_result_topology_member(body.wrapped, value)
|
|
for relation in topology_delta.relations
|
|
for value in relation.result_values
|
|
))
|
|
|
|
registry = TopologyRegistry()
|
|
anchors = [
|
|
TopologyRecord(
|
|
f"anchor:{index}", str(spec["kind"]), "f1", value=spec["value"],
|
|
source_entity=spec.get("source_entity"),
|
|
source_entities=tuple(spec.get("source_entities") or ()),
|
|
)
|
|
for index, spec in enumerate(anchor_specs)
|
|
]
|
|
for anchor in anchors:
|
|
registry.register(anchor)
|
|
registry.replace_body_topology(
|
|
"f1", "body:f1", adapter.topology_records(body, "f1", "body:f1"),
|
|
topology_delta=topology_delta, additional_predecessors=anchors,
|
|
)
|
|
|
|
def swept_selector(entity_id: str, multiplicity: str = "all_fragments") -> dict:
|
|
return {
|
|
"kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("SWEPT_FACE", allowed=["boundary", "fragment"], multiplicity=multiplicity),
|
|
"source_entity": {"sketch_id": "F0", "entity_id": entity_id},
|
|
},
|
|
}
|
|
|
|
outer = registry.resolve(swept_selector("left"), active_body_id="body:f1")
|
|
self.assertEqual(outer.status, "resolved")
|
|
self.assertEqual(len(outer.records), 1)
|
|
split_outer = registry.resolve(swept_selector("top"), active_body_id="body:f1")
|
|
self.assertEqual(split_outer.status, "resolved")
|
|
self.assertEqual(len(split_outer.records), 2)
|
|
|
|
# The shared source edge created two intermediate side walls, both
|
|
# deleted by the exact fuse. It must not acquire a nearby final face.
|
|
deleted = registry.resolve(swept_selector("divider"), active_body_id="body:f1")
|
|
self.assertEqual(deleted.status, "not_found")
|
|
self.assertEqual(deleted.diagnostic.code, "selector_kernel_history_missing")
|
|
divider_anchor = next(anchor for anchor in anchors if anchor.source_entity == ("F0", "divider"))
|
|
divider_relation = next(
|
|
relation for relation in topology_delta.relations
|
|
if relation.source_kind == "edge" and relation.result_kind == "face"
|
|
and relation.source_value.IsSame(divider_anchor.value.wrapped)
|
|
)
|
|
self.assertEqual(divider_relation.coverage, "partial")
|
|
self.assertEqual(divider_relation.status, "unknown")
|
|
self.assertEqual(divider_relation.result_values, ())
|
|
|
|
# A split source has two exact physical anchors; the one-result policy
|
|
# stays ambiguous instead of selecting one fragment by geometry.
|
|
ambiguous = registry.resolve(swept_selector("top", "one"), active_body_id="body:f1")
|
|
self.assertEqual(ambiguous.status, "not_found")
|
|
self.assertEqual(ambiguous.diagnostic.code, "selector_ambiguous")
|
|
|
|
for role in ("extrude.start", "extrude.end"):
|
|
caps = registry.resolve({
|
|
"kind": "face", "owner_feature_id": "f1", "output_role": role,
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("CAP_FACE", allowed=["boundary", "fragment"], multiplicity="all_fragments"),
|
|
"output_role": role,
|
|
},
|
|
}, active_body_id="body:f1")
|
|
self.assertEqual(caps.status, "resolved")
|
|
self.assertEqual(len(caps.records), 2)
|
|
|
|
def test_unknown_source_version_is_preserved_then_rejected_by_resolver(self) -> None:
|
|
registry = TopologyRegistry()
|
|
selector = {
|
|
"kind": "face", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intent(version=None),
|
|
}
|
|
|
|
resolution = registry.resolve(selector)
|
|
|
|
self.assertEqual(resolution.status, "not_found")
|
|
self.assertEqual(resolution.diagnostic.code, "selector_query_version_unknown")
|
|
|
|
def test_known_but_unregistered_query_version_is_rejected(self) -> None:
|
|
registry = TopologyRegistry()
|
|
selector = {
|
|
"kind": "face", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intent(version="1512"),
|
|
}
|
|
|
|
resolution = registry.resolve(selector)
|
|
|
|
self.assertEqual(resolution.status, "not_found")
|
|
self.assertEqual(resolution.diagnostic.code, "selector_query_unsupported")
|
|
self.assertEqual(resolution.diagnostic.detail["verified_versions"], ["1511"])
|
|
|
|
def test_missing_or_unregistered_standard_library_revision_is_rejected(self) -> None:
|
|
registry = TopologyRegistry()
|
|
selector = {
|
|
"kind": "face", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intent(),
|
|
}
|
|
|
|
missing = {**selector, "selector_intent": {**selector["selector_intent"], "source_query": {
|
|
"ast": {"call": "CAP_FACE"}, "featurescript_version": "1511",
|
|
}}}
|
|
missing_resolution = registry.resolve(missing)
|
|
self.assertEqual(missing_resolution.status, "not_found")
|
|
self.assertEqual(missing_resolution.diagnostic.code, "selector_query_version_unknown")
|
|
|
|
unregistered = {**selector, "selector_intent": {**selector["selector_intent"], "source_query": {
|
|
**selector["selector_intent"]["source_query"], "standard_library_version": "1512.0",
|
|
}}}
|
|
unregistered_resolution = registry.resolve(unregistered)
|
|
self.assertEqual(unregistered_resolution.status, "not_found")
|
|
self.assertEqual(unregistered_resolution.diagnostic.code, "selector_query_unsupported")
|
|
self.assertEqual(
|
|
unregistered_resolution.diagnostic.detail["verified_standard_libraries"],
|
|
[{"path": "onshape/std/geometry.fs", "version": "1511.0"}],
|
|
)
|
|
|
|
def test_unregistered_query_intent_cannot_fall_back_to_stable_id_or_geometry(self) -> None:
|
|
registry = TopologyRegistry()
|
|
face = object()
|
|
registry.replace_body_topology("base", "body:base", [
|
|
TopologyRecord(
|
|
"base:face", "face", "base", "body:base", value=face,
|
|
geometry={"center_mm": [0.0, 0.0, 0.0]},
|
|
),
|
|
])
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "base", "stable_id": "base:face",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"geometry": {"center_mm": [0.0, 0.0, 0.0]},
|
|
"selector_intent": _intent("LOFT_EDGE_CHAIN", version="1793"),
|
|
}
|
|
|
|
resolution = registry.resolve(selector, active_body_id="body:base")
|
|
|
|
self.assertEqual(resolution.status, "not_found")
|
|
self.assertEqual(resolution.diagnostic.code, "selector_query_unsupported")
|
|
self.assertEqual(resolution.diagnostic.detail["query_family"], "LOFT_EDGE_CHAIN")
|
|
self.assertEqual(resolution.diagnostic.detail["verified_versions"], [])
|
|
|
|
def test_active_stable_id_does_not_replace_provenance_evidence(self) -> None:
|
|
registry = TopologyRegistry()
|
|
source = object()
|
|
registry.replace_body_topology("base", "body:base", [
|
|
TopologyRecord("base:face", "face", "base", "body:base", value=source),
|
|
])
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "base", "stable_id": "base:face",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intent(),
|
|
}
|
|
|
|
resolution = registry.resolve(selector, active_body_id="body:base")
|
|
|
|
self.assertEqual(resolution.status, "not_found")
|
|
self.assertEqual(resolution.diagnostic.code, "selector_kernel_history_missing")
|
|
|
|
def test_output_role_requires_a_relation_allowed_by_its_policy(self) -> None:
|
|
registry = TopologyRegistry()
|
|
profile = object()
|
|
cap = object()
|
|
registry.replace_body_topology("profile", "body:profile", [
|
|
TopologyRecord("profile:face", "face", "profile", "body:profile", value=profile),
|
|
])
|
|
registry.replace_body_topology("extrude", "body:extrude", [
|
|
TopologyRecord("extrude:cap", "face", "extrude", "body:extrude", value=cap),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation("generated", "face", profile, (cap,), output_role="extrude.end"),
|
|
)))
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "extrude", "output_role": "extrude.end",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intent(allowed=["continuation"]),
|
|
}
|
|
|
|
rejected = registry.resolve(selector, active_body_id="body:extrude")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_kernel_history_missing")
|
|
|
|
selector["selector_intent"] = _intent(allowed=["boundary"])
|
|
resolved = registry.resolve(selector, active_body_id="body:extrude")
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "operation_role")
|
|
self.assertEqual(resolved.evidence["result_records"], ["extrude:cap"])
|
|
self.assertEqual(resolved.evidence["relations"][0]["derivation"], "boundary")
|
|
|
|
def test_all_fragments_rejects_a_missing_branch(self) -> None:
|
|
registry = TopologyRegistry()
|
|
source = object()
|
|
first = object()
|
|
missing_second = object()
|
|
registry.replace_body_topology("base", "body:base", [
|
|
TopologyRecord("base:face", "face", "base", "body:base", value=source),
|
|
])
|
|
registry.replace_body_topology("split", "body:split", [
|
|
TopologyRecord("split:first", "face", "split", "body:split", value=first),
|
|
], topology_delta=TopologyDelta("split", (
|
|
TopologyDeltaRelation("modified", "face", source, (first, missing_second)),
|
|
)))
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "base", "stable_id": "base:face",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intent(allowed=["fragment"], multiplicity="all_fragments"),
|
|
}
|
|
|
|
resolution = registry.resolve(selector, active_body_id="body:split")
|
|
|
|
self.assertEqual(resolution.status, "not_found")
|
|
self.assertEqual(resolution.diagnostic.code, "selector_kernel_history_missing")
|
|
|
|
def test_merge_requires_merge_policy_even_with_one_final_record(self) -> None:
|
|
registry = TopologyRegistry()
|
|
first = object()
|
|
second = object()
|
|
merged = object()
|
|
registry.replace_body_topology("base", "body:base", [
|
|
TopologyRecord("base:first", "face", "base", "body:base", value=first),
|
|
TopologyRecord("base:second", "face", "base", "body:base", value=second),
|
|
])
|
|
registry.replace_body_topology("boolean", "body:boolean", [
|
|
TopologyRecord("boolean:merged", "face", "boolean", "body:boolean", value=merged),
|
|
], topology_delta=TopologyDelta("fuse", (
|
|
TopologyDeltaRelation("modified", "face", first, (merged,)),
|
|
TopologyDeltaRelation("modified", "face", second, (merged,)),
|
|
)))
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "base", "stable_id": "base:first",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intent(allowed=["continuation"]),
|
|
}
|
|
|
|
rejected = registry.resolve(selector, active_body_id="body:boolean")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
selector["selector_intent"] = _intent(allowed=["continuation", "merge"])
|
|
resolved = registry.resolve(selector, active_body_id="body:boolean")
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.record.record_id, "boolean:merged")
|
|
|
|
def test_output_role_does_not_bypass_operation_merge_policy(self) -> None:
|
|
registry = TopologyRegistry()
|
|
first = object()
|
|
second = object()
|
|
cap = object()
|
|
registry.replace_body_topology("profiles", "body:profiles", [
|
|
TopologyRecord("profiles:first", "face", "profiles", "body:profiles", value=first),
|
|
TopologyRecord("profiles:second", "face", "profiles", "body:profiles", value=second),
|
|
])
|
|
registry.replace_body_topology("extrude", "body:extrude", [
|
|
TopologyRecord("extrude:cap", "face", "extrude", "body:extrude", value=cap),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation("generated", "face", first, (cap,), output_role="extrude.end"),
|
|
TopologyDeltaRelation("generated", "face", second, (cap,)),
|
|
)))
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "extrude", "output_role": "extrude.end",
|
|
"source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intent(allowed=["boundary"]),
|
|
}
|
|
|
|
rejected = registry.resolve(selector, active_body_id="body:extrude")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
selector["selector_intent"] = _intent(allowed=["boundary", "merge"])
|
|
resolved = registry.resolve(selector, active_body_id="body:extrude")
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.record.record_id, "extrude:cap")
|
|
|
|
def test_cross_kind_relation_keeps_distinct_source_and_result_evidence(self) -> None:
|
|
registry = TopologyRegistry()
|
|
vertex = object()
|
|
edge = object()
|
|
registry.replace_body_topology("base", "body:base", [
|
|
TopologyRecord("base:vertex", "vertex", "base", "body:base", value=vertex),
|
|
])
|
|
registry.replace_body_topology("fillet", "body:fillet", [
|
|
TopologyRecord("fillet:edge", "edge", "fillet", "body:fillet", value=edge),
|
|
], topology_delta=TopologyDelta("fillet", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", vertex, (edge,),
|
|
source_kind="vertex", result_kind="edge",
|
|
),
|
|
)))
|
|
|
|
relation = registry.topology_deltas()[0]["relations"][0]
|
|
lineage = registry.lineage()[0]
|
|
self.assertEqual(relation["source_record_ids"], ["base:vertex"])
|
|
self.assertEqual(relation["result_record_ids"], ["fillet:edge"])
|
|
self.assertEqual((relation["source_kind"], relation["result_kind"]), ("vertex", "edge"))
|
|
self.assertEqual((lineage.source_kind, lineage.result_kind), ("vertex", "edge"))
|
|
|
|
def _source_qualified_section_registry(self, *, sections: int = 1, qualified: bool = True) -> TopologyRegistry:
|
|
registry = TopologyRegistry()
|
|
raw_cap, cap = object(), object()
|
|
anchor, swept = object(), object()
|
|
registry.replace_body_topology("seed", "body:seed", [
|
|
TopologyRecord("seed:cap", "face", "seed", "body:seed", {"center_mm": [-1, 0, 0]}, raw_cap),
|
|
])
|
|
registry.replace_body_topology("f1", "body:f1", [
|
|
TopologyRecord("f1:cap", "face", "f1", "body:f1", {"center_mm": [0, 0, 0]}, cap),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation("generated", "face", raw_cap, (cap,), output_role="extrude.end"),
|
|
)))
|
|
source_anchor = TopologyRecord(
|
|
"anchor:f2", "edge", "f2", value=anchor, source_entity=("S2", "E2"),
|
|
)
|
|
registry.register(source_anchor)
|
|
registry.replace_body_topology("f2", "body:f2", [
|
|
TopologyRecord("f2:swept", "face", "f2", "body:f2", {"center_mm": [1, 0, 0]}, swept),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", anchor, (swept,), source_kind="edge", result_kind="face", derivation="boundary",
|
|
),
|
|
)), additional_predecessors=[source_anchor])
|
|
result_edges = [object() for _index in range(sections)]
|
|
registry.replace_body_topology("boolean", "body:boolean", [
|
|
TopologyRecord(f"boolean:section:{index}", "edge", "boolean", "body:boolean", value=edge)
|
|
for index, edge in enumerate(result_edges)
|
|
], topology_delta=TopologyDelta(
|
|
"subtract",
|
|
section_values=tuple(result_edges) if not qualified else (),
|
|
section_relations=(
|
|
tuple(TopologySectionRelation((cap, swept), edge) for edge in result_edges)
|
|
if qualified else ()
|
|
),
|
|
), additional_predecessors=[
|
|
TopologyRecord("f1:cap", "face", "f1", "body:f1", {"center_mm": [0, 0, 0]}, cap, output_roles=("extrude.end",)),
|
|
])
|
|
return registry
|
|
|
|
def test_intersect_resolves_only_one_source_qualified_section_edge(self) -> None:
|
|
registry = self._source_qualified_section_registry()
|
|
resolved = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intersection_intent(),
|
|
}, active_body_id="body:boolean")
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.record.record_id, "boolean:section:0")
|
|
self.assertEqual(resolved.resolution_mode, "kernel_intersection")
|
|
section_relation = next(item for item in resolved.evidence["relations"] if item["derivation"] == "intersection")
|
|
self.assertEqual(set(section_relation["source_record_ids"]), {"f1:cap", "f2:swept"})
|
|
|
|
def test_intersect_uses_the_builder_cap_snapshot_not_a_cached_continuation_role(self) -> None:
|
|
registry = self._source_qualified_section_registry()
|
|
cap = next(record for record in registry.records() if record.record_id == "f1:cap")
|
|
continued = object()
|
|
registry.replace_body_topology("f3", "body:f3", [
|
|
TopologyRecord(
|
|
"f3:continued-cap", "face", "f3", "body:f3", value=continued,
|
|
owner_feature_ids=("f1",), output_roles=("extrude.end",),
|
|
),
|
|
], topology_delta=TopologyDelta("subtract", (
|
|
TopologyDeltaRelation("preserved", "face", cap.value, (continued,)),
|
|
)))
|
|
|
|
resolved = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intersection_intent(),
|
|
}, active_body_id="body:boolean")
|
|
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "kernel_intersection")
|
|
self.assertEqual(resolved.evidence["source_records"][0], "f1:cap")
|
|
|
|
def test_intersect_rejects_duplicate_cap_builder_role_facts(self) -> None:
|
|
registry = self._source_qualified_section_registry()
|
|
registry._topology_deltas.append({
|
|
"feature_id": "f1",
|
|
"operation": "extrude",
|
|
"relations": [{
|
|
"output_role": "extrude.end",
|
|
"coverage": "complete",
|
|
"lineage_status": "proven",
|
|
"result_kind": "face",
|
|
"result_record_ids": ["f1:cap"],
|
|
}],
|
|
})
|
|
|
|
result = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intersection_intent(),
|
|
}, active_body_id="body:boolean")
|
|
|
|
self.assertEqual(result.status, "not_found")
|
|
self.assertEqual(result.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
def test_intersect_rejects_unqualified_section_edge(self) -> None:
|
|
registry = self._source_qualified_section_registry(qualified=False)
|
|
result = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intersection_intent(),
|
|
}, active_body_id="body:boolean")
|
|
|
|
self.assertEqual(result.status, "not_found")
|
|
self.assertEqual(result.diagnostic.code, "selector_kernel_history_missing")
|
|
|
|
def test_intersect_rejects_multiple_source_qualified_section_edges(self) -> None:
|
|
registry = self._source_qualified_section_registry(sections=2)
|
|
result = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intersection_intent(),
|
|
}, active_body_id="body:boolean")
|
|
|
|
self.assertEqual(result.status, "not_found")
|
|
self.assertEqual(result.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
def test_session_intersection_vertex_uses_the_common_provenance_gate(self) -> None:
|
|
"""Deferred provenance must not reach the session geometry shortcut."""
|
|
from engine.cdsl_engine.session import ExecutionSession
|
|
|
|
class RecordingAdapter:
|
|
def __init__(self) -> None:
|
|
self.intersection_calls = 0
|
|
|
|
def intersection_vertex(self, body, face_sets):
|
|
self.intersection_calls += 1
|
|
return object()
|
|
|
|
def vertex_coordinates(self, vertex):
|
|
return (1.0, 2.0, 3.0)
|
|
|
|
adapter = RecordingAdapter()
|
|
session = ExecutionSession(sketches={}, nodes={}, adapter=adapter)
|
|
session.body = object()
|
|
session.body_id = "body:base"
|
|
session._intersection_component_records = lambda _selector: [TopologyRecord(
|
|
"base:face", "face", "base", "body:base", value=object(),
|
|
)]
|
|
base_selector = {
|
|
"kind": "vertex", "owner_feature_id": "cut",
|
|
"intersection_of": [{"kind": "face"}, {"kind": "face"}],
|
|
}
|
|
|
|
for intent, expected_code in (
|
|
(_intent("INTERSECT", allowed=["intersection"], multiplicity="none"), "selector_query_unsupported"),
|
|
(_intent("INTERSECT", version="1793", allowed=["intersection"]), "selector_query_unsupported"),
|
|
(_intent("INTERSECT", version=None, allowed=["intersection"]), "selector_query_version_unknown"),
|
|
):
|
|
with self.subTest(intent=intent):
|
|
result = session.resolve({**base_selector, "selector_intent": intent})
|
|
self.assertEqual(result.status, "not_found")
|
|
self.assertEqual(result.diagnostic.code, expected_code)
|
|
self.assertEqual(adapter.intersection_calls, 0)
|
|
|
|
legacy_result = session.resolve(base_selector)
|
|
self.assertEqual(legacy_result.status, "resolved")
|
|
self.assertEqual(adapter.intersection_calls, 1)
|
|
|
|
def test_transient_prism_snapshot_is_historical_not_selectable(self) -> None:
|
|
registry = TopologyRegistry()
|
|
source_edge = object()
|
|
tool_face = object()
|
|
anchor = TopologyRecord(
|
|
"anchor:tool:E0", "edge", "tool", value=source_edge,
|
|
source_entity=("tool-sketch", "E0"),
|
|
)
|
|
registered = registry.register_transient_snapshot(
|
|
"tool",
|
|
"transient:tool",
|
|
[TopologyRecord("transient:tool:face:0", "face", "tool", "transient:tool", value=tool_face)],
|
|
topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", source_edge, (tool_face,),
|
|
source_kind="edge", result_kind="face", derivation="boundary",
|
|
),
|
|
)),
|
|
anchors=[anchor],
|
|
)
|
|
|
|
self.assertTrue(all(record.transient for record in registered))
|
|
self.assertTrue(all(record.transient for record in registry.records()))
|
|
self.assertEqual(registry.topology_deltas()[0]["output_snapshot_ids"], ["transient:tool"])
|
|
self.assertEqual(
|
|
registry.resolve({"kind": "face", "owner_feature_id": "tool"}).status,
|
|
"not_found",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "complete direct-prism history"):
|
|
registry.register_transient_snapshot(
|
|
"partial-tool",
|
|
"transient:partial-tool",
|
|
[TopologyRecord("transient:partial-tool:face:0", "face", "partial-tool", value=object())],
|
|
topology_delta=TopologyDelta("extrude", history_status="partial"),
|
|
anchors=[anchor],
|
|
)
|
|
|
|
def test_intersect_rejects_section_edge_outside_active_member(self) -> None:
|
|
registry = self._source_qualified_section_registry()
|
|
result = registry.resolve({
|
|
"kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": _intersection_intent(),
|
|
}, active_body_id="body:later")
|
|
|
|
self.assertEqual(result.status, "not_found")
|
|
self.assertEqual(result.diagnostic.code, "selector_body_member_inactive")
|
|
|
|
def test_blend_transition_evidence_requires_a_complete_three_handle_relation(self) -> None:
|
|
registry = TopologyRegistry()
|
|
source_edge, source_face = object(), object()
|
|
patch_face, modified_face, final_edge = object(), object(), object()
|
|
registry.replace_body_topology("base", "body:base", [
|
|
TopologyRecord("base:edge", "edge", "base", "body:base", value=source_edge),
|
|
TopologyRecord("base:face", "face", "base", "body:base", value=source_face),
|
|
])
|
|
registry.replace_body_topology("chamfer", "body:chamfer", [
|
|
TopologyRecord("chamfer:patch", "face", "chamfer", "body:chamfer", value=patch_face),
|
|
TopologyRecord("chamfer:modified", "face", "chamfer", "body:chamfer", value=modified_face),
|
|
TopologyRecord("chamfer:edge", "edge", "chamfer", "body:chamfer", value=final_edge),
|
|
], topology_delta=TopologyDelta("chamfer", blend_relations=(
|
|
TopologyBlendRelation(
|
|
source_edge, source_face, patch_face, modified_face, (final_edge,),
|
|
),
|
|
)))
|
|
|
|
relation = registry.topology_deltas()[0]["relations"][0]
|
|
self.assertEqual(relation["status"], "exact_blend_boundary")
|
|
self.assertEqual(relation["source_record_ids"], ["base:edge"])
|
|
self.assertEqual(relation["blend_into_source_record_ids"], ["base:face"])
|
|
self.assertEqual(relation["blend_into_result_record_ids"], ["chamfer:modified"])
|
|
self.assertEqual(relation["patch_face_record_ids"], ["chamfer:patch"])
|
|
self.assertEqual(relation["result_record_ids"], ["chamfer:edge"])
|
|
self.assertNotIn("base:edge", registry._successors)
|
|
self.assertFalse(registry.lineage())
|
|
|
|
incomplete = TopologyRegistry()
|
|
incomplete.replace_body_topology("base", "body:base", [
|
|
TopologyRecord("base:edge", "edge", "base", "body:base", value=source_edge),
|
|
TopologyRecord("base:face", "face", "base", "body:base", value=source_face),
|
|
])
|
|
incomplete.replace_body_topology("chamfer", "body:chamfer", [
|
|
TopologyRecord("chamfer:patch", "face", "chamfer", "body:chamfer", value=patch_face),
|
|
TopologyRecord("chamfer:modified", "face", "chamfer", "body:chamfer", value=modified_face),
|
|
TopologyRecord("chamfer:edge", "edge", "chamfer", "body:chamfer", value=final_edge),
|
|
], topology_delta=TopologyDelta("chamfer", blend_relations=(
|
|
TopologyBlendRelation(
|
|
source_edge, source_face, patch_face, modified_face, (), coverage="partial", status="unknown",
|
|
),
|
|
)))
|
|
self.assertEqual(
|
|
incomplete.topology_deltas()[0]["relations"][0]["status"],
|
|
"incomplete_blend_boundary",
|
|
)
|
|
|
|
def test_blend_edge_resolver_requires_one_complete_active_source_pair(self) -> None:
|
|
def registry_with_blends(
|
|
blend_factory, *, face_role: str = "extrude.end",
|
|
) -> TopologyRegistry:
|
|
registry = TopologyRegistry()
|
|
anchor_value, cap_edge_value, cap_face_value = object(), object(), object()
|
|
anchor = TopologyRecord(
|
|
"anchor:F0:E0", "edge", "f1", value=anchor_value,
|
|
source_entity=("F0", "E0"),
|
|
)
|
|
registry.register(anchor)
|
|
registry.replace_body_topology("f1", "body:f1", [
|
|
TopologyRecord("f1:cap-edge", "edge", "f1", "body:f1", value=cap_edge_value),
|
|
TopologyRecord(
|
|
"f1:cap-face", "face", "f1", "body:f1", value=cap_face_value,
|
|
output_roles=(face_role,),
|
|
),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", anchor_value, (cap_edge_value,),
|
|
source_kind="edge", result_kind="edge", derivation="boundary",
|
|
output_role="extrude.end",
|
|
),
|
|
)), additional_predecessors=[anchor])
|
|
blends = blend_factory(cap_edge_value, cap_face_value)
|
|
current = []
|
|
for index, blend in enumerate(blends):
|
|
current.extend((
|
|
TopologyRecord(f"f2:patch:{index}", "face", "f2", "body:f2", value=blend.patch_face_value),
|
|
TopologyRecord(f"f2:modified:{index}", "face", "f2", "body:f2", value=blend.blend_into_result_value),
|
|
*(
|
|
TopologyRecord(f"f2:edge:{index}:{edge_index}", "edge", "f2", "body:f2", value=value)
|
|
for edge_index, value in enumerate(blend.result_values)
|
|
),
|
|
))
|
|
registry.replace_body_topology(
|
|
"f2", "body:f2", current,
|
|
topology_delta=TopologyDelta("chamfer", blend_relations=blends),
|
|
)
|
|
return registry
|
|
|
|
patch, modified, final_edge = object(), object(), object()
|
|
# Bind the actual snapshot handles. No geometry and no unrelated
|
|
# owner can satisfy this resolver path.
|
|
registry = registry_with_blends(
|
|
lambda cap_edge, cap_face: (
|
|
TopologyBlendRelation(cap_edge, cap_face, patch, modified, (final_edge,)),
|
|
),
|
|
)
|
|
selector = {
|
|
"kind": "edge", "owner_feature_id": "f2", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("BLEND_EDGE", allowed=["boundary"]),
|
|
"kind": "edge", "evidence": "kernel_history",
|
|
"blend_sources": {
|
|
"edge": {
|
|
"query_family": "CAP_EDGE", "owner_feature_id": "f1",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
"lineage_role": "extrude.end",
|
|
},
|
|
"face": {
|
|
"query_family": "CAP_FACE", "owner_feature_id": "f1",
|
|
"output_role": "extrude.end",
|
|
},
|
|
},
|
|
},
|
|
}
|
|
resolved = registry.resolve(selector, active_body_id="body:f2")
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "blend_boundary")
|
|
self.assertEqual(resolved.record.record_id, "f2:edge:0:0")
|
|
|
|
inactive = registry.resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(inactive.status, "not_found")
|
|
self.assertEqual(inactive.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
other_edge = object()
|
|
ambiguous = registry_with_blends(
|
|
lambda cap_edge, cap_face: (
|
|
TopologyBlendRelation(cap_edge, cap_face, patch, modified, (final_edge,)),
|
|
TopologyBlendRelation(cap_edge, cap_face, object(), object(), (other_edge,)),
|
|
),
|
|
).resolve(selector, active_body_id="body:f2")
|
|
self.assertEqual(ambiguous.status, "not_found")
|
|
self.assertEqual(ambiguous.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
incomplete = registry_with_blends(
|
|
lambda cap_edge, cap_face: (
|
|
TopologyBlendRelation(cap_edge, cap_face, patch, modified, (), coverage="partial", status="unknown"),
|
|
),
|
|
).resolve(selector, active_body_id="body:f2")
|
|
self.assertEqual(incomplete.status, "not_found")
|
|
self.assertEqual(incomplete.diagnostic.code, "selector_kernel_history_missing")
|
|
|
|
wrong_role = registry_with_blends(
|
|
lambda cap_edge, cap_face: (
|
|
TopologyBlendRelation(cap_edge, cap_face, patch, modified, (final_edge,)),
|
|
),
|
|
face_role="extrude.start",
|
|
).resolve(selector, active_body_id="body:f2")
|
|
self.assertEqual(wrong_role.status, "not_found")
|
|
self.assertEqual(wrong_role.diagnostic.code, "selector_source_unavailable")
|
|
|
|
def test_blend_edge_resolver_accepts_one_exact_direct_swept_face_source(self) -> None:
|
|
registry = TopologyRegistry()
|
|
anchor_value, cap_edge_value, swept_face_value = object(), object(), object()
|
|
patch, modified, final_edge = object(), object(), object()
|
|
anchor = TopologyRecord(
|
|
"anchor:F0:E0", "edge", "f1", value=anchor_value,
|
|
source_entity=("F0", "E0"),
|
|
)
|
|
registry.register(anchor)
|
|
registry.replace_body_topology("f1", "body:f1", [
|
|
TopologyRecord("f1:cap-edge", "edge", "f1", "body:f1", value=cap_edge_value),
|
|
TopologyRecord("f1:swept-face", "face", "f1", "body:f1", value=swept_face_value),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", anchor_value, (cap_edge_value,),
|
|
source_kind="edge", result_kind="edge", derivation="boundary",
|
|
output_role="extrude.end",
|
|
),
|
|
TopologyDeltaRelation(
|
|
"generated", "face", anchor_value, (swept_face_value,),
|
|
source_kind="edge", result_kind="face", derivation="boundary",
|
|
),
|
|
)), additional_predecessors=[anchor])
|
|
registry.replace_body_topology("f2", "body:f2", [
|
|
TopologyRecord("f2:patch", "face", "f2", "body:f2", value=patch),
|
|
TopologyRecord("f2:modified", "face", "f2", "body:f2", value=modified),
|
|
TopologyRecord("f2:edge", "edge", "f2", "body:f2", value=final_edge),
|
|
], topology_delta=TopologyDelta("fillet", blend_relations=(
|
|
TopologyBlendRelation(cap_edge_value, swept_face_value, patch, modified, (final_edge,)),
|
|
)))
|
|
selector = {
|
|
"kind": "edge", "owner_feature_id": "f2", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("BLEND_EDGE", allowed=["boundary"]),
|
|
"kind": "edge", "evidence": "kernel_history",
|
|
"blend_sources": {
|
|
"edge": {
|
|
"query_family": "CAP_EDGE", "owner_feature_id": "f1",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
"lineage_role": "extrude.end",
|
|
},
|
|
"face": {
|
|
"query_family": "SWEPT_FACE", "owner_feature_id": "f1",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
resolved = registry.resolve(selector, active_body_id="body:f2")
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "blend_boundary")
|
|
self.assertEqual(resolved.record.record_id, "f2:edge")
|
|
|
|
selector["selector_intent"]["blend_sources"]["face"]["source_entity"] = {
|
|
"sketch_id": "F0", "entity_id": "missing",
|
|
}
|
|
rejected = registry.resolve(selector, active_body_id="body:f2")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_source_unavailable")
|
|
|
|
def test_blend_face_resolver_requires_one_active_generated_patch(self) -> None:
|
|
registry = TopologyRegistry()
|
|
anchor_value, cap_edge_value, patch_value = object(), object(), object()
|
|
anchor = TopologyRecord(
|
|
"anchor:F0:E0", "edge", "f1", value=anchor_value,
|
|
source_entity=("F0", "E0"),
|
|
)
|
|
registry.register(anchor)
|
|
registry.replace_body_topology("f1", "body:f1", [
|
|
TopologyRecord("f1:cap-edge", "edge", "f1", "body:f1", value=cap_edge_value),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", anchor_value, (cap_edge_value,),
|
|
source_kind="edge", result_kind="edge", derivation="boundary",
|
|
output_role="extrude.end",
|
|
),
|
|
)), additional_predecessors=[anchor])
|
|
registry.replace_body_topology("f2", "body:f2", [
|
|
TopologyRecord("f2:patch", "face", "f2", "body:f2", value=patch_value),
|
|
], topology_delta=TopologyDelta("chamfer", (
|
|
TopologyDeltaRelation(
|
|
"generated", "edge", cap_edge_value, (patch_value,),
|
|
source_kind="edge", result_kind="face", derivation="boundary",
|
|
),
|
|
)))
|
|
selector = {
|
|
"kind": "face", "owner_feature_id": "f2", "source": "runtime_snapshot", "confidence": 1.0,
|
|
"selector_intent": {
|
|
**_intent("BLEND_FACE", allowed=["boundary"]),
|
|
"kind": "face", "evidence": "kernel_history",
|
|
"blend_face_source": {
|
|
"query_family": "CAP_EDGE", "owner_feature_id": "f1",
|
|
"source_entity": {"sketch_id": "F0", "entity_id": "E0"},
|
|
"lineage_role": "extrude.end",
|
|
},
|
|
},
|
|
}
|
|
resolved = registry.resolve(selector, active_body_id="body:f2")
|
|
self.assertEqual(resolved.status, "resolved")
|
|
self.assertEqual(resolved.resolution_mode, "blend_patch_face")
|
|
self.assertEqual(resolved.record.record_id, "f2:patch")
|
|
|
|
inactive = registry.resolve(selector, active_body_id="body:f1")
|
|
self.assertEqual(inactive.status, "not_found")
|
|
self.assertEqual(inactive.diagnostic.code, "selector_relation_non_unique")
|
|
|
|
ambiguous = TopologyRegistry()
|
|
ambiguous.register(anchor)
|
|
ambiguous.replace_body_topology("f1", "body:f1", [
|
|
TopologyRecord("f1:cap-edge", "edge", "f1", "body:f1", value=cap_edge_value),
|
|
], topology_delta=TopologyDelta("extrude", (
|
|
TopologyDeltaRelation("generated", "edge", anchor_value, (cap_edge_value,), source_kind="edge", result_kind="edge", derivation="boundary", output_role="extrude.end"),
|
|
)), additional_predecessors=[anchor])
|
|
ambiguous.replace_body_topology("f2", "body:f2", [
|
|
TopologyRecord("f2:patch:a", "face", "f2", "body:f2", value=patch_value),
|
|
TopologyRecord("f2:patch:b", "face", "f2", "body:f2", value=object()),
|
|
], topology_delta=TopologyDelta("chamfer", (
|
|
TopologyDeltaRelation("generated", "edge", cap_edge_value, (patch_value, object()), source_kind="edge", result_kind="face", derivation="boundary", coverage="partial", status="unknown"),
|
|
)))
|
|
rejected = ambiguous.resolve(selector, active_body_id="body:f2")
|
|
self.assertEqual(rejected.status, "not_found")
|
|
self.assertEqual(rejected.diagnostic.code, "selector_kernel_history_missing")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|