force delete Kerberos test principals

ceod/model/KerberosService.py

@@ -1,6 +1,7 @@
 import os
 import shutil
 import subprocess
+from typing import List
 
 from zope import component
 from zope.interface import implementer
@@ -50,31 +51,34 @@ class KerberosService:
                 if princ is not None:
                     lib.krb5_free_principal(k_ctx, princ)
 
+    def _run(self, args: List[str]):
+        subprocess.run(args, check=True)
+
     def addprinc(self, principal: str, password: str):
-        subprocess.run([
+        self._run([
             'kadmin', '-k', '-p', self.admin_principal, 'addprinc',
             '-pw', password,
             '-policy', 'default',
             '+needchange',
             '+requires_preauth',
             principal
-        ], check=True)
+        ])
 
     def delprinc(self, principal: str):
-        subprocess.run([
+        self._run([
             'kadmin', '-k', '-p', self.admin_principal, 'delprinc',
             '-force',
             principal
-        ], check=True)
+        ])
 
     def change_password(self, principal: str, password: str):
-        subprocess.run([
+        self._run([
             'kadmin', '-k', '-p', self.admin_principal, 'cpw',
             '-pw', password,
             principal
-        ], check=True)
-        subprocess.run([
+        ])
+        self._run([
             'kadmin', '-k', '-p', self.admin_principal, 'modprinc',
             '+needchange',
             principal
-        ], check=True)
+        ])

tests/conftest.py

@@ -47,6 +47,17 @@ def cfg(_drone_hostname_mock):
     return _cfg
 
 
+def delete_test_princs(krb_srv):
+    proc = subprocess.run([
+        'kadmin', '-k', '-p', krb_srv.admin_principal, 'listprincs', 'test_*',
+    ], text=True, capture_output=True, check=True)
+    princs = [line.strip() for line in proc.stdout.splitlines()]
+    # remove the password prompt
+    princs = princs[1:]
+    for princ in princs:
+        krb_srv.delprinc(princ)
+
+
 @pytest.fixture(scope='session')
 def krb_srv(cfg):
     # TODO: create temporary Kerberos database using kdb5_util.
@@ -60,7 +71,10 @@ def krb_srv(cfg):
     cache_dir = cfg.get('ceod_krb5_cache_dir')
     krb = KerberosService(principal)
     component.provideUtility(krb, IKerberosService)
+
+    delete_test_princs(krb)
     yield krb
+    delete_test_princs(krb)
     shutil.rmtree(cache_dir)