<s> import os <EOL> from . import FSQTestCase <EOL> from . internal import test_type_own_mode , normalize <EOL> from . constants import ROOT1 , ROOT2 , NON_ASCII , NOT_NORMAL , ILLEGAL_NAMES , ILLEGAL_MODE , ILLEGAL_NAME , ILLEGAL_UID , ILLEGAL_UNAME , ORIG_QUEUE_UG , ORIG_ITEM_UG , MODES , UID , GID , UNAME , GNAME , NOROOT <EOL> from . . import install , uninstall , constants as _c , FSQPathError , FSQInstallError , FSQCoerceError , FSQConfigError <EOL> def _valid_uninstall ( queue ) : <EOL> dirs = os . listdir ( _c . FSQ_ROOT ) <EOL> for d in dirs : <EOL> if d == queue : <EOL> return False <EOL> return True <EOL> def _valid_install ( queue , is_down = None , is_triggered = None , user = None , <EOL> group = None , mode = None , item_user = None , item_group = None , <EOL> item_mode = None ) : <EOL> dirs = os . listdir ( os . path . join ( _c . FSQ_ROOT , queue ) ) <EOL> seen = [ ] <EOL> allowed = set ( [ _c . FSQ_FAIL , _c . FSQ_TMP , _c . FSQ_DONE , _c . FSQ_QUEUE , ] ) <EOL> if is_down : <EOL> allowed = allowed | set ( [ _c . FSQ_DOWN , ] ) <EOL> if is_triggered : <EOL> allowed = allowed | set ( [ _c . FSQ_TRIGGER , ] ) <EOL> for d in dirs : <EOL> if d in allowed : <EOL> seen . append ( d ) <EOL> t_path = os . path . join ( _c . FSQ_ROOT , queue , d ) <EOL> st = os . stat ( t_path ) <EOL> if d not in ( _c . FSQ_DOWN , _c . FSQ_TRIGGER ) : <EOL> if _c . FSQ_QUEUE_USER != ORIG_QUEUE_UG [ <NUM_LIT:0> ] : <EOL> user = <NUM_LIT:1> <EOL> if _c . FSQ_QUEUE_GROUP != ORIG_QUEUE_UG [ <NUM_LIT:1> ] : <EOL> group = <NUM_LIT:1> <EOL> test_type_own_mode ( st , t_path , '<STR_LIT:d>' , <EOL> user if user is None else UID , <EOL> group if group is None else GID , mode ) <EOL> else : <EOL> f_type = '<STR_LIT:f>' <EOL> if d == _c . FSQ_TRIGGER : <EOL> f_type = '<STR_LIT:p>' <EOL> if _c . FSQ_ITEM_USER != ORIG_ITEM_UG [ <NUM_LIT:0> ] : <EOL> user = <NUM_LIT:1> <EOL> if _c . FSQ_ITEM_GROUP != ORIG_ITEM_UG [ <NUM_LIT:1> ] : <EOL> group = <NUM_LIT:1> <EOL> test_type_own_mode ( st , t_path , f_type , <EOL> item_user if item_user is None else UID , <EOL> item_group if item_group is None else GID , item_mode ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' . format ( d ) ) <EOL> if len ( set ( seen ) ^ allowed ) : <EOL> raise ValueError ( '<STR_LIT>' '<STR_LIT>' . format ( set ( seen ) ^ allowed ) ) <EOL> return True <EOL> class TestInstallUninstall ( FSQTestCase ) : <EOL> def _cycle ( self , queue = None , ** kwargs ) : <EOL> uninstall_kwargs = { <EOL> '<STR_LIT>' : kwargs . get ( '<STR_LIT>' , None ) , <EOL> '<STR_LIT>' : kwargs . get ( '<STR_LIT>' , None ) , <EOL> '<STR_LIT>' : kwargs . get ( '<STR_LIT>' , None ) <EOL> } <EOL> queue = self . _install ( queue , ** kwargs ) <EOL> self . _uninstall ( queue , ** uninstall_kwargs ) <EOL> return queue <EOL> def _install ( self , queue = None , ** kwargs ) : <EOL> if queue is None : <EOL> queue = normalize ( ) <EOL> install ( queue , ** kwargs ) <EOL> self . assertTrue ( _valid_install ( queue , ** kwargs ) ) <EOL> return queue <EOL> def _uninstall ( self , queue , ** kwargs ) : <EOL> uninstall ( queue , ** kwargs ) <EOL> self . assertTrue ( _valid_uninstall ( queue ) ) <EOL> def test_basic ( self ) : <EOL> '''<STR_LIT>''' <EOL> for root in ROOT1 , ROOT2 : <EOL> _c . FSQ_ROOT = root <EOL> self . _cycle ( ) <EOL> for down in True , False : <EOL> self . _cycle ( is_down = down ) <EOL> for is_t in True , False : <EOL> self . _cycle ( is_triggered = is_t ) <EOL> self . _cycle ( is_down = down , is_triggered = is_t ) <EOL> queue = normalize ( ) <EOL> _c . FSQ_QUEUE , _c . FSQ_TMP , _c . FSQ_DONE , _c . FSQ_FAIL , _c . FSQ_DOWN = NOT_NORMAL <EOL> self . _cycle ( queue , is_down = down , is_triggered = is_t ) <EOL> queue = normalize ( ) <EOL> _c . FSQ_TMP = NON_ASCII <EOL> self . _cycle ( queue , is_down = down , is_triggered = is_t ) <EOL> def test_collide ( self ) : <EOL> '''<STR_LIT>''' <EOL> for root in ROOT1 , ROOT2 : <EOL> _c . FSQ_ROOT = root <EOL> queue = self . _install ( ) <EOL> self . assertRaises ( FSQInstallError , install , queue ) <EOL> self . _uninstall ( queue ) <EOL> def test_invalroot ( self ) : <EOL> '''<STR_LIT>''' <EOL> queue = normalize ( ) <EOL> _c . FSQ_ROOT = NOROOT <EOL> self . assertRaises ( FSQInstallError , install , queue ) <EOL> self . assertRaises ( FSQInstallError , uninstall , queue ) <EOL> def test_invalnames ( self ) : <EOL> '''<STR_LIT>''' <EOL> for root in ROOT1 , ROOT2 : <EOL> _c . FSQ_ROOT = root <EOL> normalize ( ) <EOL> self . assertRaises ( FSQCoerceError , install , ILLEGAL_NAME ) <EOL> self . assertRaises ( ( OSError , AttributeError , ) , _valid_install , <EOL> ILLEGAL_NAME ) <EOL> for name in ILLEGAL_NAMES : <EOL> normalize ( ) <EOL> self . assertRaises ( FSQPathError , install , name ) <EOL> self . assertRaises ( ( OSError , ValueError , ) , _valid_install , name ) <EOL> for other in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> queue = normalize ( ) <EOL> setattr ( _c , other , name ) <EOL> self . assertRaises ( FSQPathError , install , queue , <EOL> is_down = True , is_triggered = True ) <EOL> self . assertRaises ( OSError , _valid_install , queue , <EOL> is_down = True , is_triggered = True ) <EOL> queue = normalize ( ) <EOL> setattr ( _c , other , ILLEGAL_NAME ) <EOL> self . assertRaises ( FSQCoerceError , install , queue , <EOL> is_down = True , is_triggered = True ) <EOL> self . assertRaises ( ( OSError , ValueError , ) , _valid_install , <EOL> queue , is_down = True , is_triggered = True ) <EOL> def test_usergroup ( self ) : <EOL> '''<STR_LIT>''' <EOL> for uid , gid in ( ( UID , GID , ) , ( UNAME , GNAME , ) ) : <EOL> for root in ROOT1 , ROOT2 : <EOL> _c . FSQ_ROOT = root <EOL> self . _cycle ( user = uid ) <EOL> self . _cycle ( group = gid ) <EOL> self . _cycle ( user = uid , group = gid ) <EOL> queue = normalize ( ) <EOL> _c . FSQ_QUEUE_USER , _c . FSQ_QUEUE_GROUP = ( uid , gid , ) <EOL> self . _cycle ( queue ) <EOL> for down in True , False : <EOL> for is_t in True , False : <EOL> self . _cycle ( user = uid , group = gid , is_down = down , <EOL> is_triggered = is_t ) <EOL> queue = normalize ( ) <EOL> uid = gid = ILLEGAL_UID <EOL> self . assertRaises ( FSQInstallError , install , queue , user = uid ) <EOL> self . assertRaises ( OSError , _valid_install , queue , user = uid ) <EOL> self . assertRaises ( FSQInstallError , install , queue , group = gid ) <EOL> self . assertRaises ( OSError , _valid_install , queue , group = gid ) <EOL> self . assertRaises ( FSQInstallError , install , queue , user = uid , <EOL> group = gid ) <EOL> self . assertRaises ( OSError , _valid_install , queue , user = uid , <EOL> group = gid ) <EOL> queue = normalize ( ) <EOL> uid = gid = ILLEGAL_UNAME <EOL> self . assertRaises ( TypeError , install , queue , user = uid ) <EOL> self . assertRaises ( ( OSError , AttributeError , ) , _valid_install , <EOL> queue , user = uid ) <EOL> self . assertRaises ( TypeError , install , queue , group = gid ) <EOL> self . assertRaises ( ( OSError , AttributeError , ) , _valid_install , <EOL> queue , group = gid ) <EOL> self . assertRaises ( TypeError , install , queue , user = uid , <EOL> group = gid ) <EOL> self . assertRaises ( ( OSError , AttributeError , ) , _valid_install , <EOL> queue , user = uid , group = gid ) <EOL> def test_itemusergroup ( self ) : <EOL> '''<STR_LIT>''' <EOL> for uid , gid in ( ( UID , GID , ) , ( UNAME , GNAME , ) ) : <EOL> for root in ROOT1 , ROOT2 : <EOL> _c . FSQ_ROOT = root <EOL> self . _cycle ( item_user = uid ) <EOL> self . _cycle ( item_group = gid ) <EOL> self . _cycle ( item_user = uid , item_group = gid ) <EOL> queue = normalize ( ) <EOL> _c . FSQ_ITEM_USER , _c . FSQ_ITEM_GROUP = ( uid , gid , ) <EOL> self . _cycle ( ) <EOL> for down in True , False : <EOL> self . _cycle ( item_user = uid , is_down = down ) <EOL> self . _cycle ( item_group = gid , is_down = down ) <EOL> self . _cycle ( item_user = uid , item_group = gid , is_down = down ) <EOL> for is_t in True , False : <EOL> self . _cycle ( item_user = uid , is_triggered = is_t ) <EOL> self . _cycle ( item_group = gid , is_triggered = is_t ) <EOL> self . _cycle ( item_user = uid , item_group = gid , <EOL> is_triggered = is_t ) <EOL> self . _cycle ( item_user = uid , item_group = gid , <EOL> is_triggered = is_t , is_down = down ) <EOL> for quid , qgid in ( ( UID , GID , ) , ( UNAME , GNAME , ) ) : <EOL> self . _cycle ( item_user = uid , user = quid ) <EOL> self . _cycle ( item_user = uid , group = qgid ) <EOL> self . _cycle ( item_user = uid , user = quid , group = qgid ) <EOL> self . _cycle ( item_group = gid , user = quid ) <EOL> self . _cycle ( item_group = gid , group = qgid ) <EOL> self . _cycle ( item_group = gid , user = quid , group = qgid ) <EOL> self . _cycle ( item_user = uid , item_group = gid , <EOL> user = quid ) <EOL> self . _cycle ( item_user = uid , item_group = gid , <EOL> group = qgid ) <EOL> self . _cycle ( item_user = uid , item_group = gid , <EOL> user = quid , group = qgid ) <EOL> self . _cycle ( item_user = uid , item_group = gid , <EOL> is_down = down , is_triggered = is_t , <EOL> user = quid , group = qgid ) <EOL> for uid , gid in ( ( ILLEGAL_UID , ILLEGAL_UID , ) , <EOL> ( ILLEGAL_UNAME , ILLEGAL_UNAME , ) ) : <EOL> queue = normalize ( ) <EOL> uid = gid = ILLEGAL_UID <EOL> self . assertRaises ( FSQConfigError , install , queue , <EOL> item_user = uid , is_down = True ) <EOL> self . assertRaises ( ( OSError , AttributeError , ) , _valid_install , <EOL> queue , item_user = uid , is_down = True ) <EOL> self . assertRaises ( FSQConfigError , install , queue , <EOL> item_group = gid , is_down = True ) <EOL> self . assertRaises ( ( OSError , AttributeError , ) , _valid_install , <EOL> queue , item_group = gid , is_down = True ) <EOL> self . assertRaises ( FSQConfigError , install , queue , <EOL> item_user = uid , item_group = gid , is_down = True ) <EOL> self . assertRaises ( ( OSError , AttributeError , ) , _valid_install , <EOL> queue , item_user = uid , item_group = gid , <EOL> is_down = True ) <EOL> queue = self . _install ( item_user = uid ) <EOL> self . assertRaises ( FSQConfigError , uninstall , queue , <EOL> item_user = uid ) <EOL> self . assertFalse ( os . path . exists ( os . path . join ( root , queue , <EOL> _c . FSQ_DOWN ) ) ) <EOL> queue = self . _install ( item_group = gid ) <EOL> self . assertRaises ( FSQConfigError , uninstall , queue , <EOL> item_group = gid ) <EOL> self . assertFalse ( os . path . exists ( os . path . join ( root , queue , <EOL> _c . FSQ_DOWN ) ) ) <EOL> queue = self . _install ( item_user = uid , item_group = gid ) <EOL> self . assertRaises ( FSQConfigError , uninstall , queue , <EOL> item_user = uid , item_group = gid ) <EOL> self . assertFalse ( os . path . exists ( os . path . join ( root , queue , <EOL> _c . FSQ_DOWN ) ) ) <EOL> def test_mode ( self ) : <EOL> '''<STR_LIT>''' <EOL> for mode in MODES : <EOL> for root in ROOT1 , ROOT2 : <EOL> _c . FSQ_ROOT = root <EOL> self . _cycle ( mode = mode ) <EOL> queue = normalize ( ) <EOL> _c . FSQ_QUEUE_MODE = mode <EOL> self . _cycle ( queue ) <EOL> for down in True , False : <EOL> for is_t in True , False : <EOL> for uid , gid in ( ( UID , GID , ) , ( UNAME , GNAME , ) ) : <EOL> for quid , qgid in ( ( UID , GID , ) , ( UNAME , GNAME , ) ) : <EOL> self . _cycle ( mode = mode , is_down = down , <EOL> is_triggered = is_t , user = quid , <EOL> group = qgid , item_user = uid , <EOL> item_group = gid ) <EOL> queue = normalize ( ) <EOL> mode = ILLEGAL_MODE <EOL> self . assertRaises ( TypeError , install , queue , mode = mode ) <EOL> self . assertRaises ( OSError , _valid_install , queue , mode = mode ) <EOL> def test_itemmode ( self ) : <EOL> '''<STR_LIT>''' <EOL> for mode in MODES : <EOL> for root in ROOT1 , ROOT2 : <EOL> _c . FSQ_ROOT = root <EOL> self . _cycle ( item_mode = mode ) <EOL> queue = normalize ( ) <EOL> _c . FSQ_ITEM_MODE = mode <EOL> self . _cycle ( queue ) <EOL> for down in True , False : <EOL> for is_t in True , False : <EOL> for uid , gid in ( ( UID , GID , ) , ( UNAME , GNAME , ) ) : <EOL> for quid , qgid in ( ( UID , GID , ) , ( UNAME , GNAME , ) ) : <EOL> for qm in MODES : <EOL> self . _cycle ( item_mode = mode , mode = qm , <EOL> is_down = down , user = quid , <EOL> is_triggered = is_t , <EOL> group = qgid , item_user = uid , <EOL> item_group = gid ) <EOL> queue = normalize ( ) <EOL> mode = ILLEGAL_MODE <EOL> self . assertRaises ( TypeError , install , queue , item_mode = mode , <EOL> is_down = True ) <EOL> self . assertRaises ( OSError , _valid_install , queue , item_mode = mode , <EOL> is_down = True ) <EOL> queue = self . _install ( item_mode = mode ) <EOL> self . assertRaises ( TypeError , uninstall , queue , item_mode = mode ) <EOL> self . assertFalse ( _valid_uninstall ( queue ) ) </s>
