######################################## # The contents of this file are subject to the MLX PUBLIC LICENSE version # 1.0 (the "License"); you may not use this file except in # compliance with the License. # # Software distributed under the License is distributed on an "AS IS" # basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See # the License for the specific language governing rights and limitations # under the License. # # The Original Source Code is "compClust", released 2003 September 03. # # The Original Source Code was developed by the California Institute of # Technology (Caltech). Portions created by Caltech are Copyright (C) # 2002-2003 California Institute of Technology. All Rights Reserved. ######################################## # # Authors: Ben Bornstein, Diane Trout # Last Modified: 10-Apr-2001, 2:00p # """Assorted assertion utilities.""" import os import sys import types import stat ERROR_STREAM = sys.stderr def environment_variables_exist(required_variable_list, error_message=None): """environment_variables_exist(required_variable_list, error_message) Checks the operating system environment variables for the existance of each variable name in required_variable_list. If a variable name does not exist, an error message is printed. When one or more required variables do not exist, this function completes by exiting the Python interpreter (via sys.exit(1)). The error_message parameter is optional. If omitted, it defaults to \"Environment variable must be defined: [%s].\". error_message is a string that may contain one string format specifier (%s) which is replaced with the name of the missing variable. """ if error_message == None: error_message = "Environment variable must be defined: [%s]." parameters_exist( required_variable_list, os.environ, error_message ) def fs_objects_have_owner(pathname, owner=None): """fs_objects_have_owner(pathnames, owner=None) Given a list of files and/or directories verify that the owner is correct. The owner parameter should be one of: None -- if the current process should own the file. string -- if the name defined by the string should own the file. user_id -- if the numeric user id given should own the file. """ # Index into passwd structure that contains the user id PASSWD_ENT_UID = 2 # # FIXME: assert_fs_objects_have_owner() not yet implemented. # fs_obj_stat = os.stat(pathname) # # FIXME: has_owner has quite a few unix-isms in it # # if no owner lookup the current effecive user id if owner == None: owner = os.geteuid() # if owner is a name convert to uid if type(owner) == types.TypeString: import pwd try: passwd_ent = pwd.getpwname(owner) except KeyError, error: # if we can't find the name in the database it's not a valid # owner return 0 owner = passwd_ent[PASSWD_ENT_UID] return owner == fs_obj_stat[stat.ST_UID] def fs_object_has_permissions(pathname, permissions, relative=1): """boolean fs_object_has_permissions(pathname, permissions, relative) Helper function for fs_object_have_permissions """ if relative: return os.access(pathname, permissions) else: # FIXME: absolute mode assumes a unix system fs_obj_stat = os.stat(pathname) return fs_obj_stat[stat.ST_MODE] == permissions def fs_objects_have_permissions(pathnames, permissions=0700, relative=1): """fs_objects_have_permissions(pathnames, permissions, relative) Given a list of files and/or directories verify that the permissions are correct. There are two modes of operation either 'relative' mode or 'absolute' mode. In relative mode (the default) the permission checks are done with respect to the current user id. The value of the permissions flag are those provided by the os package. (os.F_OK, os.R_OK, os.W_OK, and os.X_OK) In absolute (relative=0) mode the file checks to make sure that the permissions of the file precisely match that of the octal User, Group, World bit-vector being passed in. """ # If given a sequence type iterate over the components if type(pathnames) == types.TupleType or \ type(pathnames) == types.ListType: for fs_obj in pathnames: if not fs_object_has_permissions(fs_obj, permissions, relative): return 0 # if we checked everything without returning false, # everything therefore must be okay. return 1 elif type(pathnames) == types.StringType: return fs_object_has_permissions(pathnames, permissions, relative) else: raise AttributeError("fs_objects_have_permissions requires a sequence type.") def parameters_exist(required_variable_list, dictionary, error_message=None): """parameters_exist(required_variable_list, dictionary, error_message) Checks dictionary for the existance of each variable name (key) in required_variable_list. If a variable name does not occur in dictionary, an error message is printed. When one or more required variables are missing from dictionary, this function completes by exiting the Python interpreter (via sys.exit(1)). The error_message parameter is optional. If omitted, it defaults to \"Parameter must be defined: [%s].\". error_message is a string that may contain one string format specifier (%s) which is replaced with the name of the missing variable. """ check_failed = 0 if error_message == None: error_message = "Parameter must be defined: [%s]." for required_variable in required_variable_list: if not dictionary.has_key( required_variable ): ERROR_STREAM.write( error_message % (required_variable) + "\n" ) check_failed = 1 if check_failed: sys.exit(1) def get_error_stream(): """stream = get_error_stream() Returns the error stream used by the Assert module. """ return ERROR_STREAM def set_error_stream(stream): """set_error_stream(stream) Sets the error stream used by the Assert module to stream. The stream object must have a write method defined. If it does not an AttributeError exception is raised. """ global ERROR_STREAM try: getattr( stream, "write" ) ERROR_STREAM = stream except AttributeError: raise AttributeError, \ "compClust.util.Assert.set_error_stream(): " + \ "stream must contain a write method."