|
1 | 1 | import copy |
| 2 | +import os |
| 3 | +import warnings |
2 | 4 |
|
3 | 5 | from hypergraphx.exceptions import ( |
4 | 6 | InvalidParameterError, |
@@ -45,6 +47,10 @@ def populate_from_dict(self, data): |
45 | 47 | self._empty_edges = data.get("empty_edges", {}) |
46 | 48 | self._populate_adjacency_data(data) |
47 | 49 | self._populate_extra_data(data) |
| 50 | + # If the implementation provides invariant validation, run it optionally. |
| 51 | + maybe_validate = getattr(self, "_maybe_validate_invariants", None) |
| 52 | + if callable(maybe_validate): |
| 53 | + maybe_validate() |
48 | 54 |
|
49 | 55 | def expose_attributes_for_hashing(self): |
50 | 56 | edges = [] |
@@ -165,6 +171,107 @@ def _validate_metadata_dict(self, metadata, label): |
165 | 171 | if not isinstance(metadata, dict): |
166 | 172 | raise InvalidParameterError(f"{label} metadata must be a dict.") |
167 | 173 |
|
| 174 | + # Invariants / debug helpers |
| 175 | + def _debug_invariants_enabled(self) -> bool: |
| 176 | + """ |
| 177 | + Invariant checks can be expensive. Enable them explicitly via: |
| 178 | + - running Python without -O (i.e. __debug__ is True) AND |
| 179 | + - setting HGX_DEBUG_INVARIANTS=1/true/yes/on. |
| 180 | + """ |
| 181 | + if not __debug__: |
| 182 | + return False |
| 183 | + val = os.getenv("HGX_DEBUG_INVARIANTS", "") |
| 184 | + return val.strip().lower() in {"1", "true", "yes", "on"} |
| 185 | + |
| 186 | + def validate_invariants(self) -> None: |
| 187 | + """Public hook to validate internal consistency (useful in debugging).""" |
| 188 | + self._validate_invariants() |
| 189 | + |
| 190 | + def _validate_invariants(self) -> None: |
| 191 | + """ |
| 192 | + Validate internal data-structure consistency. |
| 193 | +
|
| 194 | + This is intended for debugging and test/dev environments. |
| 195 | + """ |
| 196 | + # Edge id <-> edge key bijection. |
| 197 | + if len(self._edge_list) != len(self._reverse_edge_list): |
| 198 | + raise RuntimeError( |
| 199 | + "Invariant violated: edge_list and reverse_edge_list size mismatch." |
| 200 | + ) |
| 201 | + for edge_key, edge_id in self._edge_list.items(): |
| 202 | + if self._reverse_edge_list.get(edge_id) != edge_key: |
| 203 | + raise RuntimeError( |
| 204 | + "Invariant violated: edge_id <-> edge_key mapping is not a bijection." |
| 205 | + ) |
| 206 | + |
| 207 | + valid_edge_ids = set(self._reverse_edge_list.keys()) |
| 208 | + |
| 209 | + # Weights and edge metadata must align to existing edge_ids. |
| 210 | + for edge_id in self._weights.keys(): |
| 211 | + if edge_id not in valid_edge_ids: |
| 212 | + raise RuntimeError( |
| 213 | + "Invariant violated: weights contain unknown edge_id." |
| 214 | + ) |
| 215 | + for edge_id in self._edge_metadata.keys(): |
| 216 | + if edge_id not in valid_edge_ids: |
| 217 | + raise RuntimeError( |
| 218 | + "Invariant violated: edge_metadata contain unknown edge_id." |
| 219 | + ) |
| 220 | + |
| 221 | + # Adjacency lists contain only valid edge_ids and only reference known nodes. |
| 222 | + for name, adj in self._adjacency_maps().items(): |
| 223 | + for node, edge_ids in adj.items(): |
| 224 | + if node not in self._node_metadata: |
| 225 | + raise RuntimeError( |
| 226 | + f"Invariant violated: adjacency map {name!r} references unknown node." |
| 227 | + ) |
| 228 | + for edge_id in edge_ids: |
| 229 | + if edge_id not in valid_edge_ids: |
| 230 | + raise RuntimeError( |
| 231 | + f"Invariant violated: adjacency map {name!r} contains unknown edge_id." |
| 232 | + ) |
| 233 | + |
| 234 | + # Incidence metadata should refer to existing edges/nodes. |
| 235 | + for (edge_key, node), meta in self._incidences_metadata.items(): |
| 236 | + if edge_key not in self._edge_list: |
| 237 | + raise RuntimeError( |
| 238 | + "Invariant violated: incidence metadata references unknown edge_key." |
| 239 | + ) |
| 240 | + if node not in self._node_metadata: |
| 241 | + raise RuntimeError( |
| 242 | + "Invariant violated: incidence metadata references unknown node." |
| 243 | + ) |
| 244 | + if meta is not None and not isinstance(meta, dict): |
| 245 | + raise RuntimeError( |
| 246 | + "Invariant violated: incidence metadata must be a dict." |
| 247 | + ) |
| 248 | + |
| 249 | + def _maybe_validate_invariants(self) -> None: |
| 250 | + if self._debug_invariants_enabled(): |
| 251 | + self._validate_invariants() |
| 252 | + |
| 253 | + def _allow_unsafe_setters(self) -> bool: |
| 254 | + val = os.getenv("HGX_ALLOW_UNSAFE_SETTERS", "") |
| 255 | + return val.strip().lower() in {"1", "true", "yes", "on"} |
| 256 | + |
| 257 | + def _guard_unsafe_setter(self, name: str) -> None: |
| 258 | + """ |
| 259 | + Guard "invariant grenade" setters that can put the object into an inconsistent state. |
| 260 | +
|
| 261 | + By default they are blocked. Set HGX_ALLOW_UNSAFE_SETTERS=1 to bypass. |
| 262 | + """ |
| 263 | + if not self._allow_unsafe_setters(): |
| 264 | + raise InvalidParameterError( |
| 265 | + f"{name} is an unsafe operation and is disabled by default. " |
| 266 | + "Use populate_from_dict()/load_* APIs instead, or set " |
| 267 | + "HGX_ALLOW_UNSAFE_SETTERS=1 if you really know what you are doing." |
| 268 | + ) |
| 269 | + warnings.warn( |
| 270 | + f"{name} is unsafe and deprecated; it may be removed in a future release.", |
| 271 | + DeprecationWarning, |
| 272 | + stacklevel=3, |
| 273 | + ) |
| 274 | + |
168 | 275 | # Core node methods |
169 | 276 | def add_node(self, node, metadata=None): |
170 | 277 | """Add a node to the hypergraph if it does not already exist.""" |
|
0 commit comments