org.aspectj/lib/jython/Lib/javaos.py

r"""OS routines for Java, with some attempts to support DOS, NT, and
Posix functionality.

This exports:
  - all functions from posix, nt, dos, os2, mac, or ce, e.g. unlink, stat, etc.
  - os.path is one of the modules posixpath, ntpath, macpath, or dospath
  - os.name is 'posix', 'nt', 'dos', 'os2', 'mac', 'ce' or 'riscos'
  - os.curdir is a string representing the current directory ('.' or ':')
  - os.pardir is a string representing the parent directory ('..' or '::')
  - os.sep is the (or a most common) pathname separator ('/' or ':' or '\\')
  - os.altsep is the alternate pathname separator (None or '/')
  - os.pathsep is the component separator used in $PATH etc
  - os.linesep is the line separator in text files ('\r' or '\n' or '\r\n')
  - os.defpath is the default search path for executables

Programs that import and use 'os' stand a better chance of being
portable between different platforms.  Of course, they must then
only use functions that are defined by all platforms (e.g., unlink
and opendir), and leave all pathname manipulation to os.path
(e.g., split and join).
"""

__all__ = ["altsep", "curdir", "pardir", "sep", "pathsep", "linesep",
           "defpath", "name"]

import java
from java.io import File, BufferedReader, InputStreamReader, IOException
import javapath as path
from UserDict import UserDict
import string
import exceptions
import re
import sys
import thread


error = OSError

name = 'java' # discriminate based on JDK version?
curdir = '.'  # default to Posix for directory behavior, override below
pardir = '..' 
sep = java.io.File.separator
altsep = None
pathsep = java.io.File.pathSeparator
defpath = '.'
linesep = java.lang.System.getProperty('line.separator')

def _exit(n=0):
    java.lang.System.exit(n)

def getcwd():
    foo = File(File("foo").getAbsolutePath())
    return foo.getParent()

def chdir(path):
    raise OSError(0, 'chdir not supported in Java', path)

def listdir(path):
    l = File(path).list()
    if l is None:
        raise OSError(0, 'No such directory', path)
    return list(l)

def mkdir(path, mode='ignored'):
    if not File(path).mkdir():
        raise OSError(0, "couldn't make directory", path)

def makedirs(path, mode='ignored'):
    if not File(path).mkdirs():
        raise OSError(0, "couldn't make directories", path)

def remove(path):
    if not File(path).delete():
        raise OSError(0, "couldn't delete file", path)

def rename(path, newpath):
    if not File(path).renameTo(File(newpath)):
        raise OSError(0, "couldn't rename file", path)

def rmdir(path):
    if not File(path).delete():
        raise OSError(0, "couldn't delete directory", path)

unlink = remove

def stat(path):
    """The Java stat implementation only returns a small subset of
    the standard fields"""
    f = File(path)
    size = f.length()
    # Sadly, if the returned length is zero, we don't really know if the file
    # is zero sized or does not exist.
    if size == 0 and not f.exists():
        raise OSError(0, 'No such file or directory', path)
    mtime = f.lastModified() / 1000.0
    return (0, 0, 0, 0, 0, 0, size, mtime, mtime, 0)

def utime(path, times):
    # Only the modification time is changed (and only on java2).
    if times and hasattr(File, "setLastModified"):
        File(path).setLastModified(long(times[1] * 1000.0))

class LazyDict( UserDict ):
    """A lazy-populating User Dictionary.
    Lazy initialization is not thread-safe.
    """
    def __init__( self,
                  dict=None,
                  populate=None,
                  keyTransform=None ):
        """dict: starting dictionary of values
        populate: function that returns the populated dictionary
        keyTransform: function to normalize the keys (e.g., toupper/None)
        """
        UserDict.__init__( self, dict )
        self._populated = 0
        self.__populateFunc = populate or (lambda: {})
        self._keyTransform = keyTransform or (lambda key: key)

    def __populate( self ):
        if not self._populated:
            self.data = self.__populateFunc()
            self._populated = 1 # race condition

    ########## extend methods from UserDict by pre-populating
    def __repr__(self):
        self.__populate()
        return UserDict.__repr__( self )
    def __cmp__(self, dict):
        self.__populate()
        return UserDict.__cmp__( self, dict )
    def __len__(self):
        self.__populate()
        return UserDict.__len__( self )
    def __getitem__(self, key):
        self.__populate()
        return UserDict.__getitem__( self, self._keyTransform(key) )
    def __setitem__(self, key, item):
        self.__populate()
        UserDict.__setitem__( self, self._keyTransform(key), item )
    def __delitem__(self, key):
        self.__populate()
        UserDict.__delitem__( self, self._keyTransform(key) )
    def clear(self):
        self.__populate()
        UserDict.clear( self )
    def copy(self):
        self.__populate()
        return UserDict.copy( self )
    def keys(self):
        self.__populate()
        return UserDict.keys( self )
    def items(self):
        self.__populate()
        return UserDict.items( self )
    def values(self):
        self.__populate()
        return UserDict.values( self )
    def has_key(self, key):
        self.__populate()
        return UserDict.has_key( self, self._keyTransform(key) )
    def update(self, dict):
        self.__populate()
        UserDict.update( self, dict )
    def get(self, key, failobj=None):
        self.__populate()
        return UserDict.get( self, self._keyTransform(key), failobj )
    def setdefault(self, key, failobj=None):
        self.__populate()
        return UserDict.setdefault( self, self._keyTransform(key), failobj )
    def popitem(self):
        self.__populate()
        return UserDict.popitem( self )


class _ShellEnv:
    """Provide environment derived by spawning a subshell and parsing its
    environment.  Also supports system functions and provides empty
    environment support for platforms with unknown shell
    functionality.
    """
    def __init__( self, cmd=None, getEnv=None, keyTransform=None ):
        """cmd: list of exec() arguments to run command in subshell, or None
        getEnv: shell command to list environment variables, or None
        keyTransform: normalization function for environment keys, or None
        """
        self.cmd = cmd
        self.getEnv = getEnv
        self.environment = LazyDict(populate=self._getEnvironment,
                                    keyTransform=keyTransform)
        self._keyTransform = self.environment._keyTransform

    ########## system
    def system( self, cmd ):
        """Imitate the standard library 'system' call.
        Execute 'cmd' in a shell, and send output to stdout & stderr.
        """
        p = self.execute( cmd )

        def println( arg, write=sys.stdout.write ):
            write( arg + "\n" )
        def printlnStdErr( arg, write=sys.stderr.write ):
            write( arg + "\n" )
            
        # read stderr in new thread
        thread.start_new_thread( self._readLines,
                                ( p.getErrorStream(), printlnStdErr ))
        # read stdin in main thread
        self._readLines( p.getInputStream(), println )
        
        return p.waitFor()

    def execute( self, cmd ):
        """Execute cmd in a shell, and return the process instance"""
        shellCmd = self._formatCmd( cmd )
        if self.environment._populated:
            env = self._formatEnvironment( self.environment )
        else:
            env = None
        try:
            p = java.lang.Runtime.getRuntime().exec( shellCmd, env )
            return p
        except IOException, ex:
            raise OSError(
                0,
                "Failed to execute command (%s): %s" % ( shellCmd, ex )
                )

    ########## utility methods
    def _readLines( self, stream, func=None ):
        """Read lines of stream, and either append them to return
        array of lines, or call func on each line.
        """
        lines = []
        func = func or lines.append
        # should read both stderror and stdout in separate threads...
        bufStream = BufferedReader( InputStreamReader( stream ))
        while 1:
            line = bufStream.readLine()
            if line is None: break
            func( line )
        return lines or None

    def _formatCmd( self, cmd ):
        """Format a command for execution in a shell."""
        if self.cmd is None:
            msgFmt = "Unable to execute commands in subshell because shell" \
                     " functionality not implemented for OS %s with shell"  \
                     " setting %s. Failed command=%s""" 
            raise OSError( 0, msgFmt % ( _osType, _envType, cmd ))
            
        return self.cmd + [cmd]

    def _formatEnvironment( self, env ):
        """Format enviroment in lines suitable for Runtime.exec"""
        lines = []
        for keyValue in env.items():
            lines.append( "%s=%s" % keyValue )
        return lines

    def _getEnvironment( self ):
        """Get the environment variables by spawning a subshell.
        This allows multi-line variables as long as subsequent lines do
        not have '=' signs.
        """
        env = {}
        if self.getEnv:
            try:
                p = self.execute( self.getEnv )
                lines = self._readLines( p.getInputStream() )
                if '=' not in lines[0]:
                    print "getEnv command (%s) did not print environment.\n" \
                        "Output=%s" % (
                        self.getEnv, '\n'.join( lines )
                        )
                    return env

                for line in lines:
                    try:
                        i = line.index( '=' )
                        key = self._keyTransform(line[:i])
                        value = line[i+1:]
                    except ValueError:
                        # found no '=', so line is part of previous value
                        value = '%s\n%s' % ( value, line )
                    env[ key ] = value
            except OSError, ex:
                print "Failed to get environment, environ will be empty:", ex
        return env

def _getOsType( os=None ):
    """Select the OS behavior based on os argument, 'python.os' registry
    setting and 'os.name' Java property.
    os: explicitly select desired OS. os=None to autodetect, os='None' to
    disable 
    """
    os = os or sys.registry.getProperty( "python.os" ) or \
               java.lang.System.getProperty( "os.name" )
        
    _osTypeMap = (
        ( "nt", r"(nt)|(Windows NT)|(Windows NT 4.0)|(WindowsNT)|"
                r"(Windows 2000)|(Windows XP)|(Windows CE)" ),
        ( "dos", r"(dos)|(Windows 95)|(Windows 98)|(Windows ME)" ),
        ( "mac", r"(mac)|(MacOS.*)|(Darwin)" ),
        ( "None", r"(None)" ),
        ( "posix", r"(.*)" ), # default - posix seems to vary mast widely
        )
    for osType, pattern in _osTypeMap:
        if re.match( pattern, os ):
            break
    return osType

def _getShellEnv( envType, shellCmd, envCmd, envTransform ):
    """Create the desired environment type.
    envType: 'shell' or None
    """
    if envType == "shell":
        return _ShellEnv( shellCmd, envCmd, envTransform )
    else:
        return _ShellEnv()
    
_osType = _getOsType()
_envType = sys.registry.getProperty("python.environment", "shell")

# default to None/empty for shell and environment behavior
_shellCmd = None
_envCmd = None
_envTransform = None

# override defaults based on _osType
if _osType == "nt":
    _shellCmd = ["cmd", "/c"]
    _envCmd = "set"
    _envTransform = string.upper
elif _osType == "dos":
    _shellCmd = ["command.com", "/c"]
    _envCmd = "set"
    _envTransform = string.upper
elif _osType == "posix":
    _shellCmd = ["sh", "-c"]
    _envCmd = "env"
elif _osType == "mac":
    curdir = ':'  # override Posix directories
    pardir = '::' 
elif _osType == "None":
    pass
# else:
#    # may want a warning, but only at high verbosity:
#    warn( "Unknown os type '%s', using default behavior." % _osType )

_shellEnv = _getShellEnv( _envType, _shellCmd, _envCmd, _envTransform )

# provide environ, putenv, getenv
environ = _shellEnv.environment
putenv = environ.__setitem__
getenv = environ.__getitem__
# provide system
system = _shellEnv.system

########## test code
def _testGetOsType():
    testVals = {
        "Windows NT": "nt",
        "Windows 95": "dos",
        "MacOS": "mac",
        "Solaris": "posix",
        "Linux": "posix",
        "None": "None"
        }

    msgFmt = "_getOsType( '%s' ) should return '%s', not '%s'"
    # test basic mappings
    for key, val in testVals.items():
        got = _getOsType( key )
        assert got == val, msgFmt % ( key, val, got )

def _testCmds( _shellEnv, testCmds, whichEnv ):
    # test commands (key) and compare output to expected output (value).
    # this actually executes all the commands twice, testing the return
    # code by calling system(), and testing some of the output by calling
    # execute()
    for cmd, pattern in testCmds:
        print "\nExecuting '%s' with %s environment" % (cmd, whichEnv)
        assert not _shellEnv.system( cmd ), \
                "%s failed with %s environment" % (cmd, whichEnv)
        line = _shellEnv._readLines(
            _shellEnv.execute(cmd).getInputStream())[0]
        assert re.match( pattern, line ), \
                "expected match for %s, got %s" % ( pattern, line )
    
def _testSystem( shellEnv=_shellEnv ):
    # test system and environment functionality
    key, value = "testKey", "testValue"
    org = environ
    testCmds = [
        # test commands and regexes to match first line of expected
        # output on first and second runs
        # Note that the validation is incomplete for several of these
        # - they should validate depending on platform and pre-post, but
        # they don't.

        # no quotes, should output both words
        ("echo hello there", "hello there"),
        # should print PATH (on NT)
        ("echo PATH=%PATH%", "(PATH=.*;.*)|(PATH=%PATH%)"),
        # should print 'testKey=%testKey%' on NT before initialization,
        # should print 'testKey=' on 95 before initialization,
        # and 'testKey=testValue' after
        ("echo %s=%%%s%%" % (key,key),
                "(%s=)" % (key,)),     
        # should print PATH (on Unix)
        ( "echo PATH=$PATH", "PATH=.*" ),
        # should print 'testKey=testValue' on Unix after initialization
        ( "echo %s=$%s" % (key,key),
                "(%s=$%s)|(%s=)|(%s=%s)" % (key, key, key, key, value ) ), 
        # should output quotes on NT but not on Unix
        ( 'echo "hello there"', '"?hello there"?' ),
        # should print 'why' to stdout. 
        ( r'''jython -c "import sys;sys.stdout.write( 'why\n' )"''', "why" ),
        # should print 'why' to stderr, but it won't right now.  Have
        # to add the print to give some output...empty string matches every
        # thing...
        ( r'''jython -c "import sys;sys.stderr.write('why\n');print " ''',
          "" )
        ]
    
    assert not environ._populated, \
            "before population, environ._populated should be false"

    _testCmds( _shellEnv, testCmds, "default" )
    
    # trigger initialization of environment
    environ[ key ] = value
    
    assert environ._populated, \
            "after population, environ._populated should be true"
    assert org.get( key, None ) == value, \
            "expected stub to have %s set" % key
    assert environ.get( key, None ) == value, \
            "expected real environment to have %s set" % key

    # test system using the non-default environment
    _testCmds( _shellEnv, testCmds, "initialized" )
    
    assert environ.has_key( "PATH" ), \
            "expected environment to have PATH attribute " \
            "(this may not apply to all platforms!)"

def _testBadShell():
    # attempt to get an environment with a shell that is not startable
    se2 = _ShellEnv( ["badshell", "-c"], "set" )
    str(se2.environment) # trigger initialization
    assert not se2.environment.items(), "environment should be empty"

def _testBadGetEnv():
    # attempt to get an environment with a command that does not print an environment
    se2 = _getShellEnv( "shell", _shellCmd, _envCmd, _envTransform )
    se2.getEnv="echo This command does not print environment"
    str(se2.environment) # trigger initialization
    assert not se2.environment.items(), "environment should be empty"
    
def _test():
    _testGetOsType()
    _testBadShell()
    _testBadGetEnv()
    _testSystem()