blob: f2624496d544ebb0c1b1b8dfc9eff7a59a5c0cb4 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
"""Test hook for verifying that the primary has not stepped down or changed."""
from __future__ import absolute_import
import pymongo.errors
from . import interface
from ..fixtures import replicaset
from ... import errors
class CheckPrimary(interface.Hook):
"""Hook that checks that the primary is still primary after the test."""
def __init__(self, hook_logger, rs_fixture):
"""Initialize CheckPrimary."""
description = "Verify that the primary has not stepped down or changed"
interface.Hook.__init__(self, hook_logger, rs_fixture, description)
if not isinstance(rs_fixture, replicaset.ReplicaSetFixture):
raise TypeError("{} is not a replica set".format(rs_fixture.__class__.__name__))
self._rs_fixture = rs_fixture
self._primary_url = None
def _get_primary_url(self):
no_primary_err = errors.ServerFailure("No primary found")
for node in self._rs_fixture.nodes:
try:
is_master = node.mongo_client().admin.command("isMaster")["ismaster"]
except pymongo.errors.AutoReconnect:
raise no_primary_err
if is_master:
return node.get_driver_connection_url()
raise no_primary_err
def before_test(self, test, test_report):
"""Before test hook primary."""
self._primary_url = self._get_primary_url()
def after_test(self, test, test_report):
"""After test hook primary."""
new_primary_url = self._get_primary_url()
if new_primary_url != self._primary_url:
raise errors.ServerFailure("Primary changed, was {} and is now {}".format(
self._primary_url, new_primary_url))
|
