Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Returns:
2-tuple containing
[0] = int, file descriptor number for the file object
[1] = string, absolute pathname of a file
Raises:
OSError: when dir= is specified but does not exist
"""
# TODO: optional boolean text is unused?
# default dir affected by "global"
filename = self._TempEntryname(suffix, prefix, dir)
file_handle = \
self._filesystem.CreateFile(filename, st_mode=stat.S_IFREG | 0o600)
file_handle.opened_as = filename
fakefile = fake_filesystem.FakeFileWrapper(file_handle, filename)
file_des = self._filesystem.AddOpenFile(fakefile)
self._mktemp_retvals.append(filename)
return file_des, filename
def __init__(self, file_object, file_path, update=False, read=False, append=False,
delete_on_close=False, filesystem=None, newline=None,
binary=True, closefd=True, encoding=None):
self._file_object = file_object
self._file_path = file_path
self._append = append
self._read = read
self._update = update
self._closefd = closefd
self._file_epoch = file_object.epoch
contents = file_object.byte_contents
self._encoding = encoding
if encoding:
file_wrapper = FakeFileWrapper(file_object, file_path, update, read,
append, delete_on_close=False, filesystem=filesystem,
newline=None, binary=True, closefd=closefd)
codec_info = codecs.lookup(encoding)
self._io = codecs.StreamReaderWriter(file_wrapper, codec_info.streamreader,
codec_info.streamwriter)
else:
if sys.version_info >= (3, 0):
io_class = io.BytesIO if binary else io.StringIO
else:
io_class = cStringIO.StringIO
io_args = {} if binary else {'newline': newline}
if contents and not binary:
contents = contents.decode(encoding or locale.getpreferredencoding(False))
if contents and not update:
self._io = io_class(contents, **io_args)
else:
if truncate:
file_object.SetContents('')
else:
if must_exist:
raise IOError(errno.ENOENT, 'No such file or directory', file_path)
file_object = self.filesystem.CreateFile(
real_path, create_missing_dirs=False, apply_umask=True)
if stat.S_ISDIR(file_object.st_mode):
raise IOError(errno.EISDIR, 'Fake file object: is a directory', file_path)
# if you print obj.name, the argument to open() must be printed. Not the
# abspath, not the filename, but the actual argument.
file_object.opened_as = file_path
fakefile = FakeFileWrapper(file_object,
file_path,
update=need_write,
read=need_read,
append=append,
delete_on_close=self._delete_on_close,
filesystem=self.filesystem,
newline=newline,
binary=binary,
closefd=closefd,
encoding=encoding)
if filedes is not None:
fakefile.filedes = filedes
else:
fakefile.filedes = self.filesystem.AddOpenFile(fakefile)
return fakefile
self.fs = filesystem
def _refresh(self):
if self._stubs is not None:
self._stubs.smart_unset_all()
self._stubs = mox3_stubout.StubOutForTesting()
for name in self._fake_module_classes:
self._fake_modules[name] = self._fake_module_classes[name](self.fs)
self._fake_modules['path'] = self._fake_modules['os'].path
self.fake_open = fake_filesystem.FakeFileOpen(self.fs)
self._isStale = False
class WrittenS3File(fake_fs.FakeFileWrapper):
"""A File to be written to Amazon S3."""
def __init__(self, *args, **kwargs):
self.filesystem = kwargs['filesystem']
super(WrittenS3File, self).__init__(*args, **kwargs)
def upload(self):
"""Uploads the key to Amazon S3."""
self._io.seek(0)
self.filesystem.bucket[self.key] = self._io.read()
def close(self):
"""Close the file."""
# ignore closing a closed file
if not self._is_open():
return