Skip to content
Snippets Groups Projects
Commit d7bdcd1c authored by Piotr Maślanka's avatar Piotr Maślanka
Browse files

fix pylint and AutoflushFile and DevnullLikeObject

parent 359bb0bc
No related branches found
No related tags found
No related merge requests found
......@@ -12,12 +12,8 @@ commands:
- run:
command: |
pip install -r requirements.txt
pip install -U pytest-xdist pytest-cov pytest pytest-forked pluggy py opentracing pylint
pip install -U pytest-xdist pytest-cov pytest pytest-forked pluggy py opentracing
python setup.py install
pylint:
description: Run pylint
steps:
- run: pylint -j 4 satella || true
unit_test:
description: Run the unit tests
steps:
......@@ -29,7 +25,6 @@ jobs:
- checkout
- code-climate/install
- setup_python
- pylint
- unit_test
- run:
name: Collect results
......
engines:
plugins:
duplication:
enabled: true
config:
languages:
python:
fixme:
enabled: true
markdownlint:
enabled: true
pep8:
enabled: true
pylint:
enabled: true
exclude_paths:
- tests/**
- docs/**
......
[MASTER]
disable=
C0114, # missing-module-docstring
C0116 # missing-function-docstring
\ No newline at end of file
......@@ -5,10 +5,13 @@ The software
* added AutoflushFile.seek()
* added AutoflushFile.truncate()
* fixed behaviour with AutoflushFile.close()
* fixed behavior for AutoflushFile(..., 'wb')
* added CPUTimeManager.set_refresh_each()
* added satella.instrumentation.cpu_time.get_own_cpu_time()
Build process
-------------
* added pylint
* added pylint to CodeClimate
__version__ = '2.24.0a1'
__version__ = '2.24.0a2'
from __future__ import annotations
import codecs
import functools
import io
import os
import re
......@@ -10,7 +11,10 @@ __all__ = ['read_re_sub_and_write', 'find_files', 'split', 'read_in_file', 'writ
'write_out_file_if_different', 'make_noncolliding_name', 'try_unlink',
'DevNullFilelikeObject', 'read_lines', 'AutoflushFile']
from satella.coding.recast_exceptions import silence_excs
import warnings
from satella.coding import wraps
from satella.coding.recast_exceptions import silence_excs, reraise_as
from satella.coding.structures import Proxy
from satella.coding.typing import Predicate
......@@ -18,64 +22,91 @@ SEPARATORS = {'\\', '/'}
SEPARATORS.add(os.path.sep)
def value_error_on_closed_file(getter):
def outer(fun):
@functools.wraps(fun)
def inner(self, *args, **kwargs):
if getter(self):
raise ValueError('File closed')
return fun(self, *args, **kwargs)
return inner
return outer
closed_devnull = value_error_on_closed_file(lambda y: y.is_closed)
class DevNullFilelikeObject(io.FileIO):
"""
A /dev/null filelike object. For multiple uses.
:param binary: is this a binary file
:param ignore_typing_issues:
"""
__slots__ = 'is_closed', 'binary'
__slots__ = 'is_closed', 'binary', 'ignore_typing_issues'
def __init__(self, binary: bool = False):
def __init__(self, binary: bool = False, ignore_typing_issues: bool = False):
self.is_closed = False
self.binary = binary
self.ignore_typing_issues = ignore_typing_issues
@closed_devnull
def tell(self) -> int:
"""Return the current file offset"""
return 0
@closed_devnull
def truncate(self, __size: tp.Optional[int] = None) -> int:
"""Truncate file to __size starting bytes"""
return 0
@closed_devnull
def writable(self) -> bool:
"""Is this object writable"""
return True
@closed_devnull
def seek(self, v: int) -> int:
"""Seek to a particular file offset"""
return 0
@closed_devnull
def seekable(self) -> bool:
"""Is this file seekable?"""
return True
@closed_devnull
def read(self, byte_count: tp.Optional[int] = None) -> tp.Union[str, bytes]:
"""
:raises ValueError: this object has been closed
:raises io.UnsupportedOperation: since reading from this is forbidden
"""
if self.is_closed:
raise ValueError('Reading from closed /dev/null!')
return b'' if self.binary else ''
def write(self, x: tp.Union[str, bytes]) -> int:
@closed_devnull
def write(self, y: tp.Union[str, bytes]) -> int:
"""
Discard any amount of bytes
Discard any amount of bytes.
This will raise a RuntimeWarning warning upon writing invalid type.
:raises ValueError: this object has been closed
:return: length of written content
"""
if self.is_closed:
raise ValueError('Writing to closed /dev/null!')
return len(x)
if not self.ignore_typing_issues:
if isinstance(y, bytes) and not self.binary:
warnings.warn('Non binary data written to stream, but required binary', RuntimeWarning)
elif isinstance(y, str) and self.binary:
warnings.warn('Non binary data written to stream, but required binary', RuntimeWarning)
else:
warnings.warn('Non binary data written to stream, but required binary', RuntimeWarning)
return len(y)
@closed_devnull
def flush(self) -> None:
"""
:raises ValueError: when this object has been closed
"""
if self.is_closed:
raise ValueError('flush of closed file')
def close(self) -> None:
"""
......@@ -347,10 +378,26 @@ def write_out_file_if_different(path: str, data: tp.Union[bytes, str],
return True
is_closed_getter = value_error_on_closed_file(lambda y: y.__dict__['closed'])
def close_file_after(fun):
@wraps(fun)
def inner(self, *args, **kwargs):
try:
return fun(self, *args, **kwargs)
finally:
self._close_file()
return inner
class AutoflushFile(Proxy[io.FileIO]):
"""
A file that is supposed to be closed after each write command issued.
The file will be open only when there's an action to do on it called.
Best for appending so that other processes can read.
Use like:
......@@ -361,23 +408,34 @@ class AutoflushFile(Proxy[io.FileIO]):
>>> assert fin.read() == 'test'
"""
def __init__(self, file, mode='r', *con_args, **con_kwargs):
def __init__(self, file: str, mode: str, *con_args, **con_kwargs):
"""
:param file: path to the file
:param mode: mode to open the file with. Allowed values are w, wb, w+, a+, wb+, ab+, a, ab.
w+ and wb+ will truncate the file. Effective mode will be chosen by the class whatever just makes sense.
:raises ValueError: invalid mode chosen
"""
self.__dict__['con_kwargs'] = con_kwargs
self.__dict__['pointer'] = None
self.__dict__['closed'] = False
if mode in ('w+', 'wb+'):
if mode in ('w', 'w+', 'wb+', 'wb'):
fle = open(*(file, 'wb'))
fle.truncate(0)
fle.close()
mode = {'w': 'a', 'wb': 'ab', 'w+': 'a+', 'wb+': 'ab+', 'a': 'a', 'ab': 'ab'}[mode]
with reraise_as(KeyError, ValueError, f'Unsupported mode "{mode}"'):
mode = {'w': 'a', 'wb': 'ab', 'w+': 'a+', 'wb+': 'ab+', 'a': 'a', 'ab': 'ab'}[mode]
self.__dict__['con_args'] = (file, mode, *con_args)
fle = self._open_file()
super().__init__(fle)
self.__dict__['pointer'] = fle.tell()
self._close_file()
@is_closed_getter
@close_file_after
def seek(self, *args, **kwargs) -> int:
"""Seek to a provided position within the file"""
fle = self._open_file()
......@@ -385,6 +443,8 @@ class AutoflushFile(Proxy[io.FileIO]):
self.__dict__['pointer'] = fle.tell()
return v
@is_closed_getter
@close_file_after
def read(self, *args, **kwargs) -> tp.Union[str, bytes]:
"""
Read a file, returning the read-in data
......@@ -394,12 +454,13 @@ class AutoflushFile(Proxy[io.FileIO]):
file = self._open_file()
p = file.read(*args, **kwargs)
self.__dict__['pointer'] = file.tell()
self._close_file()
return p
def _get_file(self) -> tp.Optional[AutoflushFile]:
return self.__dict__.get('_Proxy__obj')
@is_closed_getter
@close_file_after
def readall(self) -> tp.Union[str, bytes]:
"""Read all contents into the file"""
file = self._open_file()
......@@ -422,13 +483,17 @@ class AutoflushFile(Proxy[io.FileIO]):
file.close()
self.__dict__['_Proxy__obj'] = None
@is_closed_getter
def close(self) -> None:
"""
Closes the file.
"""
self._open_file()
self._close_file()
self.__dict__['closed'] = True
@is_closed_getter
@close_file_after
def write(self, *args, **kwargs) -> int:
"""
Write a particular value to the file, close it afterwards.
......@@ -438,14 +503,14 @@ class AutoflushFile(Proxy[io.FileIO]):
file = self._open_file()
val = file.write(*args, **kwargs)
self.__dict__['pointer'] = file.tell()
self._close_file()
return val
@is_closed_getter
@close_file_after
def truncate(self, __size: tp.Optional[int] = None) -> int:
"""Truncate file to __size starting bytes"""
fle = self._open_file()
v = fle.truncate(__size)
self.__dict__['pointer'] = fle.tell()
self._close_file()
return v
......@@ -32,6 +32,11 @@ class TestFiles(unittest.TestCase):
af.close()
try_unlink('test3.txt')
af = AutoflushFile('test3.txt', 'w', encoding='utf-8')
os.unlink('test3.txt')
af.close()
self.assertRaises(ValueError, lambda: af.write('test'))
def test_read_lines(self):
lines = read_lines('LICENSE')
self.assertTrue(all(lines))
......@@ -52,6 +57,32 @@ class TestFiles(unittest.TestCase):
assert null.tell() == 0
assert null.seekable()
assert null.truncate(0) == 0
self.assertRaises(TypeError, lambda: null.write(b'ala'))
self.assertEqual(null.read(), '')
self.assertEqual(null.read(7), '')
null.flush()
null.close()
self.assertRaises(ValueError, lambda: null.write('test'))
self.assertRaises(ValueError, lambda: null.flush())
self.assertRaises(ValueError, lambda: null.read())
null.close()
def test_devnullfilelikeobject_2(self):
null = DevNullFilelikeObject(binary=True)
null.write(b'test')
null.write('ala')
null = DevNullFilelikeObject(ignore_typing_issues=True)
null.write(b'test')
null.write('test')
def test_devnullfilelikeobject_3(self):
null = DevNullFilelikeObject(binary=True)
self.assertEqual(null.write(b'ala'), 3)
assert null.seek(0) == 0
assert null.tell() == 0
assert null.seekable()
assert null.truncate(0) == 0
self.assertEqual(null.write(b'ala'), 3)
self.assertEqual(null.read(), '')
self.assertEqual(null.read(7), '')
......@@ -62,6 +93,7 @@ class TestFiles(unittest.TestCase):
self.assertRaises(ValueError, lambda: null.read())
null.close()
def try_directory(self):
os.system('mkdir test')
self.assertRaises(FileNotFoundError, lambda: read_in_file('test'))
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment