|
26 | 26 | import time |
27 | 27 | import unittest |
28 | 28 | from enum import Enum |
| 29 | +from pathlib import Path |
29 | 30 | from tempfile import NamedTemporaryFile |
30 | 31 |
|
31 | 32 | # isort: THIRDPARTY |
@@ -428,7 +429,7 @@ class FilesystemSymlinkMonitor(unittest.TestCase): |
428 | 429 |
|
429 | 430 | maxDiff = None |
430 | 431 |
|
431 | | - def run_check(self, stop_time): |
| 432 | + def run_check(self, stop_time): # pylint: disable=too-many-locals |
432 | 433 | """ |
433 | 434 | Check that the filesystem links on the D-Bus and the filesystem links |
434 | 435 | expected from looking at filesystem devicemapper paths match exactly. |
@@ -478,6 +479,61 @@ def run_check(self, stop_time): |
478 | 479 | f"{decode_dm} invocation failed: " |
479 | 480 | f"{stderrdata.decode('utf-8')}" |
480 | 481 | ) |
| 482 | + try: |
| 483 | + (_, dev, stratis, pool_name, filesystem_name) = Path( |
| 484 | + symlink |
| 485 | + ).parts |
| 486 | + except ValueError as err: |
| 487 | + raise RuntimeError( |
| 488 | + f"Symlink {symlink} did not decompose into expected parts" |
| 489 | + ) from err |
| 490 | + |
| 491 | + self.assertEqual(stratis, "stratis") |
| 492 | + self.assertEqual(dev, "dev") |
| 493 | + |
| 494 | + command = [ |
| 495 | + decode_dm, |
| 496 | + os.path.join(DEV_MAPPER, dev), |
| 497 | + "--output=filesystem-name", |
| 498 | + ] |
| 499 | + |
| 500 | + with subprocess.Popen( |
| 501 | + command, |
| 502 | + stdout=subprocess.PIPE, |
| 503 | + stderr=subprocess.PIPE, |
| 504 | + ) as proc: |
| 505 | + (stdoutdata, stderrdata) = proc.communicate() |
| 506 | + if proc.returncode == 0: |
| 507 | + self.assertEqual( |
| 508 | + stdoutdata.decode("utf-8").strip(), filesystem_name |
| 509 | + ) |
| 510 | + else: |
| 511 | + raise RuntimeError( |
| 512 | + f"{decode_dm} invocation failed: " |
| 513 | + f"{stderrdata.decode('utf-8')}" |
| 514 | + ) |
| 515 | + |
| 516 | + command = [ |
| 517 | + decode_dm, |
| 518 | + os.path.join(DEV_MAPPER, dev), |
| 519 | + "--output=pool-name", |
| 520 | + ] |
| 521 | + |
| 522 | + with subprocess.Popen( |
| 523 | + command, |
| 524 | + stdout=subprocess.PIPE, |
| 525 | + stderr=subprocess.PIPE, |
| 526 | + ) as proc: |
| 527 | + (stdoutdata, stderrdata) = proc.communicate() |
| 528 | + if proc.returncode == 0: |
| 529 | + self.assertEqual( |
| 530 | + stdoutdata.decode("utf-8").strip(), pool_name |
| 531 | + ) |
| 532 | + else: |
| 533 | + raise RuntimeError( |
| 534 | + f"{decode_dm} invocation failed: " |
| 535 | + f"{stderrdata.decode('utf-8')}" |
| 536 | + ) |
481 | 537 | except FileNotFoundError as err: |
482 | 538 | raise RuntimeError( |
483 | 539 | f"Script '{decode_dm}' missing, test could not be run" |
|
0 commit comments