(graph: Dict[str, Any], base_path: str, errors: List[str])
| 23 | |
| 24 | |
| 25 | def _analyze_graph(graph: Dict[str, Any], base_path: str, errors: List[str]) -> None: |
| 26 | # Majority voting graphs are skipped for start/end structure checks |
| 27 | is_mv = graph.get("is_majority_voting", False) |
| 28 | if is_mv: |
| 29 | return |
| 30 | |
| 31 | nodes = _node_ids(graph) |
| 32 | node_set = set(nodes) |
| 33 | |
| 34 | # Validate provided start/end (if any) reference existing nodes |
| 35 | # start = graph.get("start") |
| 36 | end = graph.get("end") |
| 37 | # if start is not None and start not in node_set: |
| 38 | # errors.append(f"{base_path}.start references unknown node id '{start}'") |
| 39 | |
| 40 | # Normalize to list |
| 41 | if end is not None: |
| 42 | if isinstance(end, str): |
| 43 | end_list = [end] |
| 44 | elif isinstance(end, list): |
| 45 | end_list = end |
| 46 | else: |
| 47 | errors.append(f"{base_path}.end must be a string or list of strings") |
| 48 | return |
| 49 | |
| 50 | # Check each node ID in the end list |
| 51 | for end_node_id in end_list: |
| 52 | if not isinstance(end_node_id, str): |
| 53 | errors.append( |
| 54 | f"{base_path}.end contains non-string element: {end_node_id}" |
| 55 | ) |
| 56 | elif end_node_id not in node_set: |
| 57 | errors.append( |
| 58 | f"{base_path}.end references unknown node id '{end_node_id}'" |
| 59 | ) |
| 60 | |
| 61 | # Compute in/out degrees within this graph scope |
| 62 | indeg = {nid: 0 for nid in nodes} |
| 63 | outdeg = {nid: 0 for nid in nodes} |
| 64 | for e in _edge_list(graph): |
| 65 | frm = e.get("from") |
| 66 | to = e.get("to") |
| 67 | if frm in outdeg: |
| 68 | outdeg[frm] += 1 |
| 69 | if to in indeg: |
| 70 | indeg[to] += 1 |
| 71 | |
| 72 | # sources = [nid for nid in nodes if indeg.get(nid, 0) == 0] |
| 73 | sinks = [nid for nid in nodes if outdeg.get(nid, 0) == 0] |
| 74 | |
| 75 | # # Rule: |
| 76 | # # - A non-cyclic (sub)graph should have exactly one natural source AND exactly one natural sink. |
| 77 | # # - Otherwise (e.g., multiple sources/sinks or cycles -> none), require explicit start or end. |
| 78 | # has_unique_source = len(sources) == 1 |
| 79 | # has_unique_sink = len(sinks) == 1 |
| 80 | # if not (has_unique_source and has_unique_sink): |
| 81 | # if start is None and end is None: |
| 82 | # errors.append( |
no test coverage detected