diff -r 9ea84f006892 Lib/argparse.py --- a/Lib/argparse.py Wed May 01 15:15:50 2013 +0200 +++ b/Lib/argparse.py Sun Sep 22 22:57:01 2013 -0700 @@ -67,6 +67,7 @@ 'ArgumentError', 'ArgumentTypeError', 'FileType', + 'FileContext', 'HelpFormatter', 'ArgumentDefaultsHelpFormatter', 'RawDescriptionHelpFormatter', @@ -1179,6 +1180,118 @@ if arg is not None]) return '%s(%s)' % (type(self).__name__, args_str) +from functools import partial as _partial +class FileContext(object): + """ + like FileType but with autoclose (if reasonable) + issue13824 + meant to be used as + with args.input() as f: ... + 3 modes? + - immediate open, essentially like FileType + - delayed open, ie to open file until use in 'with' + - checked, like delayed, but with immediate checks on file existience etc + sys.stdin/out options need some sort of cover so they can be used in context + without being closed + """ + class StdContext(object): + # a class meant to wrap stdin/out; + # allows them to be used with 'with' but without being closed + def __init__(self, stdfile): + self.file = stdfile + try: + self.name = self.file.name + except AttributeError: + self.name = self.file + + def __enter__(self): + return self.file + + def __exit__(self, exc_type, exc_val, exc_tb): + pass + + def __eq__(self, other): + # match on the file rather the context + if isinstance(other, type(self)): + return self.file == other.file + else: + return self.file == other + + def __ne__(self, other): + return not (self == other) + + def __repr__(self): + return 'StdContext(%r)'% self.file + + def __init__(self, mode='r', bufsize=-1, encoding=None, errors=None, style='delayed'): + self._mode = mode + self._bufsize = bufsize + self._encoding = encoding + self._errors = errors + self._style = style + + def _ostest(self, string): + # os.access to test this string + # raise error if problem + if string != '-': + if 'r' in self._mode: + if _os.access(string, _os.R_OK): + pass + else: + message = _("can't open '%s' for read") + raise ArgumentTypeError(message % (string,)) + if 'w' in self._mode: + dir = _os.path.dirname(string) or '.' + if _os.access(string, _os.W_OK): + pass + elif _os.access(dir, _os.W_OK): + pass + else: + message = _("can't open '%s' for write") + raise ArgumentTypeError(message % (string,)) + return string + + def __call__(self, string): + if self._style == 'delayed': + return self.__delay_call__(string) + elif self._style == 'evaluate': + # evaluate the delayed result right away + try: + result = self.__delay_call__(string) + return result() + except OSError as e: + message = _("can't open '%s': %s") + raise ArgumentTypeError(message % (string, e)) + elif self._style in ['osaccess','test']: + # test before returning a 'delayed' object + string = self._ostest(string) + result = self.__delay_call__(string) + return result + else: + raise ArgumentTypeError('Unknown FIleContext style') + + def __delay_call__(self, string): + # delayed mode + if string == '-': + if 'r' in self._mode: + return _partial(self.StdContext, _sys.stdin) + elif 'w' in self._mode: + return _partial(self.StdContext, _sys.stdout) + else: + msg = _('argument "-" with mode %r') % self._mode + raise ValueError(msg) + fn = _partial(open,string, self._mode, self._bufsize, self._encoding, + self._errors) + return fn + + def __repr__(self): + args = self._mode, self._bufsize + kwargs = [('encoding', self._encoding), ('errors', self._errors), ('style', self._style)] + args_str = ', '.join([repr(arg) for arg in args if arg != -1] + + ['%s=%r' % (kw, arg) for kw, arg in kwargs + if arg is not None]) + return '%s(%s)' % (type(self).__name__, args_str) + # =========================== # Optional and Positional Parsing # =========================== diff -r 9ea84f006892 Lib/test/test_argparse.py --- a/Lib/test/test_argparse.py Wed May 01 15:15:50 2013 +0200 +++ b/Lib/test/test_argparse.py Sun Sep 22 22:57:01 2013 -0700 @@ -110,6 +110,10 @@ setattr(result, key, old_stdout) if getattr(result, key) is sys.stderr: setattr(result, key, old_stderr) + # try to correctly capture FileContext + if isinstance(getattr(result, key), argparse.FileContext.StdContext): + if getattr(result, key) == sys.stdout: + setattr(result, key, old_stdout) return result except SystemExit: code = sys.exc_info()[1].code @@ -1668,6 +1672,184 @@ # ============ +# FileContext tests +# ============ + +class TestFileContextR(TempDirMixin, ParserTestCase): + """Test the FileContext option/argument type for reading files + with style=immediate, should be just like FileType + except sys.stdin is wrapped in a StdContext""" + + def setUp(self): + super(TestFileContextR, self).setUp() + for file_name in ['foo', 'bar']: + file = open(os.path.join(self.temp_dir, file_name), 'w') + file.write(file_name) + file.close() + self.create_readonly_file('readonly') + + argument_signatures = [ + Sig('-x', type=argparse.FileContext(style='evaluate')), + Sig('spam', type=argparse.FileContext('r',style='evaluate')), + ] + failures = ['-x', '', 'non-existent-file.txt'] + successes = [ + ('foo', NS(x=None, spam=RFile('foo'))), + ('-x foo bar', NS(x=RFile('foo'), spam=RFile('bar'))), + ('bar -x foo', NS(x=RFile('foo'), spam=RFile('bar'))), + ('-x - -', NS(x=sys.stdin, spam=sys.stdin)), + ('readonly', NS(x=None, spam=RFile('readonly'))), + ] + + +@unittest.skipIf(hasattr(os, 'geteuid') and os.geteuid() == 0, + "non-root user required") +class TestFileContextW(TempDirMixin, ParserTestCase): + """Test the FileContext option/argument type for writing files + stdout test requires a modification to stderr_to_parser_error + """ + + def setUp(self): + super(TestFileContextW, self).setUp() + self.create_readonly_file('readonly') + + argument_signatures = [ + Sig('-x', type=argparse.FileContext('w',style='evaluate')), + Sig('spam', type=argparse.FileContext('w',style='evaluate')), + ] + failures = ['-x', '', 'readonly'] + successes = [ + ('foo', NS(x=None, spam=WFile('foo'))), + ('-x foo bar', NS(x=WFile('foo'), spam=WFile('bar'))), + ('bar -x foo', NS(x=WFile('foo'), spam=WFile('bar'))), + ('-x - -', NS(x=sys.stdout, spam=sys.stdout)), + ] + + +class RDFile(object): + # object capable of recognizing an 'equivalent' FileContext object + seen = {} + + def __init__(self, name): + self.name = name + + def __eq__(self, other): + if other in self.seen: + text = self.seen[other] + else: + with other() as f: + # other is file opener wrapped in a partial + other.name = f.name + if other.name == '': + # do not attempt to read from stdin + text = self.seen[other] = other.name + else: + text = self.seen[other] = f.read() + if not isinstance(text, str): + text = text.decode('ascii') + return self.name == other.name == text + + def __repr__(self): + return "RDFile(%r)"%self.name + +class TestFileContextDelayedR(TempDirMixin, ParserTestCase): + """Test the FileContext option/argument type for reading files + with style=delayed. Values in namespace will be a wrapped + file opener.""" + + def setUp(self): + super(TestFileContextDelayedR, self).setUp() + for file_name in ['foo', 'bar']: + file = open(os.path.join(self.temp_dir, file_name), 'w') + file.write(file_name) + file.close() + self.create_readonly_file('readonly') + + argument_signatures = [ + Sig('-x', type=argparse.FileContext(style='delayed')), + Sig('spam', type=argparse.FileContext('r',style='delayed')), + ] + failures = ['-x', + '', + # 'non-existent-file.txt', # delayed does not do existence test + ] + successes = [ + ('foo', NS(x=None, spam=RDFile('foo'))), + ('-x foo bar', NS(x=RDFile('foo'), spam=RDFile('bar'))), + ('bar -x foo', NS(x=RDFile('foo'), spam=RDFile('bar'))), + ('-x - -', NS(x=RDFile(''), spam=RDFile(''))), + ('readonly', NS(x=None, spam=RDFile('readonly'))), + ] + +class TestFileContextAccessR(TempDirMixin, ParserTestCase): + """Test the FileContext option/argument type for reading files + with style=delayed. Values in namespace will be a wrapped + file opener.""" + + def setUp(self): + super(TestFileContextAccessR, self).setUp() + for file_name in ['foo', 'bar']: + file = open(os.path.join(self.temp_dir, file_name), 'w') + file.write(file_name) + file.close() + self.create_readonly_file('readonly') + + argument_signatures = [ + Sig('-x', type=argparse.FileContext(style='osaccess')), + Sig('spam', type=argparse.FileContext('r',style='osaccess')), + ] + failures = ['-x', + '', + 'non-existent-file.txt' + ] + successes = [ + ('foo', NS(x=None, spam=RDFile('foo'))), + ('-x foo bar', NS(x=RDFile('foo'), spam=RDFile('bar'))), + ('bar -x foo', NS(x=RDFile('foo'), spam=RDFile('bar'))), + ('-x - -', NS(x=RDFile(''), spam=RDFile(''))), + ('readonly', NS(x=None, spam=RDFile('readonly'))), + ] + +class TestFileContext(TempDirMixin, TestCase): + """Test FileContext without the larger framework + Deals directly with the distinctive functions of this type + """ + + def setUp(self): + super(TestFileContext, self).setUp() + for file_name in ['foo', 'bar']: + file = open(os.path.join(self.temp_dir, file_name), 'w') + file.write(file_name) + file.close() + self.create_readonly_file('readonly') + + parser = ErrorRaisingArgumentParser() + parser.add_argument('-x', type=argparse.FileContext(style='evaluate')) + parser.add_argument('spam', type=argparse.FileContext('r',style='delayed')) + self.parser = parser + + def test1(self): + args = self.parser.parse_args('-x foo bar'.split()) + self.assertRaises(AttributeError, getattr, args.spam, 'name') + # args.spam is not an opened file + with args.spam() as f: + text = f.read() + self.assertEqual(text, f.name) + self.assertEqual(f.closed, True) + self.assertEqual(args.x.name, 'foo') + with args.x as f: + text = f.read() + self.assertEqual(text, f.name) + self.assertEqual(f.closed, True) + + def test2(self): + args = self.parser.parse_args('-x - -'.split()) + self.assertEqual(args.x, sys.stdin) + with args.spam() as f: + self.assertEqual(f, sys.stdin) + self.assertEqual(f.closed, False) # should not close stdin + +# ============ # Action tests # ============