Unlock 30% off on Manual Testing Annual Plans this Holiday Season. 
Use Code: LCMYEARENDEXCL30
Use Code: LCMYEARENDEXCL30Check icon
Copied to Clipboard!
How to use fsck method in autotest
Best Python code snippet using autotest_python
test_fsck.py
Source:test_fsck.py 
1# coding=utf-82#3# Run a variety of tests against fsck_msdos4#5# Usage:6# python test_fsck.py [<fsck_msdos> [<tmp_dir>]]7#8# where <tmp_dir> is a path to a directory where disk images will be9# temporarily created. If <path_to_fsck> is specified, it is used instead10# of 'fsck_msdos' to invoke the fsck_msdos program (for example, to test11# a new build that has not been installed).12#13from __future__ import with_statement14import sys15import os16import subprocess17import struct18from msdosfs import *19from HexDump import HexDump20class LaunchError(Exception):21 def __init__(self, returncode):22 self.returncode = returncode23 if returncode < 0:24 self.message = "Program exited with signal %d" % -returncode25 else:26 self.message = "Program exited with status %d" % returncode27 28 def __str__(self):29 return self.message30class FailureExpected(Exception):31 def __init__(self, s):32 self.s = s33 def __str__(self):34 return self.s35class RepairFailed(Exception):36 def __init__(self, s):37 self.s = s38 def __str__(self):39 return "RepairFailed({0})".format(self.s)40#41# launch -- A helper to run another process and collect the standard output42# and standard error streams. If the process returns a non-zero exit43# status, then raise an exception.44#45def launch(args, **kwargs):46 print "launch:", args, kwargs47 p = subprocess.Popen(args, **kwargs)48 stdout, stderr = p.communicate()49 if p.returncode != 0:50 raise LaunchError(p.returncode)51 return stdout, stderr52#53# 1. Make a disk image file54# 2. Attach the image file, without mounting55# ---- Begin per-test stuff ----56# 3. newfs_msdos the image57# 4. Fill image with content58# 5. fsck_msdos -n the image59# 6. fsck_msdos -y the image60# 7. Run /sbin/fsck_msdos against image61# ---- End per-test stuff ----62# 8. Detach the image63# 9. Delete the image file64#65#66# Run tests on 20GiB FAT32 sparse disk image67#68def test_fat32(dir, fsck, newfs): 69 #70 # Create a 20GB disk image in @dir71 #72 dmg = os.path.join(dir, 'Test20GB.sparseimage')73 launch('hdiutil create -size 20g -type SPARSE -layout NONE'.split()+[dmg])74 newfs_opts = "-F 32 -b 4096 -v TEST20GB".split()75 76 #77 # Attach the image78 #79 disk = launch(['hdiutil', 'attach', '-nomount', dmg], stdout=subprocess.PIPE)[0].rstrip()80 rdisk = disk.replace('/dev/disk', '/dev/rdisk')81 82 #83 # Run tests84 #85 # TODO: Known good disk86 # empty file87 # one cluster file88 # larger file89 # one cluster directory90 # larger directory91 #92 test_quick(rdisk, fsck, newfs, newfs_opts)93 test_bad_args(rdisk, fsck, newfs, newfs_opts)94 test_maxmem(rdisk, fsck, newfs, newfs_opts)95 test_empty(rdisk, fsck, newfs, newfs_opts)96 test_boot_sector(rdisk, fsck, newfs, newfs_opts)97 test_boot_fat32(rdisk, fsck, newfs, newfs_opts) # FAT32 only!98 test_fsinfo(rdisk, fsck, newfs, newfs_opts) # FAT32 only!99 fat_too_small(rdisk, fsck, newfs, newfs_opts)100 orphan_clusters(rdisk, fsck, newfs, newfs_opts)101 file_excess_clusters(rdisk, fsck, newfs, newfs_opts)102 file_bad_clusters(rdisk, fsck, newfs, newfs_opts)103 dir_bad_start(rdisk, fsck, newfs, newfs_opts)104 root_bad_start(rdisk, fsck, newfs, newfs_opts) # FAT32 only!105 root_bad_first_cluster(rdisk, fsck, newfs, newfs_opts) # FAT32 only!106 dir_size_dots(rdisk, fsck, newfs, newfs_opts)107 long_name(rdisk, fsck, newfs, newfs_opts)108 past_end_of_dir(rdisk, fsck, newfs, newfs_opts)109 past_end_of_dir(rdisk, fsck, newfs, newfs_opts, True)110 fat_bad_0_or_1(rdisk, fsck, newfs, newfs_opts)111 fat_mark_clean_corrupt(rdisk, fsck, newfs, newfs_opts)112 fat_mark_clean_ok(rdisk, fsck, newfs, newfs_opts)113 file_4GB(rdisk, fsck, newfs, newfs_opts)114 file_4GB_excess_clusters(rdisk, fsck, newfs, newfs_opts)115 directory_garbage(rdisk, fsck, newfs, newfs_opts)116 117 #118 # Detach the image119 #120 launch(['diskutil', 'eject', disk])121 122 #123 # Delete the image file124 #125 os.remove(dmg)126#127# Run tests on 160MiB FAT16 image128#129def test_fat16(dir, fsck, newfs): 130 #131 # Create a 160MB disk image in @dir132 #133 dmg = os.path.join(dir, 'Test160MB.dmg')134 f = file(dmg, "w")135 f.truncate(160*1024*1024)136 f.close()137 newfs_opts = "-F 16 -b 4096 -v TEST160MB".split()138 139 #140 # Attach the image141 #142 disk = launch(['hdiutil', 'attach', '-nomount', dmg], stdout=subprocess.PIPE)[0].rstrip()143 rdisk = disk.replace('/dev/disk', '/dev/rdisk')144 145 #146 # Run tests147 #148 # TODO: Known good disk149 # empty file150 # one cluster file151 # larger file152 # one cluster directory153 # larger directory154 #155 test_quick(rdisk, fsck, newfs, newfs_opts)156 test_bad_args(rdisk, fsck, newfs, newfs_opts)157 test_maxmem(rdisk, fsck, newfs, newfs_opts)158 test_empty(rdisk, fsck, newfs, newfs_opts)159 test_boot_sector(rdisk, fsck, newfs, newfs_opts)160 fat_too_small(rdisk, fsck, newfs, newfs_opts)161 orphan_clusters(rdisk, fsck, newfs, newfs_opts)162 file_excess_clusters(rdisk, fsck, newfs, newfs_opts)163 file_bad_clusters(rdisk, fsck, newfs, newfs_opts)164 dir_bad_start(rdisk, fsck, newfs, newfs_opts)165 dir_size_dots(rdisk, fsck, newfs, newfs_opts)166 long_name(rdisk, fsck, newfs, newfs_opts)167 past_end_of_dir(rdisk, fsck, newfs, newfs_opts)168 past_end_of_dir(rdisk, fsck, newfs, newfs_opts, True)169 fat_bad_0_or_1(rdisk, fsck, newfs, newfs_opts)170 fat_mark_clean_corrupt(rdisk, fsck, newfs, newfs_opts)171 fat_mark_clean_ok(rdisk, fsck, newfs, newfs_opts)172 directory_garbage(rdisk, fsck, newfs, newfs_opts)173 174 #175 # Detach the image176 #177 launch(['diskutil', 'eject', disk])178 179 #180 # Delete the image file181 #182 os.remove(dmg)183#184# Run tests on 15MiB FAT12 image185#186def test_fat12(dir, fsck, newfs): 187 #188 # Create a 15MB disk image in @dir189 #190 dmg = os.path.join(dir, 'Test15MB.dmg')191 f = file(dmg, "w")192 f.truncate(15*1024*1024)193 f.close()194 newfs_opts = "-F 12 -b 4096 -v TEST15MB".split()195 196 #197 # Attach the image198 #199 disk = launch(['hdiutil', 'attach', '-nomount', dmg], stdout=subprocess.PIPE)[0].rstrip()200 rdisk = disk.replace('/dev/disk', '/dev/rdisk')201 202 #203 # Run tests204 #205 # TODO: Known good disk206 # empty file207 # one cluster file208 # larger file209 # one cluster directory210 # larger directory211 #212 test_quick(rdisk, fsck, newfs, newfs_opts)213 test_bad_args(rdisk, fsck, newfs, newfs_opts)214 test_maxmem(rdisk, fsck, newfs, newfs_opts)215 test_empty(rdisk, fsck, newfs, newfs_opts)216 test_boot_sector(rdisk, fsck, newfs, newfs_opts)217 fat_too_small(rdisk, fsck, newfs, newfs_opts)218 orphan_clusters(rdisk, fsck, newfs, newfs_opts)219 file_excess_clusters(rdisk, fsck, newfs, newfs_opts)220 file_bad_clusters(rdisk, fsck, newfs, newfs_opts)221 dir_bad_start(rdisk, fsck, newfs, newfs_opts)222 dir_size_dots(rdisk, fsck, newfs, newfs_opts)223 long_name(rdisk, fsck, newfs, newfs_opts)224 past_end_of_dir(rdisk, fsck, newfs, newfs_opts)225 past_end_of_dir(rdisk, fsck, newfs, newfs_opts, True)226 fat_bad_0_or_1(rdisk, fsck, newfs, newfs_opts)227 directory_garbage(rdisk, fsck, newfs, newfs_opts)228 229 #230 # Detach the image231 #232 launch(['diskutil', 'eject', disk])233 234 #235 # Delete the image file236 #237 os.remove(dmg)238#239# Run tests on 100MiB FAT12 image240#241def test_fat12_100MB(dir, fsck, newfs): 242 #243 # Create a 100MB disk image in @dir244 #245 dmg = os.path.join(dir, 'Test100MB.dmg')246 f = file(dmg, "w")247 f.truncate(100*1024*1024)248 f.close()249 newfs_opts = "-F 12 -b 32768 -v TEST100MB".split()250 251 #252 # Attach the image253 #254 disk = launch(['hdiutil', 'attach', '-nomount', dmg], stdout=subprocess.PIPE)[0].rstrip()255 rdisk = disk.replace('/dev/disk', '/dev/rdisk')256 257 #258 # Run tests259 #260 test_18523205(rdisk, fsck, newfs, newfs_opts)261 262 #263 # Detach the image264 #265 launch(['diskutil', 'eject', disk])266 267 #268 # Delete the image file269 #270 os.remove(dmg)271#272# A minimal test -- make sure fsck_msdos runs on an empty image273#274def test_empty(disk, fsck, newfs, newfs_opts):275 #276 # newfs the disk277 #278 launch([newfs]+newfs_opts+[disk])279 280 #281 # fsck the disk282 #283 launch([fsck, '-n', disk])284#285# Make a volume with allocated but unreferenced cluster chains286#287def orphan_clusters(disk, fsck, newfs, newfs_opts):288 launch([newfs]+newfs_opts+[disk])289 290 #291 # Create some cluster chains not referenced by any file or directory292 #293 f = file(disk, "r+")294 v = msdosfs(f)295 v.allocate(7, 100)296 v.allocate(23, 150)297 v.allocate(1, 190)298 v.flush()299 del v300 f.close()301 del f302 303 try:304 launch([fsck, '-n', disk])305 except LaunchError:306 pass307 launch([fsck, '-p', disk])308 launch(['/sbin/fsck_msdos', '-n', disk])309#310# Make a file with excess clusters allocated311# One file with EOF == 0312# One file with EOF != 0313# Files with excess clusters that are cross-linked314# First excess cluster is cross-linked315# Other excess cluster is cross-linked316# Excess clusters end with free/bad/reserved cluster317# First excess cluster is free/bad/reserved318# Other excess cluster is free/bad/reserved319#320def file_excess_clusters(disk, fsck, newfs, newfs_opts):321 launch([newfs]+newfs_opts+[disk])322 #323 # Create files with too many clusters for their size324 #325 f = file(disk, "r+")326 v = msdosfs(f)327 head=v.allocate(7)328 v.root().mkfile('FOO', head=head, length=6*v.bytesPerCluster)329 head=v.allocate(1)330 v.root().mkfile('BAR', head=head, length=0)331 332 #333 # LINK1 is OK.334 # LINK2 contains excess clusters; the first is cross-linked with LINK1335 # LINK3 contains excess clusters; the second is cross-linked with LINK1336 #337 clusters = v.fat.find(9)338 head = v.fat.chain(clusters)339 v.root().mkfile('LINK1', head=head, length=8*v.bytesPerCluster+1)340 head = v.fat.allocate(3, last=clusters[7])341 v.root().mkfile('LINK2', head=head, length=2*v.bytesPerCluster+3)342 head = v.fat.allocate(5, last=clusters[8])343 v.root().mkfile('LINK3', head=head, length=3*v.bytesPerCluster+5)344 if v.fsinfo:345 v.fsinfo.allocate(9+3+5)346 #347 # FREE1 has its first excess cluster marked free348 # BAD3 has its third excess cluster marked bad349 #350 head = v.allocate(11, last=CLUST_BAD)351 v.root().mkfile('BAD3', head=head, length=8*v.bytesPerCluster+300)352 head = v.allocate(8, last=CLUST_FREE)353 v.root().mkfile('FREE1', head=head, length=6*v.bytesPerCluster+100)354 355 v.flush()356 del v357 f.close()358 del f359 360 try:361 launch([fsck, '-n', disk])362 except LaunchError:363 pass364 launch([fsck, '-y', disk])365 launch(['/sbin/fsck_msdos', '-n', disk])366#367# Make files with bad clusters in their chains368# FILE1 file with middle cluster free369# FILE2 file with middle cluster bad/reserved370# FILE3 file with middle cluster points to out of range cluster371# FILE4 file with middle cluster that is cross-linked (to same file)372# FILE5 file whose head is "free"373# FILE6 file whose head is "bad"374# FILE7 file whose head is out of range375# FILE8 file whose head is cross-linked376#377def file_bad_clusters(disk, fsck, newfs, newfs_opts):378 launch([newfs]+newfs_opts+[disk])379 f = file(disk, "r+")380 v = msdosfs(f)381 382 clusters = v.fat.find(5)383 to_free = clusters[2]384 head = v.fat.chain(clusters)385 v.root().mkfile('FILE1', head=head, length=6*v.bytesPerCluster+111)386 if v.fsinfo:387 v.fsinfo.allocate(5)388 389 clusters = v.fat.find(5)390 head = v.fat.chain(clusters)391 v.root().mkfile('FILE2', head=head, length=4*v.bytesPerCluster+222)392 v.fat[clusters[2]] = CLUST_RSRVD393 if v.fsinfo:394 v.fsinfo.allocate(5)395 396 clusters = v.fat.find(5)397 head = v.fat.chain(clusters)398 v.root().mkfile('FILE3', head=head, length=4*v.bytesPerCluster+333)399 v.fat[clusters[2]] = 1400 if v.fsinfo:401 v.fsinfo.allocate(5)402 403 clusters = v.fat.find(5)404 head = v.fat.chain(clusters)405 v.root().mkfile('FILE4', head=head, length=4*v.bytesPerCluster+44)406 v.fat[clusters[2]] = clusters[1]407 if v.fsinfo:408 v.fsinfo.allocate(5)409 410 v.root().mkfile('FILE5', head=CLUST_FREE, length=4*v.bytesPerCluster+55)411 v.root().mkfile('FILE6', head=CLUST_BAD, length=4*v.bytesPerCluster+66)412 v.root().mkfile('FILE7', head=CLUST_RSRVD-1, length=4*v.bytesPerCluster+77)413 head = v.allocate(5)414 v.root().mkfile('FOO', head=head, length=4*v.bytesPerCluster+99)415 v.root().mkfile('FILE8', head=head, length=4*v.bytesPerCluster+88)416 # Free the middle cluster of FILE1 now that we've finished allocating417 v.fat[to_free] = CLUST_FREE418 419 v.flush()420 del v421 f.close()422 del f423 424 try:425 launch([fsck, '-n', disk])426 except LaunchError:427 pass428 launch([fsck, '-y', disk])429 launch(['/sbin/fsck_msdos', '-n', disk])430#431# Make directories whose starting cluster number is free/bad/reserved/out of range432# DIR1 start cluster is free433# DIR2 start cluster is reserved434# DIR3 start cluster is bad435# DIR4 start cluster is EOF436# DIR5 start cluster is 1437# DIR6 start cluster is one more than max valid cluster438#439def dir_bad_start(disk, fsck, newfs, newfs_opts):440 def mkdir(parent, name, head):441 bytes = make_long_dirent(name, ATTR_DIRECTORY, head=head)442 slots = len(bytes)/32443 slot = parent.find_slots(slots, grow=True)444 parent.write_slots(slot, bytes)445 launch([newfs]+newfs_opts+[disk])446 f = file(disk, "r+")447 v = msdosfs(f)448 root = v.root()449 450 mkdir(root, 'DIR1', CLUST_FREE)451 mkdir(root, 'DIR2', CLUST_RSRVD)452 mkdir(root, 'DIR3', CLUST_BAD)453 mkdir(root, 'DIR4', CLUST_EOF)454 mkdir(root, 'DIR5', 1)455 mkdir(root, 'DIR6', v.clusters+2)456 457 v.flush()458 del v459 f.close()460 del f461 462 try:463 launch([fsck, '-n', disk])464 except LaunchError:465 pass466 launch([fsck, '-y', disk])467 launch(['/sbin/fsck_msdos', '-n', disk])468#469# Root dir's starting cluster number is free/bad/reserved/out of range470#471# NOTE: This test is only applicable to FAT32!472#473def root_bad_start(disk, fsck, newfs, newfs_opts):474 def set_root_start(disk, head):475 dev = file(disk, "r+")476 dev.seek(0)477 bytes = dev.read(512)478 bytes = bytes[0:44] + struct.pack("<I", head) + bytes[48:]479 dev.seek(0)480 dev.write(bytes)481 dev.close()482 del dev483 launch([newfs]+newfs_opts+[disk])484 f = file(disk, "r+")485 v = msdosfs(f)486 clusters = v.clusters487 v.flush()488 del v489 f.close()490 del f491 492 for head in [CLUST_FREE, CLUST_RSRVD, CLUST_BAD, CLUST_EOF, 1, clusters+2]:493 set_root_start(disk, head)494 495 try:496 launch([fsck, '-n', disk])497 except LaunchError:498 pass499 try:500 launch([fsck, '-y', disk])501 except LaunchError:502 pass503 try:504 launch(['/sbin/fsck_msdos', '-n', disk])505 except LaunchError:506 pass507#508# Root dir's first cluster is free/bad/reserved509#510# NOTE: This test is only applicable to FAT32!511#512def root_bad_first_cluster(disk, fsck, newfs, newfs_opts):513 for link in [CLUST_FREE, CLUST_RSRVD, CLUST_BAD]:514 launch([newfs]+newfs_opts+[disk])515 516 f = file(disk, "r+")517 v = msdosfs(f)518 v.fat[v.rootCluster] = link519 v.flush()520 del v521 f.close()522 del f523 524 try:525 launch([fsck, '-n', disk])526 except LaunchError:527 pass528 launch([fsck, '-y', disk])529 launch(['/sbin/fsck_msdos', '-n', disk])530#531# Create subdirectories with the following problems:532# Size (length) field is non-zero533# "." entry has wrong starting cluster534# ".." entry start cluster is non-zero, and parent is root535# ".." entry start cluster is zero, and parent is not root536# ".." entry start cluster is incorrect537#538def dir_size_dots(disk, fsck, newfs, newfs_opts):539 launch([newfs]+newfs_opts+[disk])540 f = file(disk, "r+")541 v = msdosfs(f)542 root = v.root()543 544 # Make a couple of directories without any problems545 child = root.mkdir('CHILD')546 grand = child.mkdir('GRAND')547 # Directory has non-zero size548 dir = root.mkdir('BADSIZE', length=666)549 # "." entry has incorrect start cluster550 dir = root.mkdir('BADDOT')551 fields = parse_dirent(dir.read_slots(0))552 fields['head'] = fields['head'] + 30553 dir.write_slots(0, make_dirent(**fields))554 # ".." entry has non-zero start cluster, but parent is root555 dir = root.mkdir('DOTDOT.NZ')556 fields = parse_dirent(dir.read_slots(0))557 fields['head'] = 47558 dir.write_slots(0, make_dirent(**fields))559 # ".." entry has zero start cluster, but parent is not root560 dir = child.mkdir('DOTDOT.ZER')561 fields = parse_dirent(dir.read_slots(0))562 fields['head'] = 0563 dir.write_slots(0, make_dirent(**fields))564 # ".." entry start cluster is incorrect (parent is not root)565 dir = grand.mkdir('DOTDOT.BAD')566 fields = parse_dirent(dir.read_slots(0))567 fields['head'] = fields['head'] + 30568 dir.write_slots(0, make_dirent(**fields))569 570 v.flush()571 del v572 f.close()573 del f574 try:575 launch([fsck, '-n', disk])576 except LaunchError:577 pass578 launch([fsck, '-y', disk])579 launch(['/sbin/fsck_msdos', '-n', disk])580def long_name(disk, fsck, newfs, newfs_opts):581 launch([newfs]+newfs_opts+[disk])582 f = file(disk, "r+")583 v = msdosfs(f)584 root = v.root()585 586 # Long name entries (valid or not!) preceding volume label587 bytes = make_long_dirent('Test1GB', ATTR_VOLUME_ID)588 root.write_slots(0, bytes)589 590 # Create a file with a known good long name591 root.mkfile('The quick brown fox jumped over the lazy dog')592 593 # Create a file with a known good short name594 root.mkfile('foo.bar')595 596 # Create a file with invalid long name entries (bad checksums)597 bytes = make_long_dirent('Greetings and felicitations my friends', ATTR_ARCHIVE)598 bytes = bytes[0:-32] + 'HELLO ' + bytes[-21:]599 assert len(bytes) % 32 == 0600 slots = len(bytes) / 32601 slot = root.find_slots(slots)602 root.write_slots(slot, bytes)603 604 subdir = root.mkdir('SubDir')605 606 # Create a file with incomplete long name entries607 # Missing first (LONG_NAME_LAST) entry608 bytes = make_long_dirent('To be or not to be', ATTR_ARCHIVE)[32:]609 slots = len(bytes) / 32610 slot = subdir.find_slots(slots)611 subdir.write_slots(slot, bytes)612 613 # Missing middle (second) long entry614 bytes = make_long_dirent('A Man a Plan a Canal Panama', ATTR_ARCHIVE)615 bytes = bytes[:32] + bytes[64:]616 slots = len(bytes) / 32617 slot = subdir.find_slots(slots)618 subdir.write_slots(slot, bytes)619 # Missing last long entry620 bytes = make_long_dirent('We the People in order to form a more perfect union', ATTR_ARCHIVE)621 bytes = bytes[0:-64] + bytes[-32:]622 slots = len(bytes) / 32623 slot = subdir.find_slots(slots)624 subdir.write_slots(slot, bytes)625 subdir = root.mkdir('Bad Orders')626 627 # Bad order value: first628 bytes = make_long_dirent('One is the loneliest number', ATTR_ARCHIVE)629 bytes = chr(ord(bytes[0])+7) + bytes[1:]630 slots = len(bytes) / 32631 slot = subdir.find_slots(slots)632 subdir.write_slots(slot, bytes)633 634 # Bad order value: middle635 bytes = make_long_dirent('It takes two to tango or so they say', ATTR_ARCHIVE)636 bytes = bytes[:32] + chr(ord(bytes[32])+7) + bytes[33:]637 slots = len(bytes) / 32638 slot = subdir.find_slots(slots)639 subdir.write_slots(slot, bytes)640 641 # Bad order value: last642 bytes = make_long_dirent('Threes Company becomes Threes A Crowd', ATTR_ARCHIVE)643 bytes = bytes[:-64] + chr(ord(bytes[-64])+7) + bytes[-63:]644 slots = len(bytes) / 32645 slot = subdir.find_slots(slots)646 subdir.write_slots(slot, bytes)647 648 # Long name entries (valid or not, with no short entry) at end of directory649 bytes = make_long_dirent('Four score and seven years ago', ATTR_ARCHIVE)650 bytes = bytes[0:-32] # Remove the short name entry651 assert len(bytes) % 32 == 0652 slots = len(bytes) / 32653 slot = root.find_slots(slots)654 root.write_slots(slot, bytes)655 656 v.flush()657 del v658 f.close()659 del f660 661 try:662 launch([fsck, '-n', disk])663 except LaunchError:664 pass665 launch([fsck, '-y', disk])666 launch(['/sbin/fsck_msdos', '-n', disk])667def past_end_of_dir(disk, fsck, newfs, newfs_opts, multiple_clusters=False):668 launch([newfs]+newfs_opts+[disk])669 f = file(disk, "r+")670 v = msdosfs(f)671 root = v.root()672 673 if multiple_clusters:674 subdir_clusters = v.fat.find(10)675 else:676 subdir_clusters = None677 subdir = root.mkdir('SubDir', subdir_clusters)678 subdir.mkfile('Good Sub File')679 root.mkfile('Good Root File')680 681 # Make an entry that will be replaced by end-of-directory682 slotEOF = root.find_slots(1)683 root.mkfile('EOF')684 685 # Make some valid file entries past end of directory686 root.mkfile('BADFILE')687 root.mkdir('Bad Dir')688 root.mkfile('Bad File 2')689 690 # Overwrite 'EOF' entry with end-of-directory marker691 root.write_slots(slotEOF, '\x00' * 32)692 693 # Make an entry that will be replaced by end-of-directory694 slotEOF = subdir.find_slots(1)695 subdir.mkfile('EOF')696 697 # Make some valid file entries past end of directory698 subdir.mkfile('BADFILE')699 subdir.mkdir('Bad Dir')700 subdir.mkfile('Bad File 2')701 702 # If desired, make a whole bunch more entries that will cause703 # the directory to grow into at least one more cluster.704 # See Radar #xxxx.705 if multiple_clusters:706 base_name = "This file name is long so that it can take up plenty of room in the directory "707 entry_length = len(make_long_dirent(base_name, 0))708 num_entries = 4 * (v.bytesPerCluster // entry_length)709 for i in xrange(num_entries):710 subdir.mkfile(base_name+str(i))711 712 # Overwrite 'EOF' entry with end-of-directory marker713 subdir.write_slots(slotEOF, '\x00' * 32)714 715 v.flush()716 del v717 f.close()718 del f719 720 try:721 launch([fsck, '-n', disk])722 except LaunchError:723 pass724 launch([fsck, '-y', disk])725 launch(['/sbin/fsck_msdos', '-n', disk])726#727# Stomp the first two FAT entries.728#729def fat_bad_0_or_1(disk, fsck, newfs, newfs_opts):730 launch([newfs]+newfs_opts+[disk])731 f = file(disk, "r+")732 v = msdosfs(f)733 v.fat[0] = 0734 v.fat[1] = 1735 736 v.flush()737 del v738 f.close()739 del f740 741 try:742 launch([fsck, '-n', disk])743 except LaunchError:744 pass745 launch([fsck, '-y', disk])746 launch(['/sbin/fsck_msdos', '-n', disk])747#748# Mark the volume dirty, and cause some minor damage (orphan clusters).749# Make sure the volume gets marked clean afterwards.750#751def fat_mark_clean_corrupt(disk, fsck, newfs, newfs_opts):752 launch([newfs]+newfs_opts+[disk])753 f = file(disk, "r+")754 v = msdosfs(f)755 # Mark the volume "dirty" by clearing the "clean" bit.756 if v.type == 32:757 v.fat[1] = v.fat[1] & 0x07FFFFFF758 else:759 v.fat[1] = v.fat[1] & 0x7FFF760 761 # Allocate some clusters, so there is something to repair.762 v.allocate(3)763 764 v.flush()765 del v766 f.close()767 del f768 769 try:770 launch([fsck, '-n', disk])771 except LaunchError:772 pass773 launch([fsck, '-y', disk])774 775 f = file(disk, "r")776 v = msdosfs(f)777 # Make sure the "clean" bit is now set.778 if v.type == 32:779 clean = v.fat[1] & 0x08000000780 else:781 clean = v.fat[1] & 0x8000782 if not clean:783 raise RuntimeError("Volume still dirty!")784 785 v.flush()786 del v787 f.close()788 del f789 launch(['/sbin/fsck_msdos', '-n', disk])790#791# Mark the volume dirty (with no corruption).792# Make sure the volume gets marked clean afterwards.793# Make sure the exit status is 0, even with "-n".794#795def fat_mark_clean_ok(disk, fsck, newfs, newfs_opts):796 launch([newfs]+newfs_opts+[disk])797 798 f = file(disk, "r+")799 v = msdosfs(f)800 801 # Mark the volume "dirty" by clearing the "clean" bit.802 if v.type == 32:803 v.fat[1] = v.fat[1] & 0x07FFFFFF804 else:805 v.fat[1] = v.fat[1] & 0x7FFF806 807 v.flush()808 del v809 f.close()810 del f811 812 # Make sure that we ask the user to mark the disk clean, but don't return813 # a non-zero exit status if the user declines.814 stdout, stderr = launch([fsck, '-n', disk], stdout=subprocess.PIPE, stderr=subprocess.PIPE)815 assert "\nMARK FILE SYSTEM CLEAN? no\n" in stdout816 assert "\n***** FILE SYSTEM IS LEFT MARKED AS DIRTY *****\n" in stdout817 818 stdout, stderr = launch([fsck, '-y', disk], stdout=subprocess.PIPE, stderr=subprocess.PIPE)819 assert "\nMARK FILE SYSTEM CLEAN? yes\n" in stdout820 assert "\nMARKING FILE SYSTEM CLEAN\n" in stdout821 822 f = file(disk, "r")823 v = msdosfs(f)824 825 # Make sure the "clean" bit is now set.826 if v.type == 32:827 clean = v.fat[1] & 0x08000000828 else:829 clean = v.fat[1] & 0x8000830 if not clean:831 raise RuntimeError("Volume still dirty!")832 833 v.flush()834 del v835 f.close()836 del f837 838 launch(['/sbin/fsck_msdos', '-n', disk])839#840# Make a file whose physical size is 4GB. The logical size is 4GB-100.841# This is actually NOT corrupt; it's here to verify that fsck_msdos does not842# try to truncate the file due to overflow of the physical size. [4988133]843#844def file_4GB(disk, fsck, newfs, newfs_opts):845 launch([newfs]+newfs_opts+[disk])846 #847 # Create a file whose size is 4GB-100. That means its physical size will848 # be rounded up to the next multiple of the cluster size, meaning the849 # physical size will be 4GB.850 #851 print "# Creating a 4GiB file. This may take some time."852 f = file(disk, "r+")853 v = msdosfs(f)854 four_GB = 4*1024*1024*1024855 clusters = four_GB / v.bytesPerCluster856 head = v.allocate(clusters)857 v.root().mkfile('4GB', head=head, length=four_GB-100)858 859 v.flush()860 del v861 f.close()862 del f863 864 launch([fsck, '-n', disk])865#866# Make a file with excess clusters allocated: over 4GB worth of clusters867#868# TODO: What combination of files do we want to test with?869# TODO: A smallish logical size870# TODO: A logical size just under 4GB871# TODO: Cross-linked files?872# TODO: Cross linked beyond 4GB?873# TODO: Cross linked before 4GB?874#875def file_4GB_excess_clusters(disk, fsck, newfs, newfs_opts):876 launch([newfs]+newfs_opts+[disk])877 #878 # Create files with too many clusters for their size879 #880 print "# Creating a 4GiB+ file. This may take some time."881 f = file(disk, "r+")882 v = msdosfs(f)883 four_GB = 4*1024*1024*1024884 clusters = four_GB / v.bytesPerCluster885 head=v.allocate(clusters+7)886 v.root().mkfile('FOO', head=head, length=5*v.bytesPerCluster-100)887 head=v.allocate(clusters+3)888 v.root().mkfile('BAR', head=head, length=four_GB-30)889 890 v.flush()891 del v892 f.close()893 del f894 895 # TODO: Need a better way to assert that the disk is corrupt to start with896 try:897 launch([fsck, '-n', disk])898 except LaunchError:899 pass900 launch([fsck, '-y', disk])901 launch([fsck, '-n', disk])902#903# Test the "-q" ("quick") option which reports whether the "dirty" flag in904# the FAT has been set. The dirty flag is only defined for FAT16 and FAT32.905# For FAT12, we actually do a full verify of the volume and return that the906# volume is clean if it has no problems, dirty if a problem was detected.907#908# NOTE: Assumes newfs_opts[1] is "12", "16" or "32" to indicate which FAT909# type is being tested.910#911def test_quick(disk, fsck, newfs, newfs_opts):912 assert newfs_opts[1] in ["12", "16", "32"]913 launch([newfs]+newfs_opts+[disk])914 # Try a quick check of a volume that is clean915 launch([fsck, '-q', disk])916 917 # Make the volume dirty918 f = file(disk, "r+")919 v = msdosfs(f)920 if newfs_opts[1] in ["16", "32"]:921 if newfs_opts[1] == "16":922 v.fat[1] &= 0x7FFF923 else:924 v.fat[1] &= 0x07FFFFFF925 else:926 # Corrupt a FAT12 volume so that it looks dirty.927 # Allocate some clusters, so there is something to repair.928 v.allocate(3)929 v.flush()930 del v931 f.close()932 del f933 934 # Quick check a dirty volume935 try:936 launch([fsck, '-q', disk])937 except LaunchError:938 pass939 else:940 raise FailureExpected("Volume not dirty?")941#942# Test the "-M" (memory limit) option. This just tests the argument parsing.943# It does not verify that fsck_msdos actually limits its memory usage.944#945def test_maxmem(disk, fsck, newfs, newfs_opts):946 launch([newfs]+newfs_opts+[disk])947 launch([fsck, '-M', '1m', disk])948 launch([fsck, '-M', '900k', disk])949#950# Test several combinations of bad arguments.951#952def test_bad_args(disk, fsck, newfs, newfs_opts):953 launch([newfs]+newfs_opts+[disk])954 955 try:956 launch([fsck, '-M', 'foo', disk])957 except LaunchError:958 pass959 else:960 raise FailureExpected("Expected bad argument: -M foo")961 try:962 launch([fsck, '-p', '-M', 'foo', disk])963 except LaunchError:964 pass965 else:966 raise FailureExpected("Expected bad argument: -p -M foo")967 try:968 launch([fsck, '-M', '1x', disk])969 except LaunchError:970 pass971 else:972 raise FailureExpected("Expected bad argument: -M 1x")973 try:974 launch([fsck, '-z', disk])975 except LaunchError:976 pass977 else:978 raise FailureExpected("Expected bad argument: -z")979 try:980 launch([fsck])981 except LaunchError:982 pass983 else:984 raise FailureExpected("Expected usage (no disk given)")985#986# Test several cases of bad values in the boot sector.987# Assumes testing on a disk with 512 bytes per sector.988# These are all fatal, so don't try repairing.989#990def test_boot_sector(disk, fsck, newfs, newfs_opts):991 # Corrupt the jump instruction992 def bad_jump(bytes):993 return 'H+' + bytes[2:]994 995 # Corrupt the sector size (set it to 0x03FF)996 def bad_sector_size(bytes):997 return bytes[0:11] + '\xff\x03' + bytes[13:]998 999 # Corrupt the sectors per cluster value1000 def bad_sec_per_clust(bytes):1001 return bytes[0:13] + '\x07' + bytes[14:]1002 1003 for func, reason in [(bad_jump,"Bad boot jump"),1004 (bad_sector_size, "Bad sector size"),1005 (bad_sec_per_clust, "Bad sectors per cluster")]:1006 launch([newfs]+newfs_opts+[disk])1007 with open(disk, "r+") as f:1008 bytes = f.read(512)1009 f.seek(0)1010 bytes = func(bytes)1011 f.write(bytes)1012 try:1013 launch([fsck, '-n', disk])1014 except LaunchError:1015 pass1016 else:1017 raise FailureExpected(reason)1018#1019# Test several cases of bad values in the boot sector (FAT32 only).1020# These are all fatal, so don't try repairing.1021#1022def test_boot_fat32(disk, fsck, newfs, newfs_opts):1023 # Non-zero number of root directory entries1024 def bad_root_count(bytes):1025 return bytes[0:17] + '\x00\x02' + bytes[19:]1026 1027 # Non-zero file system version1028 def bad_version(bytes):1029 return bytes[0:42] + '\x00\x01' + bytes[44:]1030 1031 for func, reason in [(bad_root_count,"Bad root entry count"),1032 (bad_version, "Bad filesystem version")]:1033 launch([newfs]+newfs_opts+[disk])1034 with open(disk, "r+") as f:1035 bytes = f.read(512)1036 f.seek(0)1037 bytes = func(bytes)1038 f.write(bytes)1039 try:1040 launch([fsck, '-n', disk])1041 except LaunchError:1042 pass1043 else:1044 raise FailureExpected(reason)1045#1046# Test several cases of bad values in the boot sector (FAT32 only).1047#1048def test_fsinfo(disk, fsck, newfs, newfs_opts):1049 def bad_leading_sig(bytes):1050 return 'RRAA' + bytes[4:]1051 1052 def bad_sig2(bytes):1053 return bytes[0:484] + 'rraa' + bytes[488:]1054 1055 def bad_trailing_sig(bytes):1056 return bytes[0:508] + '\xff\x00\xaa\x55' + bytes[512:]1057 1058 def zero_free_count(bytes):1059 return bytes[0:488] + '\x00\x00\x00\x00' + bytes[492:]1060 1061 def unknown_free_count(bytes):1062 return bytes[0:488] + '\xFF\xFF\xFF\xFF' + bytes[492:]1063 1064 # Figure out where the FSInfo sector ended up1065 launch([newfs]+newfs_opts+[disk])1066 with open(disk, "r+") as f:1067 bytes = f.read(512)1068 fsinfo = ord(bytes[48]) + 256 * ord(bytes[49])1069 1070 # Test each of the FSInfo corruptions1071 for func, reason in [(bad_leading_sig, "Bad leading signature"),1072 (bad_sig2, "Bad structure signature"),1073 (bad_trailing_sig, "Bad trailing signature"),1074 ]:1075 launch([newfs]+newfs_opts+[disk])1076 with open(disk, "r+") as f:1077 f.seek(fsinfo * 512)1078 bytes = f.read(512)1079 f.seek(fsinfo * 512)1080 bytes = func(bytes)1081 f.write(bytes)1082 try:1083 launch([fsck, '-n', disk])1084 except LaunchError:1085 pass1086 else:1087 raise FailureExpected(reason)1088 launch([fsck, '-y', disk])1089 launch([fsck, '-n', disk])1090 # Test the free cluster count field. This minor error DOES NOT1091 # set the exit status, but it is repairable.1092 for func, reason, msg in [1093 (zero_free_count, "Zero free cluster count", "Free space in FSInfo block (0) not correct (5232654)\n"),1094 (unknown_free_count, "Unknown free cluster count","Free space in FSInfo block is unset (should be 5232654)\n")]:1095 launch([newfs]+newfs_opts+[disk])1096 with open(disk, "r+") as f:1097 f.seek(fsinfo * 512)1098 bytes = f.read(512)1099 f.seek(fsinfo * 512)1100 bytes = func(bytes)1101 f.write(bytes)1102 stdout,stderr = launch([fsck, '-n', disk], stdout=subprocess.PIPE)1103 sys.stdout.write(stdout)1104 if msg not in stdout:1105 raise FailureExpected(reason)1106 stdout,stderr = launch([fsck, '-y', disk], stdout=subprocess.PIPE)1107 sys.stdout.write(stdout)1108 if msg not in stdout:1109 raise FailureExpected(reason)1110 stdout,stderr = launch([fsck, '-n', disk], stdout=subprocess.PIPE)1111 sys.stdout.write(stdout)1112 if msg in stdout:1113 raise RepairFailed(reason)1114#1115# Test when the FAT has too few sectors for the number of clusters.1116#1117# NOTE: Relies on the fact that newfs_msdos places a FAT32 root1118# directory in cluster #2 (immediately following the FATs).1119#1120def fat_too_small(disk, fsck, newfs, newfs_opts):1121 launch([newfs]+newfs_opts+[disk])1122 1123 with open(disk, "r+") as f:1124 bytes = f.read(512)1125 numFATs = ord(bytes[16])1126 reserved = ord(bytes[14]) + 256 * ord(bytes[15])1127 # Decrement the number of sectors per FAT1128 if bytes[22:24] != '\x00\x00':1129 fat_sectors = struct.unpack("<H", bytes[22:24])[0]1130 bytes = bytes[0:22] + struct.pack("<H", fat_sectors - 1) + bytes[24:]1131 else:1132 fat_sectors = struct.unpack("<I", bytes[36:40])[0]1133 bytes = bytes[0:36] + struct.pack("<I", fat_sectors - 1) + bytes[40:]1134 f.seek(0)1135 f.write(bytes)1136 1137 # Copy the root directory from old location to new location1138 f.seek(512 * (reserved + numFATs * fat_sectors))1139 bytes = f.read(65536)1140 f.seek(512 * (reserved + numFATs * fat_sectors - numFATs))1141 f.write(bytes)1142 # NOTE: FAT too small is NOT a fatal error1143 launch([fsck, '-y', disk]) # Need a way to test for expected output1144 launch([fsck, '-n', disk])1145#1146# Test trying to repair a disk where a directory has been overwritten by1147# garbage (and therefore shouldn't be parsed as a directory). See1148# <rdar://problem/15384158> Better repairs for directories containing garbage1149#1150def directory_garbage(disk, fsck, newfs, newfs_opts):1151 launch([newfs]+newfs_opts+[disk])1152 f = file(disk, "r+")1153 v = msdosfs(f)1154 root = v.root()1155 # Create a subdirectory with some children1156 subdir = root.mkdir('EFI', v.fat.find(2))1157 for i in range(10):1158 subdir.mkfile("Child {}".format(i), content="This is child number {}".format(i))1159 #1160 # Now clobber the contents of the subdirectory by overwriting it with text.1161 #1162 # Note: the length of this text is 887 bytes, which could be larger than a1163 # cluster (minimum cluster size is 512). Be sure the directory is large enough!1164 #1165 subdir.pwrite(0, """Here's to the crazy ones. The misfits. The rebels. The troublemakers. The round pegs in the square holes.1166The ones who see things differently. They're not fond of rules. And they have no respect for the status quo. You can quote them, disagree with them, glorify or vilify them.1167About the only thing you can't do is ignore them. Because they change things. They invent. They imagine. They heal. They explore. They create. They inspire. They push the human race forward.1168Maybe they have to be crazy.1169How else can you stare at an empty canvas and see a work of art? Or sit in silence and hear a song that's never been written? Or gaze at a red planet and see a laboratory on wheels?1170We make tools for these kinds of people.1171While some see them as the crazy ones, we see genius. Because the people who are crazy enough to think they can change the world, are the ones who do.1172""")1173 1174 v.flush()1175 f.close()1176 try:1177 launch([fsck, '-n', disk])1178 except LaunchError:1179 pass1180 else:1181 raise FailureExpected("Subdirectory /EFI should be corrupt")1182 launch([fsck, '-y', disk])1183 launch([fsck, '-n', disk])1184 1185 # Make sure /EFI has been changed to a file1186 f = file(disk, "r+")1187 v = msdosfs(f)1188 root = v.root()1189 efi = root.pread(32, 32) # The directory entry for /EFI1190 assert(efi[0:11] == "EFI ")1191 assert(efi[11] == "\x00") # No longer has ATTR_DIRECTORY set1192 f.close()1193 1194#1195# Test quick check and repair on a FAT12 volume with large cluster size.1196# See <rdar://problem/18523205> Alert "OSX canât repair the disk XXXâ appears when mount FAT12 storage1197#1198def test_18523205(disk, fsck, newfs, newfs_opts):1199 launch([newfs]+newfs_opts+[disk])1200 1201 # Create some subdirectories with children1202 with open(disk, "r+") as f:1203 v = msdosfs(f)1204 root = v.root()1205 1206 for i in range(10):1207 subdir = root.mkdir('Folder {}'.format(i))1208 subdir.mkfile('Child {}'.format(i), content='This is child number {}'.format(i))1209 1210 v.flush()1211 # Try quick check, verify and explicit repair 1212 launch([fsck, '-q', disk])1213 launch([fsck, '-n', disk])1214 launch([fsck, '-y', disk])1215#1216# When run as a script, run the test suite.1217#1218# Usage:1219# python test_fsck.py [<fsck_msdos> [<tmp_dir>]]1220#1221if __name__ == '__main__':1222 #1223 # Set up defaults1224 #1225 dir = '/tmp'1226 fsck = 'fsck_msdos'1227 newfs = 'newfs_msdos'1228 if len(sys.argv) > 1:1229 fsck = sys.argv[1]1230 if len(sys.argv) > 2:1231 dir = sys.argv[2]1232 if len(sys.argv) > 3:1233 print "%s: Too many arguments!" % sys.argv[0]1234 print "Usage: %s [<fsck_msdos> [<tmp_dir>]]"1235 sys.exit(1)1236 1237 #1238 # Run the test suite1239 #1240 test_fat32(dir, fsck, newfs)1241 test_fat16(dir, fsck, newfs)1242 test_fat12(dir, fsck, newfs)1243 test_fat12_100MB(dir, fsck, newfs)1244 ...
systemd-fsckd
Source:systemd-fsckd
1#!/usr/bin/python32# autopkgtest check: Ensure that systemd-fsckd can report progress and cancel3# (C) 2015 Canonical Ltd.4# Author: Didier Roche <didrocks@ubuntu.com>5from contextlib import suppress6import inspect7import fileinput8import os9import subprocess10import shutil11import stat12import sys13import unittest14from time import sleep, time15GRUB_AUTOPKGTEST_CONFIG_PATH = "/etc/default/grub.d/90-autopkgtest.cfg"16TEST_AUTOPKGTEST_CONFIG_PATH = "/etc/default/grub.d/99-fsckdtest.cfg"17SYSTEMD_ETC_SYSTEM_UNIT_DIR = "/etc/systemd/system/"18SYSTEMD_PROCESS_KILLER_PATH = os.path.join(SYSTEMD_ETC_SYSTEM_UNIT_DIR, "process-killer.service")19SYSTEMD_FSCK_ROOT_PATH = "/lib/systemd/system/systemd-fsck-root.service"20SYSTEMD_FSCK_ROOT_ENABLE_PATH = os.path.join(SYSTEMD_ETC_SYSTEM_UNIT_DIR, 'local-fs.target.wants/systemd-fsck-root.service')21SYSTEM_FSCK_PATH = '/sbin/fsck'22PROCESS_KILLER_PATH = '/sbin/process-killer'23SAVED_FSCK_PATH = "{}.real".format(SYSTEM_FSCK_PATH)24FSCKD_TIMEOUT = 3025class FsckdTest(unittest.TestCase):26 '''Check that we run, report and can cancel fsck'''27 def __init__(self, test_name, after_reboot, return_code):28 super().__init__(test_name)29 self._test_name = test_name30 self._after_reboot = after_reboot31 self._return_code = return_code32 def setUp(self):33 super().setUp()34 # ensure we have our root fsck enabled by default (it detects it runs in a vm and doesn't pull the target)35 # note that it can already exists in case of a reboot (as there was no tearDown as we wanted)36 os.makedirs(os.path.dirname(SYSTEMD_FSCK_ROOT_ENABLE_PATH), exist_ok=True)37 with suppress(FileExistsError):38 os.symlink(SYSTEMD_FSCK_ROOT_PATH, SYSTEMD_FSCK_ROOT_ENABLE_PATH)39 enable_plymouth()40 # note that the saved real fsck can still exists in case of a reboot (as there was no tearDown as we wanted)41 if not os.path.isfile(SAVED_FSCK_PATH):42 os.rename(SYSTEM_FSCK_PATH, SAVED_FSCK_PATH)43 # install mock fsck and killer44 self.install_bin(os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'test', 'mocks', 'fsck'),45 SYSTEM_FSCK_PATH)46 self.install_bin(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'process-killer'),47 PROCESS_KILLER_PATH)48 self.files_to_clean = [SYSTEMD_FSCK_ROOT_ENABLE_PATH, SYSTEM_FSCK_PATH, SYSTEMD_PROCESS_KILLER_PATH, PROCESS_KILLER_PATH]49 def tearDown(self):50 # tearDown is only called once the test really ended (not while rebooting during tests)51 for f in self.files_to_clean:52 with suppress(FileNotFoundError):53 os.remove(f)54 os.rename(SAVED_FSCK_PATH, SYSTEM_FSCK_PATH)55 super().tearDown()56 def test_fsckd_run(self):57 '''Ensure we can reboot after a fsck was processed'''58 if not self._after_reboot:59 self.reboot()60 else:61 self.assertFsckdStop()62 self.assertFsckProceeded()63 self.assertSystemRunning()64 def test_fsckd_run_without_plymouth(self):65 '''Ensure we can reboot without plymouth after a fsck was processed'''66 if not self._after_reboot:67 enable_plymouth(enable=False)68 self.reboot()69 else:70 self.assertFsckdStop()71 self.assertFsckProceeded(with_plymouth=False)72 self.assertSystemRunning()73 def test_fsck_with_failure(self):74 '''Ensure that a failing fsck doesn't prevent fsckd to stop'''75 if not self._after_reboot:76 self.install_process_killer_unit('fsck')77 self.reboot()78 else:79 self.assertFsckdStop()80 self.assertTrue(self.was_running('process-killer'))81 self.assertFalse(self.is_failed_unit('process-killer'))82 self.assertFsckProceeded()83 self.assertSystemRunning()84 def test_systemd_fsck_with_failure(self):85 '''Ensure that a failing systemd-fsck doesn't prevent fsckd to stop'''86 if not self._after_reboot:87 self.install_process_killer_unit('systemd-fsck', kill=True)88 self.reboot()89 else:90 self.assertFsckdStop()91 self.assertProcessKilled()92 self.assertTrue(self.is_failed_unit('systemd-fsck-root'))93 self.assertTrue(self.was_running('systemd-fsckd'))94 self.assertTrue(self.was_running('plymouth-start'))95 self.assertSystemRunning()96 def test_systemd_fsckd_with_failure(self):97 '''Ensure that a failing systemd-fsckd doesn't prevent system to boot'''98 if not self._after_reboot:99 self.install_process_killer_unit('systemd-fsckd', kill=True)100 self.reboot()101 else:102 self.assertFsckdStop()103 self.assertProcessKilled()104 self.assertFalse(self.is_failed_unit('systemd-fsck-root'))105 self.assertTrue(self.is_failed_unit('systemd-fsckd'))106 self.assertTrue(self.was_running('plymouth-start'))107 self.assertSystemRunning()108 def test_systemd_fsck_with_plymouth_failure(self):109 '''Ensure that a failing plymouth doesn't prevent fsckd to reconnect/exit'''110 if not self._after_reboot:111 self.install_process_killer_unit('plymouthd', kill=True)112 self.reboot()113 else:114 self.assertFsckdStop()115 self.assertProcessKilled()116 self.assertFsckProceeded()117 self.assertTrue(self.is_failed_unit('plymouth-start'))118 self.assertSystemRunning()119 def install_bin(self, source, dest):120 '''install mock fsck'''121 shutil.copy2(source, dest)122 st = os.stat(dest)123 os.chmod(dest, st.st_mode | stat.S_IEXEC)124 def is_active_unit(self, unit):125 '''Check that given unit is active'''126 if subprocess.call(['systemctl', '-q', 'is-active', unit]) != 0:127 return False128 return True129 def is_failed_unit(self, unit):130 '''Check that given unit failed'''131 p = subprocess.Popen(['systemctl', 'is-active', unit], stdout=subprocess.PIPE)132 out, err = p.communicate()133 if b'failed' in out:134 return True135 return False136 def was_running(self, unit):137 '''Check that a given unit has been running'''138 out = subprocess.check_output(['systemctl', 'show', '--property=ExecMainExitTimestampMonotonic', unit])139 if b'=0' in out:140 return False141 return True142 def assertFsckdStop(self):143 '''Ensure systemd-fsckd stops, which indicates no more fsck activity'''144 timeout = time() + FSCKD_TIMEOUT145 while time() < timeout:146 if not self.is_active_unit('systemd-fsckd'):147 return148 sleep(1)149 raise Exception("systemd-fsckd still active after {}s".format(FSCKD_TIMEOUT))150 def assertFsckProceeded(self, with_plymouth=True):151 '''Assert we executed most of the fsck-related services successfully'''152 self.assertTrue(self.was_running('systemd-fsckd'))153 self.assertFalse(self.is_failed_unit('systemd-fsckd'))154 self.assertTrue(self.is_active_unit('systemd-fsck-root')) # remains active after exit155 if with_plymouth:156 self.assertTrue(self.was_running('plymouth-start'))157 else:158 self.assertFalse(self.was_running('plymouth-start'))159 def assertSystemRunning(self):160 '''Assert we started lightdm and xorg (and so, consider the system running)'''161 out = subprocess.check_output(['ps', 'u', '-C', 'lightdm'])162 self.assertIn(b'lightdm --session', out)163 out = subprocess.check_output(['ps', 'u', '-C', 'Xorg'])164 self.assertIn(b':0', out)165 self.assertTrue(self.is_active_unit('lightdm'))166 def assertProcessKilled(self):167 '''Assert the targeted process was killed successfully'''168 self.assertTrue(self.was_running('process-killer'))169 self.assertFalse(self.is_failed_unit('process-killer'))170 def reboot(self):171 '''Reboot the system with the current test marker'''172 subprocess.check_call(['/tmp/autopkgtest-reboot', "{}:{}".format(self._test_name, self._return_code)])173 def install_process_killer_unit(self, process_name, kill=False):174 '''Create a systemd unit which will kill process_name'''175 with open(SYSTEMD_PROCESS_KILLER_PATH, 'w') as f:176 f.write('''[Unit]177DefaultDependencies=no178[Service]179Type=simple180ExecStart=/usr/bin/timeout 10 {} {}181[Install]182WantedBy=systemd-fsck-root.service'''.format(PROCESS_KILLER_PATH,183 '--signal SIGKILL {}'.format(process_name) if kill else process_name))184 subprocess.check_call(['systemctl', 'daemon-reload'])185 subprocess.check_call(['systemctl', 'enable', 'process-killer'], stderr=subprocess.DEVNULL)186def enable_plymouth(enable=True):187 '''ensure plymouth is enabled in grub config (doesn't reboot)'''188 plymouth_enabled = 'splash' in open('/boot/grub/grub.cfg').read()189 if enable and not plymouth_enabled:190 if os.path.exists(GRUB_AUTOPKGTEST_CONFIG_PATH):191 shutil.copy2(GRUB_AUTOPKGTEST_CONFIG_PATH, TEST_AUTOPKGTEST_CONFIG_PATH)192 for line in fileinput.input([TEST_AUTOPKGTEST_CONFIG_PATH], inplace=True):193 if line.startswith("GRUB_CMDLINE_LINUX_DEFAULT"):194 print(line[:line.rfind('"')] + ' splash quiet"\n')195 else:196 with open(TEST_AUTOPKGTEST_CONFIG_PATH, 'w') as f:197 f.write('GRUB_CMDLINE_LINUX_DEFAULT="console=ttyS0 splash quiet"\n')198 elif not enable and plymouth_enabled:199 with suppress(FileNotFoundError):200 os.remove(TEST_AUTOPKGTEST_CONFIG_PATH)201 subprocess.check_call(['update-grub'], stderr=subprocess.DEVNULL)202def boot_with_systemd_distro():203 '''Reboot with systemd as init and distro setup for grub'''204 enable_plymouth()205 subprocess.check_call(['/tmp/autopkgtest-reboot', 'systemd-started'])206def getAllTests(unitTestClass):207 '''get all test names in predictable sorted order from unitTestClass'''208 return sorted([test[0] for test in inspect.getmembers(unitTestClass, predicate=inspect.isfunction)209 if test[0].startswith('test_')])210# ADT_REBOOT_MARK contains the test name to pursue after reboot211# (to check results and states after reboot, mostly).212# we append the previous global return code (0 or 1) to it.213# Example: ADT_REBOOT_MARK=test_foo:0214if __name__ == '__main__':215 all_tests = getAllTests(FsckdTest)216 reboot_marker = os.getenv('ADT_REBOOT_MARK')217 current_test_after_reboot = ""218 if not reboot_marker:219 boot_with_systemd_distro()220 # first test221 if reboot_marker == "systemd-started":222 current_test = all_tests[0]223 return_code = 0224 else:225 (current_test_after_reboot, return_code) = reboot_marker.split(':')226 current_test = current_test_after_reboot227 return_code = int(return_code)228 # loop on remaining tests to run229 try:230 remaining_tests = all_tests[all_tests.index(current_test):]231 except ValueError:232 print("Invalid value for ADT_REBOOT_MARK, {} is not a valid test name".format(reboot_marker))233 sys.exit(2)234 # run all remaining tests235 for test_name in remaining_tests:236 after_reboot = False237 # if this tests needed a reboot (and it has been performed), executes second part of it238 if test_name == current_test_after_reboot:239 after_reboot = True240 suite = unittest.TestSuite()241 suite.addTest(FsckdTest(test_name, after_reboot, return_code))242 result = unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)243 if len(result.failures) != 0 or len(result.errors) != 0:244 return_code = 1...
fsck_test.py
Source:fsck_test.py
...34 return self.create_hg_repo("main")35 def setup_eden_test(self) -> None:36 super().setup_eden_test()37 self.overlay = overlay_mod.OverlayStore(self.eden, self.mount_path)38 def run_fsck(self, *args: str) -> Tuple[int, str]:39 """Run `eden fsck [args]` and return a tuple of the return code and40 the combined stdout and stderr.41 The command output will be decoded as UTF-8 and returned as a string.42 """43 cmd_result = self.eden.run_unchecked(44 "fsck", *args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT45 )46 fsck_out = cmd_result.stdout.decode("utf-8", errors="replace")47 return (cmd_result.returncode, fsck_out)48 def test_fsck_force_and_check_only(self) -> None:49 """Test the behavior of the --force and --check-only fsck flags."""50 foo_overlay_path = self.overlay.materialize_file(pathlib.Path("doc/foo.txt"))51 # Running fsck with the mount still mounted should fail52 returncode, fsck_out = self.run_fsck(self.mount)53 self.assertIn(f"Not checking {self.mount}", fsck_out)54 self.assertEqual(FSCK_RETCODE_SKIPPED, returncode)55 # Running fsck with --force should override that56 returncode, fsck_out = self.run_fsck(self.mount, "--force")57 self.assertIn("warning: could not obtain lock", fsck_out)58 self.assertIn("scanning anyway due to --force", fsck_out)59 self.assertIn(f"Checking {self.mount}", fsck_out)60 self.assertEqual(FSCK_RETCODE_OK, returncode)61 # fsck should perform the check normally without --force62 # if the mount is not mounted63 self.eden.run_cmd("unmount", self.mount)64 returncode, fsck_out = self.run_fsck(self.mount)65 self.assertIn(f"Checking {self.mount}", fsck_out)66 self.assertIn("No issues found", fsck_out)67 self.assertEqual(FSCK_RETCODE_OK, returncode)68 # Truncate the overlay file for doc/foo.txt to 0 length69 with foo_overlay_path.open("wb"):70 pass71 # Running fsck with --check-only should report the error but not try to fix it.72 returncode, fsck_out = self.run_fsck("--check-only")73 self.assertIn(f"Checking {self.mount}", fsck_out)74 self.assertRegex(75 fsck_out,76 r"invalid overlay file for materialized file .* \(doc/foo.txt\).*: "77 r"zero-sized overlay file",78 )79 self.assertRegex(fsck_out, r"\b1 errors")80 self.assertRegex(fsck_out, "Not fixing errors: --check-only was specified")81 self.assertEqual(FSCK_RETCODE_ERRORS, returncode)82 # Running fsck with no arguments should attempt to fix the errors83 returncode, fsck_out = self.run_fsck()84 self.assertRegex(85 fsck_out,86 r"invalid overlay file for materialized file .* \(doc/foo.txt\).*: "87 r"zero-sized overlay file",88 )89 self.assertRegex(fsck_out, r"\b1 errors")90 self.assertRegex(fsck_out, "Beginning repairs")91 self.assertRegex(92 fsck_out, "replacing corrupt file inode 'doc/foo.txt' with an empty file"93 )94 self.assertRegex(fsck_out, "Fixed 1 of 1 issues")95 self.assertEqual(FSCK_RETCODE_ERRORS, returncode)96 # There should be no more errors if we run fsck again97 returncode, fsck_out = self.run_fsck()98 self.assertIn(f"Checking {self.mount}", fsck_out)99 self.assertIn("No issues found", fsck_out)100 self.assertEqual(FSCK_RETCODE_OK, returncode)101 def test_fsck_multiple_mounts(self) -> None:102 mount2 = Path(self.mounts_dir) / "second_mount"103 mount3 = Path(self.mounts_dir) / "third_mount"104 mount4 = Path(self.mounts_dir) / "fourth_mount"105 self.eden.clone(self.repo.path, mount2)106 self.eden.clone(self.repo.path, mount3)107 self.eden.clone(self.repo.path, mount4)108 # Unmount all but mount3109 self.eden.unmount(Path(self.mount))110 self.eden.unmount(mount2)111 self.eden.unmount(mount4)112 # Running fsck should check all but mount3113 returncode, fsck_out = self.run_fsck()114 self.assertIn(f"Checking {self.mount}", fsck_out)115 self.assertIn(f"Checking {mount2}", fsck_out)116 self.assertIn(f"Not checking {mount3}", fsck_out)117 self.assertIn(f"Checking {mount4}", fsck_out)118 self.assertEqual(FSCK_RETCODE_SKIPPED, returncode)119 # Running fsck with --force should check everything120 returncode, fsck_out = self.run_fsck("--force")121 self.assertIn(f"Checking {self.mount}", fsck_out)122 self.assertIn(f"Checking {mount2}", fsck_out)123 self.assertIn(f"Checking {mount3}", fsck_out)124 self.assertIn(f"Checking {mount4}", fsck_out)125 self.assertEqual(FSCK_RETCODE_OK, returncode)126@testcase.eden_test127class FsckTestNoEdenfs(testcase.IntegrationTestCase, TemporaryDirectoryMixin):128 def test_fsck_no_checkouts(self) -> None:129 tmp_dir = self.make_temporary_directory()130 eden = edenclient.EdenFS(Path(tmp_dir))131 cmd_result = eden.run_unchecked(132 "fsck",133 stdout=subprocess.PIPE,134 stderr=subprocess.PIPE,...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
