tyler.durden
/
powerline
mirror of https://github.com/powerline/powerline.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add fixes for flake8 to that stuff

This commit is contained in:
ZyX 2013-04-06 18:45:57 +04:00
parent 42ee82c1de
commit e68bae6409
4 changed files with 35 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
powerline/lib/file_watcher.py
View File

@ -12,8 +12,8 @@ from threading import RLock
from powerline.lib.monotonic import monotonic
from powerline.lib.inotify import INotify, INotifyError
class INotifyWatch(INotify):
class INotifyWatch(INotify):
is_stat_based = False
def __init__(self, expire_time=10):

17
powerline/lib/inotify.py
View File

@ -4,13 +4,18 @@ from __future__ import unicode_literals, absolute_import
__copyright__ = '2013, Kovid Goyal <kovid at kovidgoyal.net>'
__docformat__ = 'restructuredtext en'
import sys, os, errno
import sys
import os
import errno
class INotifyError(Exception):
pass
_inotify = None
def load_inotify():
''' Initialize the inotify library '''
global _inotify
@ -53,8 +58,8 @@ def load_inotify():
_inotify = (init1, add_watch, rm_watch, read)
return _inotify
class INotify(object):
class INotify(object):
# See <sys/inotify.h> for the flags defined below
# Supported events suitable for MASK parameter of INOTIFY_ADD_WATCH.
@ -98,7 +103,8 @@ class INotify(object):
NONBLOCK = 0x800
def __init__(self, cloexec=True, nonblock=True):
import ctypes, struct
import ctypes
import struct
self._init1, self._add_watch, self._rm_watch, self._read = load_inotify()
flags = 0
if cloexec:
@ -164,12 +170,9 @@ class INotify(object):
pos += self.hdr.size
name = None
if get_name:
name = raw[pos:pos+name_len].rstrip(b'\0').decode(self.fenc)
name = raw[pos:pos + name_len].rstrip(b'\0').decode(self.fenc)
pos += name_len
self.process_event(wd, mask, cookie, name)
def process_event(self, *args):
raise NotImplementedError()

34
powerline/lib/tree_watcher.py
View File

@ -1,26 +1,28 @@
#!/usr/bin/env python
# vim:fileencoding=UTF-8:noet
from __future__ import (unicode_literals, absolute_import, print_function)
__copyright__ = '2013, Kovid Goyal <kovid at kovidgoyal.net>'
__docformat__ = 'restructuredtext en'
import sys, os, errno
import sys
import os
import errno
from time import sleep
from powerline.lib.monotonic import monotonic
from powerline.lib.inotify import INotify, INotifyError
class NoSuchDir(ValueError):
pass
class DirTooLarge(ValueError):
class DirTooLarge(ValueError):
def __init__(self, bdir):
ValueError.__init__(self, 'The directory %s is too large to monitor. Try increasing the value in /proc/sys/fs/inotify/max_user_watches'%bdir)
ValueError.__init__(self, 'The directory {0} is too large to monitor. Try increasing the value in /proc/sys/fs/inotify/max_user_watches'.format(bdir))
class INotifyTreeWatcher(INotify):
is_dummy = False
def __init__(self, basedir):
@ -49,13 +51,13 @@ class INotifyTreeWatcher(INotify):
# The entry could have been deleted between listdir() and
# add_watch().
if top_level:
raise NoSuchDir('The dir %s does not exist'%base)
raise NoSuchDir('The dir {0} does not exist'.format(base))
return
if e.errno == errno.EACCES:
# We silently ignore entries for which we dont have permission,
# unless they are the top level dir
if top_level:
raise NoSuchDir('You do not have permission to monitor %s'%base)
raise NoSuchDir('You do not have permission to monitor {0}'.format(base))
return
raise
else:
@ -67,20 +69,22 @@ class INotifyTreeWatcher(INotify):
# The dir was deleted/replaced between the add_watch()
# and listdir()
if top_level:
raise NoSuchDir('The dir %s does not exist'%base)
raise NoSuchDir('The dir {0} does not exist'.format(base))
return
raise
for x in files:
self.add_watches(os.path.join(base, x), top_level=False)
elif top_level:
# The top level dir is a file, not good.
raise NoSuchDir('The dir %s does not exist'%base)
raise NoSuchDir('The dir {0} does not exist'.format(base))
def add_watch(self, path):
import ctypes
bpath = path if isinstance(path, bytes) else path.encode(self.fenc)
wd = self._add_watch(self._inotify_fd, ctypes.c_char_p(bpath),
self.DONT_FOLLOW | self.ONLYDIR | # Ignore symlinks and watch only directories
# Ignore symlinks and watch only directories
self.DONT_FOLLOW | self.ONLYDIR |
self.MODIFY | self.CREATE | self.DELETE |
self.MOVE_SELF | self.MOVED_FROM | self.MOVED_TO |
self.ATTRIB | self.MOVE_SELF | self.DELETE_SELF)
@ -88,7 +92,7 @@ class INotifyTreeWatcher(INotify):
eno = ctypes.get_errno()
if eno == errno.ENOTDIR:
return False
raise OSError(eno, 'Failed to add watch for: %s: %s'%(path, self.os.strerror(eno)))
raise OSError(eno, 'Failed to add watch for: {0}: {1}'.format(path, self.os.strerror(eno)))
self.watched_dirs[path] = wd
self.watched_rmap[wd] = path
return True
@ -122,8 +126,8 @@ class INotifyTreeWatcher(INotify):
self.modified = False
return ret
class DummyTreeWatcher(object):
class DummyTreeWatcher(object):
is_dummy = True
def __init__(self, basedir):
@ -132,8 +136,8 @@ class DummyTreeWatcher(object):
def __call__(self):
return False
class TreeWatcher(object):
class TreeWatcher(object):
def __init__(self, expire_time=10):
self.watches = {}
self.last_query_times = {}
@ -145,7 +149,7 @@ class TreeWatcher(object):
w = INotifyTreeWatcher(path)
except (INotifyError, DirTooLarge) as e:
if logger is not None:
logger.warn('Failed to watch path: %s with error: %s'%(path, e))
logger.warn('Failed to watch path: {0} with error: {1}'.format(path, e))
w = DummyTreeWatcher(path)
self.watches[path] = w
return w
@ -193,5 +197,3 @@ if __name__ == '__main__':
sleep(1)
except KeyboardInterrupt:
raise SystemExit(0)

12
tests/test_lib.py
View File

@ -38,8 +38,8 @@ class TestLib(TestCase):
self.assertEqual(humanize_bytes(1000000000, si_prefix=True), '1.00 GB')
self.assertEqual(humanize_bytes(1000000000, si_prefix=False), '953.7 MiB')
class TestFilesystemWatchers(TestCase):
class TestFilesystemWatchers(TestCase):
def do_test_for_change(self, watcher, path):
import time
st = time.time()
@ -47,7 +47,7 @@ class TestFilesystemWatchers(TestCase):
if watcher(path):
return
time.sleep(0.1)
self.fail('The change to %s was not detected'%path)
self.fail('The change to {0} was not detected'.format(path))
def test_file_watcher(self):
from powerline.lib.file_watcher import create_file_watcher
@ -107,9 +107,9 @@ class TestFilesystemWatchers(TestCase):
changed()
os.mkdir(subdir)
changed()
os.rename(subdir, subdir+'1')
os.rename(subdir, subdir + '1')
changed()
shutil.rmtree(subdir+'1')
shutil.rmtree(subdir + '1')
changed()
os.mkdir(subdir)
f = os.path.join(subdir, 'f')
@ -118,7 +118,7 @@ class TestFilesystemWatchers(TestCase):
with open(f, 'a') as s:
s.write(' ')
changed()
os.rename(f, f+'1')
os.rename(f, f + '1')
changed()
use_mercurial = use_bzr = sys.version_info < (3, 0)
@ -192,6 +192,7 @@ HG_REPO = 'hg_repo' + os.environ.get('PYTHON', '')
BZR_REPO = 'bzr_repo' + os.environ.get('PYTHON', '')
INOTIFY_DIR = 'inotify' + os.environ.get('PYTHON', '')
def setUpModule():
global old_cwd
global old_HGRCPATH
@ -217,7 +218,6 @@ def setUpModule():
os.mkdir(INOTIFY_DIR)
def tearDownModule():
global old_cwd
global old_HGRCPATH