(self)
| 393 | self.assertTrue(t.description.has_readable_server(Secondary(tag_sets=[{"tag": "exists"}]))) |
| 394 | |
| 395 | def test_close(self): |
| 396 | t = create_mock_topology(replica_set_name="rs") |
| 397 | got_hello( |
| 398 | t, |
| 399 | ("a", 27017), |
| 400 | {"ok": 1, HelloCompat.LEGACY_CMD: True, "setName": "rs", "hosts": ["a", "b"]}, |
| 401 | ) |
| 402 | |
| 403 | got_hello( |
| 404 | t, |
| 405 | ("b", 27017), |
| 406 | { |
| 407 | "ok": 1, |
| 408 | HelloCompat.LEGACY_CMD: False, |
| 409 | "secondary": True, |
| 410 | "setName": "rs", |
| 411 | "hosts": ["a", "b"], |
| 412 | }, |
| 413 | ) |
| 414 | |
| 415 | self.assertEqual(SERVER_TYPE.RSPrimary, get_type(t, "a")) |
| 416 | self.assertEqual(SERVER_TYPE.RSSecondary, get_type(t, "b")) |
| 417 | self.assertTrue(get_monitor(t, "a").opened) |
| 418 | self.assertTrue(get_monitor(t, "b").opened) |
| 419 | self.assertEqual(TOPOLOGY_TYPE.ReplicaSetWithPrimary, t.description.topology_type) |
| 420 | |
| 421 | t.close() |
| 422 | self.assertEqual(2, len(t.description.server_descriptions())) |
| 423 | self.assertEqual(SERVER_TYPE.Unknown, get_type(t, "a")) |
| 424 | self.assertEqual(SERVER_TYPE.Unknown, get_type(t, "b")) |
| 425 | self.assertFalse(get_monitor(t, "a").opened) |
| 426 | self.assertFalse(get_monitor(t, "b").opened) |
| 427 | self.assertEqual("rs", t.description.replica_set_name) |
| 428 | self.assertEqual(TOPOLOGY_TYPE.ReplicaSetNoPrimary, t.description.topology_type) |
| 429 | |
| 430 | # A closed topology should not be updated when receiving a hello. |
| 431 | got_hello( |
| 432 | t, |
| 433 | ("a", 27017), |
| 434 | {"ok": 1, HelloCompat.LEGACY_CMD: True, "setName": "rs", "hosts": ["a", "b", "c"]}, |
| 435 | ) |
| 436 | |
| 437 | self.assertEqual(2, len(t.description.server_descriptions())) |
| 438 | self.assertEqual(SERVER_TYPE.Unknown, get_type(t, "a")) |
| 439 | self.assertEqual(SERVER_TYPE.Unknown, get_type(t, "b")) |
| 440 | self.assertFalse(get_monitor(t, "a").opened) |
| 441 | self.assertFalse(get_monitor(t, "b").opened) |
| 442 | # Server c should not have been added. |
| 443 | self.assertEqual(None, get_server(t, "c")) |
| 444 | self.assertEqual("rs", t.description.replica_set_name) |
| 445 | self.assertEqual(TOPOLOGY_TYPE.ReplicaSetNoPrimary, t.description.topology_type) |
| 446 | |
| 447 | def test_handle_error(self): |
| 448 | t = create_mock_topology(replica_set_name="rs") |
nothing calls this directly
no test coverage detected