forked from nuxeo/FunkLoad
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathutils.py
More file actions
305 lines (260 loc) · 8.79 KB
/
Copy pathutils.py
File metadata and controls
305 lines (260 loc) · 8.79 KB
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
# (C) Copyright 2005 Nuxeo SAS <http://nuxeo.com>
# Author: [email protected]
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
# 02111-1307, USA.
#
"""FunkLoad common utils.
$Id: utils.py 24649 2005-08-29 14:20:19Z bdelbosc $
"""
import os
import sys
import time
import logging
from time import sleep
from socket import error as SocketError
from xmlrpclib import ServerProxy
MIN_SLEEPTIME = 0.005 # minimum sleep time to let
# python threads working properly
def thread_sleep(seconds=0):
"""Sleep seconds.
Insure that seconds is at least MIN_SLEEPTIME to let
threads working properly."""
#if seconds:
# trace('sleep %s' % seconds)
sleep(max(abs(seconds), MIN_SLEEPTIME))
# ------------------------------------------------------------
# semaphores
#
g_recording = False
g_running = False
def recording():
"""A semaphore to tell the running threads when to begin recording."""
global g_recording
return g_recording
def set_recording_flag(value):
"""Enable recording."""
global g_recording
g_recording = value
def running():
"""A semaphore to tell the running threads that it should continue running
ftest."""
global g_running
return g_running
def set_running_flag(value):
"""Set running mode on."""
global g_running
g_running = value
# ------------------------------------------------------------
# daemon
#
# See the Chad J. Schroeder example for a full explanation
# this version does not chdir to '/' to keep relative path
def create_daemon():
"""Detach a process from the controlling terminal and run it in the
background as a daemon.
"""
try:
pid = os.fork()
except OSError, msg:
raise Exception, "%s [%d]" % (msg.strerror, msg.errno)
if (pid == 0):
os.setsid()
try:
pid = os.fork()
except OSError, msg:
raise Exception, "%s [%d]" % (msg.strerror, msg.errno)
if (pid == 0):
os.umask(0)
else:
os._exit(0)
else:
sleep(.5)
os._exit(0)
import resource
maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1]
if (maxfd == resource.RLIM_INFINITY):
maxfd = 1024
for fd in range(0, maxfd):
try:
os.close(fd)
except OSError:
pass
os.open('/dev/null', os.O_RDWR)
os.dup2(0, 1)
os.dup2(0, 2)
return(0)
# ------------------------------------------------------------
# meta method name encodage
#
MMN_SEP = ':' # meta method name separator
def mmn_is_bench(meta_method_name):
"""Is it a meta method name ?."""
return meta_method_name.count(MMN_SEP) and True or False
def mmn_encode(method_name, cycle, cvus, thread_id):
"""Encode a extra information into a method_name."""
return MMN_SEP.join((method_name, str(cycle), str(cvus), str(thread_id)))
def mmn_decode(meta_method_name):
"""Decode a meta method name."""
if mmn_is_bench(meta_method_name):
method_name, cycle, cvus, thread_id = meta_method_name.split(MMN_SEP)
return (method_name, int(cycle), int(cvus), int(thread_id))
else:
return (meta_method_name, 1, 0, 1)
# ------------------------------------------------------------
# logging
#
def get_default_logger(log_to, log_path=None, level=logging.DEBUG,
name='FunkLoad'):
"""Get a logger."""
logger = logging.getLogger(name)
if logger.handlers:
# already setup
return logger
if log_to.count("console"):
hdlr = logging.StreamHandler()
logger.addHandler(hdlr)
if log_to.count("file") and log_path:
formatter = logging.Formatter(
'%(asctime)s %(levelname)s %(message)s')
hdlr = logging.FileHandler(log_path)
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
if log_to.count("xml") and log_path:
if os.access(log_path, os.F_OK):
os.rename(log_path, log_path + '.bak-' + str(int(time.time())))
hdlr = logging.FileHandler(log_path)
logger.addHandler(hdlr)
logger.setLevel(level)
return logger
def close_logger(name):
"""Close the logger."""
logger = logging.getLogger(name)
for hdlr in logger.handlers:
logger.removeHandler(hdlr)
def trace(message):
"""Simple print to stdout
Not thread safe."""
sys.stdout.write(message)
sys.stdout.flush()
# ------------------------------------------------------------
# xmlrpc
#
def xmlrpc_get_credential(host, port, group=None):
"""Get credential thru xmlrpc credential_server."""
url = "http://%s:%s" % (host, port)
server = ServerProxy(url)
try:
return server.getCredential(group)
except SocketError:
raise SocketError(
'No Credential server reachable at %s, use fl-credential-ctl '
'to start the credential server.' % url)
def xmlrpc_list_groups(host, port):
"""Get list of groups thru xmlrpc credential_server."""
url = "http://%s:%s" % (host, port)
server = ServerProxy(url)
try:
return server.listGroups()
except SocketError:
raise SocketError(
'No Credential server reachable at %s, use fl-credential-ctl '
'to start the credential server.' % url)
def xmlrpc_list_credentials(host, port, group=None):
"""Get list of users thru xmlrpc credential_server."""
url = "http://%s:%s" % (host, port)
server = ServerProxy(url)
try:
return server.listCredentials(group)
except SocketError:
raise SocketError(
'No Credential server reachable at %s, use fl-credential-ctl '
'to start the credential server.' % url)
# ------------------------------------------------------------
# misc
#
def get_version():
"""Retrun the FunkLoad package version."""
from pkg_resources import get_distribution
return get_distribution('funkload').version
_COLOR = {'green': "\x1b[32;01m",
'red': "\x1b[31;01m",
'reset': "\x1b[0m"
}
def red_str(text):
"""Return red text."""
global _COLOR
return _COLOR['red'] + text + _COLOR['reset']
def green_str(text):
"""Return green text."""
global _COLOR
return _COLOR['green'] + text + _COLOR['reset']
def is_html(text):
"""Simple check that return True if the text is an html page."""
if '<html' in text[:300].lower():
return True
return False
# credits goes to Subways and Django folks
class BaseFilter(object):
"""Base filter."""
def __ror__(self, other):
return other # pass-thru
def __call__(self, other):
return other | self
class truncate(BaseFilter):
"""Middle truncate string up to length."""
def __init__(self, length=40, extra='...'):
self.length = length
self.extra = extra
def __ror__(self, other):
if len(other) > self.length:
mid_size = (self.length - 3) / 2
other = other[:mid_size] + self.extra + other[-mid_size:]
return other
def is_valid_html(html=None, file_path=None, accept_warning=False):
"""Ask tidy if the html is valid.
Return a tuple (status, errors)
"""
if not file_path:
fd, file_path = mkstemp(prefix='fl-tidy', suffix='.html')
os.write(fd, html)
os.close(fd)
tidy_cmd = 'tidy -errors %s' % file_path
ret, output = getstatusoutput(tidy_cmd)
status = False
if ret == 0:
status = True
elif ret == 256:
# got warnings
if accept_warning:
status = True
elif ret > 512:
if 'command not found' in output:
raise RuntimeError('tidy command not found, please install tidy.')
raise RuntimeError('Executing [%s] return: %s ouput: %s' %
(tidy_cmd, ret, output))
return status, output
class Data:
'''Simple "sentinel" class that lets us identify user data
and content type in POST'''
def __init__(self, content_type, data):
self.content_type = content_type
self.data = data
def __cmp__(self, other):
diff = cmp(self.content_type, other.content_type)
if not diff:
diff = cmp(self.data, other.data)
return diff
def __repr__(self):
return "[User data " + self.content_type + "]"