Utility class that implements an entity map as per the unified test format specification.
| 218 | |
| 219 | |
| 220 | class EntityMapUtil: |
| 221 | """Utility class that implements an entity map as per the unified |
| 222 | test format specification. |
| 223 | """ |
| 224 | |
| 225 | def __init__(self, test_class): |
| 226 | self._entities: Dict[str, Any] = {} |
| 227 | self._listeners: Dict[str, EventListenerUtil] = {} |
| 228 | self._session_lsids: Dict[str, Mapping[str, Any]] = {} |
| 229 | self.test: UnifiedSpecTestMixinV1 = test_class |
| 230 | |
| 231 | def __contains__(self, item): |
| 232 | return item in self._entities |
| 233 | |
| 234 | def __len__(self): |
| 235 | return len(self._entities) |
| 236 | |
| 237 | def __getitem__(self, item): |
| 238 | try: |
| 239 | return self._entities[item] |
| 240 | except KeyError: |
| 241 | self.test.fail(f"Could not find entity named {item} in map") |
| 242 | |
| 243 | def __setitem__(self, key, value): |
| 244 | if not isinstance(key, str): |
| 245 | self.test.fail("Expected entity name of type str, got %s" % (type(key))) |
| 246 | |
| 247 | if key in self._entities: |
| 248 | self.test.fail(f"Entity named {key} already in map") |
| 249 | |
| 250 | self._entities[key] = value |
| 251 | |
| 252 | def _handle_placeholders(self, spec: dict, current: dict, path: str) -> Any: |
| 253 | if "$$placeholder" in current: |
| 254 | if path not in PLACEHOLDER_MAP: |
| 255 | raise ValueError(f"Could not find a placeholder value for {path}") |
| 256 | return PLACEHOLDER_MAP[path] |
| 257 | |
| 258 | # Distinguish between temp and non-temp aws credentials. |
| 259 | if path.endswith("/kmsProviders/aws") and "sessionToken" in current: |
| 260 | path = path.replace("aws", "aws_temp") |
| 261 | |
| 262 | for key in list(current): |
| 263 | value = current[key] |
| 264 | if isinstance(value, dict): |
| 265 | subpath = f"{path}/{key}" |
| 266 | current[key] = self._handle_placeholders(spec, value, subpath) |
| 267 | return current |
| 268 | |
| 269 | def _create_entity(self, entity_spec, uri=None): |
| 270 | if len(entity_spec) != 1: |
| 271 | self.test.fail(f"Entity spec {entity_spec} did not contain exactly one top-level key") |
| 272 | |
| 273 | entity_type, spec = next(iter(entity_spec.items())) |
| 274 | spec = self._handle_placeholders(spec, spec, "") |
| 275 | if entity_type == "client": |
| 276 | kwargs: dict = {} |
| 277 | observe_events = spec.get("observeEvents", []) |