1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
'''
@author: Marek Palatinus <marek@palatinus.cz>
@license: Public domain
DebugFS is a wrapper around filesystems to help developers
debug their work. I wrote this class mainly for debugging
TahoeLAFS and for fine tuning TahoeLAFS over Dokan with higher-level
aplications like Total Comander, Winamp etc. Did you know
that Total Commander need to open file before it delete them? :-)
I hope DebugFS can be helpful also for other filesystem developers,
especially for those who are trying to implement their first one (like me).
DebugFS prints to stdout (by default) all attempts to
filesystem interface, prints parameters and results.
Basic usage:
fs = DebugFS(OSFS('~'), identifier='OSFS@home', \
skip=('_lock', 'listdir', 'listdirinfo'))
print fs.listdir('.')
print fs.unsupportedfunction()
Error levels:
DEBUG: Print everything (asking for methods, calls, response, exception)
INFO: Print calls, responses, exception
ERROR: Print only exceptions
CRITICAL: Print only exceptions not derived from fs.errors.FSError
How to change error level:
import logging
logger = logging.getLogger('fs.debugfs')
logger.setLevel(logging.CRITICAL)
fs = DebugFS(OSFS('~')
print fs.listdir('.')
'''
import logging
from logging import DEBUG, INFO, ERROR, CRITICAL
import sys
import fs
from fs.errors import FSError
logger = fs.getLogger('fs.debugfs')
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())
class DebugFS(object):
def __init__(self, fs, identifier=None, skip=(), verbose=True):
'''
fs - Reference to object to debug
identifier - Custom string-like object will be added
to each log line as identifier.
skip - list of method names which DebugFS should not log
'''
self.__wrapped_fs = fs
self.__identifier = identifier
self.__skip = skip
self.__verbose = verbose
super(DebugFS, self).__init__()
def __log(self, level, message):
if self.__identifier:
logger.log(level, '(%s) %s' % (self.__identifier, message))
else:
logger.log(level, message)
def __parse_param(self, value):
if isinstance(value, basestring):
if len(value) > 60:
value = "%s ... (length %d)" % (repr(value[:60]), len(value))
else:
value = repr(value)
elif isinstance(value, list):
value = "%s (%d items)" % (repr(value[:3]), len(value))
elif isinstance(value, dict):
items = {}
for k, v in value.items()[:3]:
items[k] = v
value = "%s (%d items)" % (repr(items), len(value))
else:
value = repr(value)
return value
def __parse_args(self, *arguments, **kwargs):
args = [self.__parse_param(a) for a in arguments]
for k, v in kwargs.items():
args.append("%s=%s" % (k, self.__parse_param(v)))
args = ','.join(args)
if args: args = "(%s)" % args
return args
def __report(self, msg, key, value, *arguments, **kwargs):
if key in self.__skip: return
args = self.__parse_args(*arguments, **kwargs)
value = self.__parse_param(value)
self.__log(INFO, "%s %s%s -> %s" % (msg, str(key), args, value))
def __getattr__(self, key):
if key.startswith('__'):
# Internal calls, nothing interesting
return object.__getattribute__(self, key)
try:
attr = getattr(self.__wrapped_fs, key)
except AttributeError, e:
self.__log(DEBUG, "Asking for not implemented method %s" % key)
raise e
except Exception, e:
self.__log(CRITICAL, "Exception %s: %s" % \
(e.__class__.__name__, str(e)))
raise e
if not callable(attr):
if key not in self.__skip:
self.__report("Get attribute", key, attr)
return attr
def _method(*args, **kwargs):
try:
value = attr(*args, **kwargs)
self.__report("Call method", key, value, *args, **kwargs)
except FSError, e:
self.__log(ERROR, "Call method %s%s -> Exception %s: %s" % \
(key, self.__parse_args(*args, **kwargs), \
e.__class__.__name__, str(e)))
(exc_type,exc_inst,tb) = sys.exc_info()
raise e, None, tb
except Exception, e:
self.__log(CRITICAL,
"Call method %s%s -> Non-FS exception %s: %s" %\
(key, self.__parse_args(*args, **kwargs), \
e.__class__.__name__, str(e)))
(exc_type,exc_inst,tb) = sys.exc_info()
raise e, None, tb
return value
if self.__verbose:
if key not in self.__skip:
self.__log(DEBUG, "Asking for method %s" % key)
return _method
|
