Menu
▾
▴
[Jython-dev] javaos.py from 2.1 extended for OS/2
|
From: Ype K. <yk...@xs...> - 2002-01-10 19:35:19
|
Dear developers,
This works mostly fine using javaos.py library below:
import os
os._ _testSystem() () # ok.
os._testGetOsType() # fails on None, but this is not very irrelevant.
javaos.py from 2.1 only needed adding OS/2 as os.name and using the same
functionality as NT (yes, once upon a time there was a common predecessor..)
Have fun,
Ype
"""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)" ),
( "os2", r"(OS/2)" ),
( "mac", r"(mac)|(MacOS.*)|(Darwin)" ),
( "mac", r"(mac)|(MacOS.*)|(Darwin)" ),
( "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") or (_osType == "os2"):
_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",
"OS/2": "os2",
"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()
--
|
