(self, entity_spec, uri=None)
| 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", []) |
| 278 | |
| 279 | if "autoEncryptOpts" in spec: |
| 280 | auto_encrypt_opts = spec["autoEncryptOpts"].copy() |
| 281 | auto_encrypt_kwargs: dict = dict(kms_tls_options=DEFAULT_KMS_TLS) |
| 282 | kms_providers = auto_encrypt_opts.pop("kmsProviders", ALL_KMS_PROVIDERS.copy()) |
| 283 | key_vault_namespace = auto_encrypt_opts.pop("keyVaultNamespace") |
| 284 | extra_opts = auto_encrypt_opts.pop("extraOptions", {}) |
| 285 | for key, value in extra_opts.items(): |
| 286 | auto_encrypt_kwargs[camel_to_snake(key)] = value |
| 287 | for key, value in auto_encrypt_opts.items(): |
| 288 | auto_encrypt_kwargs[camel_to_snake(key)] = value |
| 289 | auto_encryption_opts = AutoEncryptionOpts( |
| 290 | kms_providers, key_vault_namespace, **auto_encrypt_kwargs |
| 291 | ) |
| 292 | kwargs["auto_encryption_opts"] = auto_encryption_opts |
| 293 | |
| 294 | # The unified tests use topologyOpeningEvent, we use topologyOpenedEvent |
| 295 | for i in range(len(observe_events)): |
| 296 | if "topologyOpeningEvent" == observe_events[i]: |
| 297 | observe_events[i] = "topologyOpenedEvent" |
| 298 | ignore_commands = spec.get("ignoreCommandMonitoringEvents", []) |
| 299 | observe_sensitive_commands = spec.get("observeSensitiveCommands", False) |
| 300 | ignore_commands = [cmd.lower() for cmd in ignore_commands] |
| 301 | listener = EventListenerUtil( |
| 302 | observe_events, |
| 303 | ignore_commands, |
| 304 | observe_sensitive_commands, |
| 305 | spec.get("storeEventsAsEntities"), |
| 306 | self, |
| 307 | ) |
| 308 | self._listeners[spec["id"]] = listener |
| 309 | kwargs["event_listeners"] = [listener] |
| 310 | if spec.get("useMultipleMongoses"): |
| 311 | if client_context.load_balancer: |
| 312 | kwargs["h"] = client_context.MULTI_MONGOS_LB_URI |
| 313 | elif client_context.is_mongos: |
| 314 | kwargs["h"] = client_context.mongos_seeds() |
| 315 | kwargs.update(spec.get("uriOptions", {})) |
| 316 | server_api = spec.get("serverApi") |
| 317 | if "waitQueueSize" in kwargs: |
| 318 | raise unittest.SkipTest("PyMongo does not support waitQueueSize") |
| 319 | if "waitQueueMultiple" in kwargs: |
| 320 | raise unittest.SkipTest("PyMongo does not support waitQueueMultiple") |
| 321 | if server_api: |
| 322 | kwargs["server_api"] = ServerApi( |
| 323 | server_api["version"], |
| 324 | strict=server_api.get("strict"), |
| 325 | deprecation_errors=server_api.get("deprecationErrors"), |
| 326 | ) |
no test coverage detected