Issue 5588 - Fix CI tests

Description:

Fix ACL tests that no longer return IndexError but instead return empty list
Fix db_home tests when in container and regular db dir is used instead
Fix repl monitor test where args_instance was not declared before treating it as dict

relates: 389ds#5588

Reviewed by: ?

dirsrvtests/tests/suites/acl/globalgroup_test.py

@@ -203,8 +203,7 @@ def test_deny_group_member_all_rights_to_group_members(topo, aci_of_user, add_te
     conn = UserAccount(topo.standalone, "uid=Ted Morris, ou=Accounting, {}".format(DEFAULT_SUFFIX)).bind(PW_DM)
     # group BIG_GLOBAL no access
     user = UserAccount(conn, 'uid=test_user_1000,ou=ACLGroup,dc=example,dc=com')
-    with pytest.raises(IndexError):
-        user.get_attr_val_utf8('uid')
+    assert len(user.get_attr_val_utf8('uid')) == 0
     UserAccount(topo.standalone, 'uid=test_user_1000,ou=ACLGroup,dc=example,dc=com').delete()
 
 

dirsrvtests/tests/suites/acl/misc_test.py

@@ -474,28 +474,17 @@ def finofaci():
 
     # Search fo existing base DN
     test = Domain(conn, DEFAULT_SUFFIX)
-    try:
-        test.get_attr_vals_utf8_l('dc')
-        assert False
-    except IndexError:
-        pass
+    assert len(test.get_attr_vals_utf8_l('dc')) == 0
 
     # Search for a non existent bases
     subtree = Domain(conn, "ou=does_not_exist," + DEFAULT_SUFFIX)
-    try:
-        subtree.get_attr_vals_utf8_l('objectclass')
-    except IndexError:
-        pass
+    assert len(subtree.get_attr_vals_utf8_l('objectclass')) == 0
+
     subtree = Domain(conn, "ou=also does not exist,ou=does_not_exist," + DEFAULT_SUFFIX)
-    try:
-        subtree.get_attr_vals_utf8_l('objectclass')
-    except IndexError:
-        pass
+    assert len(subtree.get_attr_vals_utf8_l('objectclass')) == 0
+
     # Try ONE level search instead of BASE
-    try:
-        Accounts(conn, "ou=does_not_exist," + DEFAULT_SUFFIX).filter("(objectclass=top)", scope=ldap.SCOPE_ONELEVEL)
-    except IndexError:
-        pass
+    assert len(Accounts(conn, "ou=does_not_exist," + DEFAULT_SUFFIX).filter("(objectclass=top)", scope=ldap.SCOPE_ONELEVEL)) == 0
 
     # add aci
     suffix.add('aci', ACI)

dirsrvtests/tests/suites/acl/modify_test.py

@@ -530,8 +530,7 @@ def test_aci_with_both_allow_and_deny(topo, aci_of_user, cleanup_tree):
     assert UserAccount(conn, USER_WITH_ACI_DELADD).get_attr_val('uid')
     conn = UserAccount(topo.standalone, USER_WITH_ACI_DELADD).bind(PW_DM)
     # aci with both allow and deny, testing deny
-    with pytest.raises(IndexError):
-        UserAccount(conn, USER_WITH_ACI_DELADD).get_attr_val('uid')
+    assert len(UserAccount(conn, USER_WITH_ACI_DELADD).get_attr_val('uid')) == 0
 
 
 def test_allow_owner_to_modify_entry(topo, aci_of_user, cleanup_tree, request):

dirsrvtests/tests/suites/acl/search_real_part2_test.py

@@ -356,12 +356,10 @@ def test_deny_all_access_with_targetfilter_using_boolean_or_of_two_equality_sear
     conn = UserAccount(topo.standalone, USER_ANANDA).bind(PW_DM)
     # aci will deny_all_access_with_targetfilter_using_boolean_or_of_two_equality_search
     user = UserAccount(conn, USER_ANANDA)
-    with pytest.raises(IndexError):
-        user.get_attr_val_utf8('uid')
+    assert len(user.get_attr_val_utf8('uid')) == 0
     # aci will deny_all_access_with_targetfilter_using_boolean_or_of_two_equality_search
     user = UserAccount(conn, USER_ANUJ)
-    with pytest.raises(IndexError):
-        user.get_attr_val_utf8('uid')
+    assert len(user.get_attr_val_utf8('uid')) == 0
     # with root no blockage
     assert UserAccount(topo.standalone, USER_ANANDA).get_attr_val_utf8('uid') == 'Ananda Borah'
     # with root no blockage
@@ -466,8 +464,7 @@ def test_deny_all_access_with_targetfilter_using_presence_search(
     conn = UserAccount(topo.standalone, USER_ANANDA).bind(PW_DM)
     # aci will eny_all_access_with_targetfilter_using_presence_search
     user = UserAccount(conn, 'uid=test_user_1000,ou=People,{}'.format(DEFAULT_SUFFIX))
-    with pytest.raises(IndexError):
-        user.get_attr_val_utf8('cn')
+    assert len(user.get_attr_val_utf8('cn')) == 0
     # with root no blockage
     assert UserAccount(topo.standalone, 'uid=test_user_1000,ou=People,{}'.format(DEFAULT_SUFFIX)).get_attr_val_utf8('cn') == 'test_user_1000'
 

dirsrvtests/tests/suites/acl/valueacl_part2_test.py

@@ -428,10 +428,9 @@ def test_targattrfilters_keyword(topo):
     conn = UserAccount(topo.standalone, "cn=acientryusr1,ou=bug979515,{}".format(DEFAULT_SUFFIX)).bind(PW_DM)
     #  Testing the targattrfilters keyword that allows access control based on the value of the attributes being added (or deleted))
     user = UserAccount(conn, "cn=acientryusr1,ou=bug979515,{}".format(DEFAULT_SUFFIX))
-    with pytest.raises(IndexError):
-        user.get_attr_vals('mail')
-        user.get_attr_vals('telephoneNumber')
-        user.get_attr_vals('cn')
+    assert len(user.get_attr_vals('mail')) == 0
+    assert len(user.get_attr_vals('telephoneNumber')) == 0
+    assert len(user.get_attr_vals('cn')) == 0
     user = UserAccount(topo.standalone, "cn=acientryusr1,ou=bug979515,{}".format(DEFAULT_SUFFIX))
     user.get_attr_vals('mail')
     user.get_attr_vals('telephoneNumber')

dirsrvtests/tests/suites/setup_ds/db_home_test.py

@@ -53,7 +53,11 @@ def test_check_db_home_dir_in_config(topo):
     """
 
     standalone = topo.standalone
-    dbhome_value = '/dev/shm/slapd-{}'.format(standalone.serverid)
+
+    if standalone.is_in_container():
+        dbhome_value = standalone.db_dir
+    else:
+        dbhome_value = '/dev/shm/slapd-{}'.format(standalone.serverid)
     bdb_ldbmconfig = BDB_LDBMConfig(standalone)
 
     log.info('Check the config value of nsslapd-db-home-directory')
@@ -82,7 +86,10 @@ def test_check_db_home_dir_contents(topo):
 
     standalone = topo.standalone
     file_list = ['__db.001', '__db.002', '__db.003', 'DBVERSION']
-    dbhome_value = '/dev/shm/slapd-{}/'.format(standalone.serverid)
+    if standalone.is_in_container():
+        dbhome_value = standalone.db_dir
+    else:
+        dbhome_value = '/dev/shm/slapd-{}/'.format(standalone.serverid)
     old_dbhome = '/var/lib/dirsrv/slapd-{}/db'.format(standalone.serverid)
     existing_files = list(next(os.walk(dbhome_value))[2])
     old_location_files = list(next(os.walk(old_dbhome))[2])
@@ -117,7 +124,10 @@ def test_check_db_home_dir_in_dse(topo):
 
     standalone = topo.standalone
     bdb_ldbmconfig = BDB_LDBMConfig(standalone)
-    dbhome_value = '/dev/shm/slapd-{}'.format(standalone.serverid)
+    if standalone.is_in_container():
+        dbhome_value = standalone.db_dir
+    else:
+        dbhome_value = '/dev/shm/slapd-{}'.format(standalone.serverid)
     dse_ldif = DSEldif(standalone)
 
     log.info('Check value of nsslapd-db-home-directory in dse.ldif')
@@ -142,7 +152,10 @@ def test_check_db_home_dir_in_defaults(topo):
     """
 
     standalone = topo.standalone
-    dbhome_value = 'db_home_dir = /dev/shm/slapd-{instance_name}'
+    if standalone.is_in_container():
+        dbhome_value = 'db_home_dir = ' + standalone.db_dir
+    else:
+        dbhome_value = 'db_home_dir = /dev/shm/slapd-{instance_name}'
 
     log.info('Get defaults.inf path')
     def_loc = standalone.ds_paths._get_defaults_loc(DEFAULTS_PATH)
@@ -174,7 +187,10 @@ def test_delete_db_home_dir(topo):
 
     standalone = topo.standalone
     file_list = ['__db.001', '__db.002', '__db.003', 'DBVERSION']
-    dbhome_value = '/dev/shm/slapd-{}/'.format(standalone.serverid)
+    if standalone.is_in_container():
+        dbhome_value = standalone.db_dir
+    else:
+        dbhome_value = '/dev/shm/slapd-{}/'.format(standalone.serverid)
     existing_files = list(next(os.walk(dbhome_value))[2])
 
     log.info('Stop the instance')

src/lib389/lib389/__init__.py

@@ -3502,3 +3502,10 @@ def is_rootdn_bound(self):
                 return True
 
         return False
+
+    def is_in_container(self):
+        if not self._containerised:
+            container_result = subprocess.run(["systemd-detect-virt", "-c"], stdout=subprocess.PIPE)
+        if self._containerised or container_result.returncode == 0:
+            return True
+        return False

src/lib389/lib389/replica.py

@@ -2705,7 +2705,7 @@ def generate_report(self, get_credentials, use_json=False):
 
             # Open a connection to the consumer
             supplier_inst = DirSrv(verbose=self._instance.verbose)
-            args_instance[SER_HOST] = supplier_hostname
+            args_instance = {SER_HOST: supplier_hostname}
             if supplier_protocol == "ssl" or supplier_protocol == "ldaps":
                 args_instance[SER_SECURE_PORT] = int(supplier_port)
             else: