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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
|
import time
import ldap
import logging
import pytest
from lib389 import DirSrv, Entry, tools, tasks
from lib389.tools import DirSrvTools
from lib389._constants import *
from lib389.properties import *
from lib389.tasks import *
from lib389.utils import *
from lib389.topologies import topology_m3 as T
import socket
# Skip on older versions
pytestmark = [pytest.mark.tier2,
pytest.mark.skipif(ds_is_older('1.3.6'), reason="Not implemented")]
DEBUGGING = os.getenv("DEBUGGING", default=False)
if DEBUGGING:
logging.getLogger(__name__).setLevel(logging.DEBUG)
else:
logging.getLogger(__name__).setLevel(logging.INFO)
log = logging.getLogger(__name__)
def test_ticket49020(T):
A = T.ms['supplier1']
B = T.ms['supplier2']
C = T.ms['supplier3']
A.enableReplLogging()
B.enableReplLogging()
C.enableReplLogging()
AtoB = A.agreement.list(suffix=DEFAULT_SUFFIX)[0].dn
AtoC = A.agreement.list(suffix=DEFAULT_SUFFIX)[1].dn
CtoB = C.agreement.list(suffix=DEFAULT_SUFFIX)[1].dn
A.agreement.pause(AtoB)
C.agreement.pause(CtoB)
time.sleep(5)
name = "userX"
dn = "cn={},{}".format(name, DEFAULT_SUFFIX)
A.add_s(Entry((dn, {'objectclass': "top person".split(),
'sn': name,'cn': name})))
A.agreement.init(DEFAULT_SUFFIX, socket.gethostname(), PORT_SUPPLIER_3)
time.sleep(5)
for i in range(1,11):
name = "userY{}".format(i)
dn = "cn={},{}".format(name, DEFAULT_SUFFIX)
A.add_s(Entry((dn, {'objectclass': "top person".split(),
'sn': name,'cn': name})))
time.sleep(5)
C.agreement.resume(CtoB)
time.sleep(5)
A_entries = A.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
'(objectClass=person)')
B_entries = B.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
'(objectClass=person)')
C_entries = C.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
'(objectClass=person)')
assert len(A_entries) == len(C_entries)
assert len(B_entries) == len(A_entries) - 11
if __name__ == '__main__':
# Run isolated
# -s for DEBUG mode
CURRENT_FILE = os.path.realpath(__file__)
pytest.main("-s %s" % CURRENT_FILE)
|
