66Filesystem monitoring with Fuse and Python
77https://github.com/pleiszenburg/loggedfs-python
88
9- tests/loggedfs_libtest/pre .py: Stuff happening before test(s)
9+ tests/lib/base .py: Base class for test infrastructure
1010
1111 Copyright (C) 2017-2018 Sebastian M. Ernst <[email protected] > 1212
3030# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
3131
3232import os
33+ import time
3334
3435from .const import (
3536 TEST_FS_EXT4 ,
3637 TEST_FS_LOGGEDFS ,
38+ TEST_FSCK_FN ,
3739 TEST_FSTEST_LOG_FN ,
3840 TEST_IMAGE_FN ,
3941 TEST_IMAGE_SIZE_MB ,
4042 TEST_MOUNT_CHILD_PATH ,
4143 TEST_MOUNT_PARENT_PATH ,
44+ TEST_LOG_HEAD ,
4245 TEST_LOG_PATH ,
4346 TEST_ROOT_PATH ,
4447 TEST_LOGGEDFS_CFG_FN ,
4548 TEST_LOGGEDFS_ERR_FN ,
4649 TEST_LOGGEDFS_LOG_FN ,
47- TEST_LOGGEDFS_OUT_FN
50+ TEST_LOGGEDFS_OUT_FN ,
4851 )
4952from .mount import (
5053 attach_loop_device ,
5457 mount ,
5558 mount_loggedfs_python ,
5659 umount ,
57- umount_fuse
60+ umount_fuse ,
5861 )
59- from .lib import (
62+ from .procio import (
63+ ck_filesystem ,
6064 create_zero_file ,
6165 mk_filesystem ,
6266 run_command ,
63- write_file
67+ write_file ,
6468 )
6569
6670
6771# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
6872# CLASS: (1/3) PRE
6973# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
7074
71- class fstest_pre_class ():
75+ class fstest_base_class ():
7276
7377
7478 with_sudo = True
7579 fs_type = TEST_FS_LOGGEDFS
7680 travis = False
7781
7882
79- def init (self , fs_type = TEST_FS_LOGGEDFS ):
83+ def init_a_members (self , fs_type = TEST_FS_LOGGEDFS ):
8084
8185 self .fs_type = fs_type
82- self .set_paths ()
8386
84- if not self .travis :
85- self .__cleanup_logfiles__ () # rm -r log_dir
86- self .__cleanup_mountpoint__ (self .mount_child_abs_path ) # umount & rmdir
87- self .__cleanup_mountpoint__ (self .mount_parent_abs_path ) # umount & rmdir
88- self .__cleanup_loop_devices__ () # losetup -d /dev/loopX
89- self .__cleanup_image__ () # rm file
87+ self .prj_abs_path = os .getcwd ()
88+ self .root_abs_path = os .path .abspath (os .path .join (self .prj_abs_path , TEST_ROOT_PATH ))
89+ assert os .path .isdir (self .root_abs_path )
90+
91+ self .logs_abs_path = os .path .join (self .root_abs_path , TEST_LOG_PATH )
92+
93+ self .image_abs_path = os .path .join ('/dev/shm' , TEST_IMAGE_FN )
94+
95+ self .mount_parent_abs_path = os .path .join (self .root_abs_path , TEST_MOUNT_PARENT_PATH )
96+ self .mount_child_abs_path = os .path .join (self .mount_parent_abs_path , TEST_MOUNT_CHILD_PATH )
97+
98+ self .loggedfs_log_abs_path = os .path .join (self .logs_abs_path , TEST_LOGGEDFS_LOG_FN )
99+ self .loggedfs_cfg_abs_path = os .path .join (self .root_abs_path , TEST_LOGGEDFS_CFG_FN )
100+
101+ if 'TRAVIS' in os .environ .keys ():
102+ self .travis = os .environ ['TRAVIS' ] == 'true'
103+
104+ self .fstest_log_abs_path = os .path .join (self .logs_abs_path , TEST_FSTEST_LOG_FN )
105+
106+
107+ def init_b_cleanup (self ):
108+
109+ if self .travis :
110+ return
111+
112+ self .__cleanup_logfiles__ () # rm -r log_dir
113+ self .__cleanup_mountpoint__ (self .mount_child_abs_path ) # umount & rmdir
114+ self .__cleanup_mountpoint__ (self .mount_parent_abs_path ) # umount & rmdir
115+ self .__cleanup_loop_devices__ () # losetup -d /dev/loopX
116+ self .__cleanup_image__ () # rm file
117+
118+
119+ def init_c_parentfs (self ):
90120
91121 if not self .travis :
92122 create_zero_file (self .image_abs_path , TEST_IMAGE_SIZE_MB )
@@ -98,35 +128,66 @@ def init(self, fs_type = TEST_FS_LOGGEDFS):
98128 self .__mk_dir__ (self .mount_child_abs_path , in_fs_root = True )
99129 self .__mk_dir__ (self .logs_abs_path )
100130
131+
132+ def init_d_childfs (self ):
133+
101134 open (self .loggedfs_log_abs_path , 'a' ).close () # HACK create empty loggedfs log file
102135 if self .fs_type == TEST_FS_LOGGEDFS :
103136 self .__mount_child_fs__ ()
137+ open (self .fstest_log_abs_path , 'a' ).close () # HACK create empty fstest log file
104138
105- open (self .fstest_log_abs_path , 'a' ).close ()
106139
107- os . chdir (self . mount_child_abs_path )
140+ def destroy_a_childfs (self ):
108141
142+ if not self .fs_type == TEST_FS_LOGGEDFS :
143+ return
109144
110- def set_paths (self ):
145+ if not is_path_mountpoint (self .mount_child_abs_path ):
146+ return
111147
112- self .prj_abs_path = os . getcwd ( )
113- self . root_abs_path = os . path . abspath ( os . path . join ( self . prj_abs_path , TEST_ROOT_PATH ))
114- assert os . path . isdir (self .root_abs_path )
148+ umount_child_status = umount_fuse ( self .mount_child_abs_path , sudo = self . with_sudo )
149+ assert umount_child_status
150+ assert not is_path_mountpoint (self .mount_child_abs_path )
115151
116- self . logs_abs_path = os . path . join ( self . root_abs_path , TEST_LOG_PATH )
152+ time . sleep ( 0.1 ) # HACK ... otherwise parent will be busy
117153
118- self .image_abs_path = os .path .join ('/dev/shm' , TEST_IMAGE_FN )
119154
120- self .mount_parent_abs_path = os .path .join (self .root_abs_path , TEST_MOUNT_PARENT_PATH )
121- self .mount_child_abs_path = os .path .join (self .mount_parent_abs_path , TEST_MOUNT_CHILD_PATH )
155+ def destroy_b_parentfs (self ):
122156
123- self . loggedfs_log_abs_path = os . path . join ( self .logs_abs_path , TEST_LOGGEDFS_LOG_FN )
124- self . loggedfs_cfg_abs_path = os . path . join ( self . root_abs_path , TEST_LOGGEDFS_CFG_FN )
157+ if self .travis :
158+ return
125159
126- if 'TRAVIS' in os .environ .keys ():
127- self .travis = os .environ ['TRAVIS' ] == 'true'
160+ # TODO mountpoint checks ...
128161
129- self .fstest_log_abs_path = os .path .join (self .logs_abs_path , TEST_FSTEST_LOG_FN )
162+ umount_parent_status = umount (self .mount_parent_abs_path , sudo = True )
163+ assert umount_parent_status
164+ assert not is_path_mountpoint (self .mount_parent_abs_path )
165+
166+ loop_device_list = find_loop_devices (self .image_abs_path )
167+ assert isinstance (loop_device_list , list )
168+ assert len (loop_device_list ) == 1
169+ loop_device_path = loop_device_list [0 ]
170+
171+ ck_status_code , ck_out , ck_err = ck_filesystem (loop_device_path )
172+ write_file (
173+ os .path .join (self .logs_abs_path , TEST_FSCK_FN ),
174+ '' .join ([
175+ TEST_LOG_HEAD % 'EXIT STATUS' ,
176+ '%d\n ' % ck_status_code ,
177+ TEST_LOG_HEAD % 'OUT' ,
178+ ck_out ,
179+ TEST_LOG_HEAD % 'ERR' ,
180+ ck_err
181+ ])
182+ )
183+
184+ detach_status = detach_loop_device (loop_device_path )
185+ assert detach_status
186+
187+ assert not bool (ck_status_code ) # not 0 for just about any type of error! Therefore asserting at the very end.
188+
189+ os .remove (self .image_abs_path )
190+ assert not os .path .exists (self .image_abs_path )
130191
131192
132193 def __attach_loop_device__ (self ):
0 commit comments