2626import time
2727import unittest
2828from enum import Enum
29+ from pathlib import Path
2930from tempfile import NamedTemporaryFile
3031
3132# isort: THIRDPARTY
@@ -428,7 +429,7 @@ class FilesystemSymlinkMonitor(unittest.TestCase):
428429
429430 maxDiff = None
430431
431- def run_check (self , stop_time ):
432+ def run_check (self , stop_time ): # pylint: disable=too-many-locals
432433 """
433434 Check that the filesystem links on the D-Bus and the filesystem links
434435 expected from looking at filesystem devicemapper paths match exactly.
@@ -455,11 +456,14 @@ def run_check(self, stop_time):
455456
456457 try :
457458 found = 0
458- for dev in fnmatch .filter (os .listdir (DEV_MAPPER ), "stratis-1-*-thin-fs-*" ):
459+ for device in fnmatch .filter (
460+ os .listdir (DEV_MAPPER ), "stratis-1-*-thin-fs-*"
461+ ):
462+ device_path = os .path .join (DEV_MAPPER , device )
459463 found += 1
460464 command = [
461465 decode_dm ,
462- os . path . join ( DEV_MAPPER , dev ) ,
466+ device_path ,
463467 "--output=symlink" ,
464468 ]
465469 try :
@@ -478,6 +482,61 @@ def run_check(self, stop_time):
478482 f"{ decode_dm } invocation failed: "
479483 f"{ stderrdata .decode ('utf-8' )} "
480484 )
485+ try :
486+ (_ , dev , stratis , pool_name , filesystem_name ) = Path (
487+ symlink
488+ ).parts
489+ except ValueError as err :
490+ raise RuntimeError (
491+ f"Symlink { symlink } did not decompose into expected parts"
492+ ) from err
493+
494+ self .assertEqual (stratis , "stratis" )
495+ self .assertEqual (dev , "dev" )
496+
497+ command = [
498+ decode_dm ,
499+ device_path ,
500+ "--output=filesystem-name" ,
501+ ]
502+
503+ with subprocess .Popen (
504+ command ,
505+ stdout = subprocess .PIPE ,
506+ stderr = subprocess .PIPE ,
507+ ) as proc :
508+ (stdoutdata , stderrdata ) = proc .communicate ()
509+ if proc .returncode == 0 :
510+ self .assertEqual (
511+ stdoutdata .decode ("utf-8" ).strip (), filesystem_name
512+ )
513+ else :
514+ raise RuntimeError (
515+ f"{ decode_dm } invocation failed: "
516+ f"{ stderrdata .decode ('utf-8' )} "
517+ )
518+
519+ command = [
520+ decode_dm ,
521+ device_path ,
522+ "--output=pool-name" ,
523+ ]
524+
525+ with subprocess .Popen (
526+ command ,
527+ stdout = subprocess .PIPE ,
528+ stderr = subprocess .PIPE ,
529+ ) as proc :
530+ (stdoutdata , stderrdata ) = proc .communicate ()
531+ if proc .returncode == 0 :
532+ self .assertEqual (
533+ stdoutdata .decode ("utf-8" ).strip (), pool_name
534+ )
535+ else :
536+ raise RuntimeError (
537+ f"{ decode_dm } invocation failed: "
538+ f"{ stderrdata .decode ('utf-8' )} "
539+ )
481540 except FileNotFoundError as err :
482541 raise RuntimeError (
483542 f"Script '{ decode_dm } ' missing, test could not be run"
0 commit comments