def open()

in luigi/mock.py [0:0]

• 45 lines of code
• 9 McCabe index (conditional complexity)

    def open(self, mode='r'):
        fn = self.path
        mock_target = self

        class Buffer(BytesIO):
            # Just to be able to do writing + reading from the same buffer

            _write_line = True

            def set_wrapper(self, wrapper):
                self.wrapper = wrapper

            def write(self, data):
                if mock_target._mirror_on_stderr:
                    if self._write_line:
                        sys.stderr.write(fn + ": ")
                    if bytes:
                        sys.stderr.write(data.decode('utf8'))
                    else:
                        sys.stderr.write(data)
                    if (data[-1]) == '\n':
                        self._write_line = True
                    else:
                        self._write_line = False
                super(Buffer, self).write(data)

            def close(self):
                if mode[0] == 'w':
                    try:
                        mock_target.wrapper.flush()
                    except AttributeError:
                        pass
                    mock_target.fs.get_all_data()[fn] = self.getvalue()
                super(Buffer, self).close()

            def __exit__(self, exc_type, exc_val, exc_tb):
                if not exc_type:
                    self.close()

            def __enter__(self):
                return self

            def readable(self):
                return mode[0] == 'r'

            def writeable(self):
                return mode[0] == 'w'

            def seekable(self):
                return False

        if mode[0] == 'w':
            wrapper = self.format.pipe_writer(Buffer())
            wrapper.set_wrapper(wrapper)
            return wrapper
        else:
            return self.format.pipe_reader(Buffer(self.fs.get_all_data()[fn]))