Coverage for src/lcdoc/call_flows/call_flow_logging.py : 28.37%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import os, json, sys, time lp|features/lp/python/call_flow_logging/index.mdpytest
2from lcdoc.tools import exists, project, write_file, read_file, to_list lp|features/lp/python/call_flow_logging/index.mdpytest
3import inspect, shutil lp|features/lp/python/call_flow_logging/index.mdpytest
4from functools import partial, wraps lp|features/lp/python/call_flow_logging/index.mdpytest
5from lcdoc.call_flows.call_flow_charting import make_call_flow_chart lp|features/lp/python/call_flow_logging/index.mdpytest
7from lcdoc.call_flows.markdown import Mkdocs, deindent, to_min_header_level lp|features/lp/python/call_flow_logging/index.mdpytest
9from pprint import pformat lp|features/lp/python/call_flow_logging/index.mdpytest
11from inflection import humanize lp|features/lp/python/call_flow_logging/index.mdpytest
13from importlib import import_module lp|features/lp/python/call_flow_logging/index.mdpytest
14from .auto_docs import mod_doc lp|features/lp/python/call_flow_logging/index.mdpytest
17# ------------------------------------------------------------------------------ tools
18now = time.time lp|features/lp/python/call_flow_logging/index.mdpytest
21class ILS: lp|features/lp/python/call_flow_logging/index.mdpytest
22 """
23 Call Flow Logger State
25 Normally populated/cleared at start and end of a pytest.
26 """
28 traced = set() # all traced code objects lp|features/lp/python/call_flow_logging/index.mdpytest
29 max_trace = 100 lp|features/lp/python/call_flow_logging/index.mdpytest
30 call_chain = _ = [] lp|features/lp/python/call_flow_logging/index.mdpytest
31 counter = 0 lp|features/lp/python/call_flow_logging/index.mdpytest
32 calls_by_frame = {} lp|features/lp/python/call_flow_logging/index.mdpytest
33 parents = {} lp|features/lp/python/call_flow_logging/index.mdpytest
34 parents_by_code = {} lp|features/lp/python/call_flow_logging/index.mdpytest
35 doc_msgs = [] lp|features/lp/python/call_flow_logging/index.mdpytest
36 # sources = {}
38 # wrapped = {}
40 def clear(): lp|features/lp/python/call_flow_logging/index.mdpytest
41 ILS.max_trace = 100
42 ILS.call_chain.clear()
43 ILS.calls_by_frame.clear()
44 ILS.counter = 0
45 ILS.traced.clear()
46 ILS.parents.clear()
47 ILS.parents_by_code.clear()
48 ILS.doc_msgs.clear()
51def call_flow_note(msg, **kw): lp|features/lp/python/call_flow_logging/index.mdpytest
52 msg = json.dumps([msg, kw], indent=2) if kw else msg
53 s = {
54 'input': None,
55 'fn_mod': 'note',
56 'output': None,
57 'thread_req': thread(),
58 'name': msg,
59 't0': now(),
60 }
61 ILS.call_chain.append(s)
64def unwrap_partial(f): lp|features/lp/python/call_flow_logging/index.mdpytest
65 while hasattr(f, 'func'): 65 ↛ 66line 65 didn't jump to line 66, because the condition on line 65 was never truelp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
66 f = f.func
67 return f lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
70from threading import current_thread lp|features/lp/python/call_flow_logging/index.mdpytest
72thread = lambda: current_thread().name.replace('ummy-', '').replace('hread-', '') 72 ↛ exitline 72 didn't run the lambda on line 72lp|features/lp/python/call_flow_logging/index.mdpytest
75class SetTrace: lp|features/lp/python/call_flow_logging/index.mdpytest
76 """
77 sys.settrace induced callbacks are passed into this
78 (when the called function is watched, i.e. added to ILS.traced)
79 """
81 def request(frame): lp|features/lp/python/call_flow_logging/index.mdpytest
82 if ILS.counter > ILS.max_trace:
83 call_flow_note('Reached max traced calls count', max_trace=ILS.max_trace)
84 sys.settrace(None)
85 return
86 ILS.counter += 1
88 # need to do (and screw time) this while tracing
89 co = frame.f_code
90 # frm = SetTrace.get_traced_sender(frame)
92 # if ILS.counter == 5:
93 # f = frame.f_back
94 # while f:
95 # print('sender', f, inp)
96 # f = f.f_back
98 t0 = now()
99 p = ILS.parents_by_code.get(co)
100 if p:
101 if len(p) == 1:
102 n = p[0]
103 else:
104 n = '%s (%s)' % (p[-1], '.'.join(p[:-1]))
105 else:
106 n = co.co_name
107 inp = dumps(frame.f_locals)
108 # inp = ''
109 s = {
110 'thread_req': thread(),
111 'counter': ILS.counter,
112 'input': inp,
113 'frame': frame,
114 'parents': ILS.parents_by_code.get(co),
115 'name': n,
116 'line': frame.f_lineno,
117 't0': now(),
118 'dt': -1,
119 'output': 'n.a.',
120 'fn_mod': co.co_filename,
121 }
122 ILS.call_chain.append(s)
123 ILS.calls_by_frame[frame] = s
125 def response(frame, arg): lp|features/lp/python/call_flow_logging/index.mdpytest
126 try:
127 s = ILS.calls_by_frame[frame]
128 except Exception as ex:
129 return
130 s['dt'] = now() - s['t0']
131 s['output'] = dumps(arg)
132 s['thread_resp'] = thread()
133 ILS.call_chain.append(s)
135 def tracer(frame, event, arg, counter=0): lp|features/lp/python/call_flow_logging/index.mdpytest
136 """The settrace function. You can't pdb here!"""
137 if not frame.f_code in ILS.traced:
138 return
139 if event == 'call':
140 SetTrace.request(frame)
141 # getting a callback on traced function exit:
142 return SetTrace.tracer
143 elif event == 'return':
144 SetTrace.response(frame, arg)
147is_parent = lambda o: inspect.isclass(o) or inspect.ismodule(o) lp|features/lp/python/call_flow_logging/index.mdpytestpytest|tests.test_cfl.test_one
148is_func = inspect.isfunction or inspect.ismethod lp|features/lp/python/call_flow_logging/index.mdpytest
151def trace_object(
152 obj,
153 pth=None,
154 containing_mod=None,
155 filters=(
156 lambda k: not k.startswith('_'),
157 lambda k, v: inspect.isclass(v) or is_func(v),
158 ),
159 blacklist=(),
160):
161 """recursive
163 partials:
164 For now we ignore them, treat them like attributes.
165 The functions they wrap, when contained by a traced object will be documented,
166 with all args.
167 In order to document partials we would have to wrap them into new functions, set into the parent.
169 filters: for keys and keys + values
170 """
171 if is_parent(obj): lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
172 if not pth: lp|features/lp/python/call_flow_logging/index.md
173 pth = (obj.__name__,) lp|features/lp/python/call_flow_logging/index.md
174 ILS.parents[pth] = obj lp|features/lp/python/call_flow_logging/index.md
175 containing_mod = pth[0] if inspect.ismodule(obj) else obj.__module__ lp|features/lp/python/call_flow_logging/index.md
176 for k in filter(filters[0], dir(obj)): lp|features/lp/python/call_flow_logging/index.md
177 v = getattr(obj, k) lp|features/lp/python/call_flow_logging/index.md
178 if filters[1](k, v): lp|features/lp/python/call_flow_logging/index.md
179 pth1 = pth + (k,) lp|features/lp/python/call_flow_logging/index.md
180 ILS.parents[pth1] = v lp|features/lp/python/call_flow_logging/index.md
181 trace_object(v, pth1, containing_mod, filters, blacklist) lp|features/lp/python/call_flow_logging/index.md
182 elif is_func(obj): 182 ↛ exitline 182 didn't return from function 'trace_object', because the condition on line 182 was never falselp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
183 spth = str(pth) + str(obj) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
184 if any([b for b in blacklist if b in spth]): 184 ↛ 186line 184 didn't jump to line 186, because the condition on line 184 was never truelp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
185 # TODO: Just return in settrace and write the info into the spec of the call:
186 print('blacklisted for tracing', str(obj))
187 return
189 if containing_mod and not obj.__module__.startswith(containing_mod): 189 ↛ 190line 189 didn't jump to line 190, because the condition on line 189 was never truelp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
190 return
191 co = obj.__code__ lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
192 ILS.parents_by_code[co] = pth lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
193 ILS.traced.add(co) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
194 # print(obj)
197def trace_func(traced_func, settrace=True): lp|features/lp/python/call_flow_logging/index.mdpytest
198 """trace a function w/o context"""
199 func = unwrap_partial(traced_func) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
200 ILS.traced.add(func.__code__) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
201 # collector = collector or ILS.call_chain.append
202 if settrace: 202 ↛ exitline 202 didn't return from function 'trace_func', because the condition on line 202 was never falselp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
203 sys.settrace(SetTrace.tracer) # partial(tracer, collector=collector)) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
206# we already dumps formatted, so that the js does not need to parse/stringify:
207def dumps(s): lp|features/lp/python/call_flow_logging/index.mdpytest
208 try:
209 return json.dumps(s, default=str, indent=4, sort_keys=True)
210 except Exception as ex:
211 # sorting sometimes not works (e.g. 2 classes as keys)
212 return pformat(s)
215fmts = {'mkdocs': Mkdocs} lp|features/lp/python/call_flow_logging/index.mdpytest
218def autodoc_dir(mod, _d_dest): lp|features/lp/python/call_flow_logging/index.mdpytest
219 if _d_dest != 'auto':
220 return _d_dest
222 if not inspect.ismodule(mod):
223 breakpoint() # FIXME BREAKPOINT
224 raise
225 modn, fn = mod.__name__, mod.__file__
227 r = project.root()
228 if fn.startswith(r):
229 # fn '/home/gk/repos/lc-python/tests/operators/test_build.py' -> tests/operators:
230 d_mod = fn.rsplit(r, 1)[1][1:].rsplit('/', 1)[0]
231 else:
232 d_mod = modn.replace('.', '/')
233 _d_dest = project.root() + '/build/autodocs/' + d_mod
234 return _d_dest
237flag_defaults = {} lp|features/lp/python/call_flow_logging/index.mdpytest
240def set_flags(flags, unset=False): lp|features/lp/python/call_flow_logging/index.mdpytest
241 if not unset:
242 try:
243 FLG.log_level # is always imported
244 except UnparsedFlagAccessError:
245 from devapp.app import init_app_parse_flags
247 init_app_parse_flags('pytest')
248 for k in dir(FLG):
249 flag_defaults[k] = getattr(FLG, k)
250 M = flags
251 else:
252 M = flag_defaults
254 return [setattr(FLG, k, v) for k, v in M.items()]
257import os lp|features/lp/python/call_flow_logging/index.mdpytest
258from functools import partial lp|features/lp/python/call_flow_logging/index.mdpytest
261def pytest_plot_dest(dest): lp|features/lp/python/call_flow_logging/index.mdpytest
262 cur_test = lambda: os.environ['PYTEST_CURRENT_TEST']
263 # if os.path.isfile(dest):
264 fn_t = cur_test().split('.py::', 1)[1].replace('::', '.').replace(' (call)', '')
265 dest = dest.rsplit('.', 1)[0] + '/' + fn_t
267 # fn_t = cur_test().rsplit('/', 1)[-1].replace(' (call)', '')
268 # '%s/%s._plot_tag_.graph_easy.src' % (dest, fn_t)
269 return dest + '/_plot_tag_.graph_easy.src'
272def plot_build_flow(dest): lp|features/lp/python/call_flow_logging/index.mdpytest
273 f = {
274 'plot_mode': 'src',
275 'plot_before_build': True,
276 'plot_write_flow_json': 'prebuild',
277 'plot_after_build': True,
278 'plot_destination': partial(pytest_plot_dest, dest=dest),
279 }
280 return f
283def add_doc_msg(msg, code=None, **kw): lp|features/lp/python/call_flow_logging/index.mdpytest
284 if code:
285 T = fmts[kw.pop('fmt', 'mkdocs')]
286 c = T.code(kw.pop('lang', 'js'), code)
287 msg = (T.closed_admon if kw.pop('closed', 0) else T.admon)(msg, c, 'info')
288 kw = None
289 ILS.doc_msgs.append([msg, kw])
292def document(
293 trace,
294 max_trace=100,
295 fmt='mkdocs',
296 blacklist=(),
297 call_seq_closed=True,
298 flags=None,
299 dest=None,
300):
301 """
302 Decorating call flow triggering functions (usually pytest functions) with this
304 dest:
305 - if not a file equal to d_dest below
306 - if file: d_dest is dir named file w/o ext. e.g. /foo/bar.md -> /foo/bar/
309 """
311 def check_tracable(t): lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
312 if ( 312 ↛ 317line 312 didn't jump to line 317
313 not isinstance(t, type)
314 and not callable(t)
315 and not getattr(type(t), '__name__') == 'module'
316 ):
317 raise Exception('Can only trace modules, classes and callables, got: %s' % t)
319 [check_tracable(t) for t in to_list(trace)] lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
320 if not dest: 320 ↛ 325line 320 didn't jump to line 325, because the condition on line 320 was never truelp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
321 # dest is empty if env var make_autodocs is not set.
322 # Then we do nothing.
323 # noop if env var not set, we don't want to trace at every pytest run, that
324 # would distract the developer when writing tests:
325 def decorator(func):
326 @wraps(func)
327 def noop_wrapper(*args, **kwargs):
328 return func(*args, **kwargs)
330 return noop_wrapper
332 return decorator
334 ILS.max_trace = max_trace # limit traced calls lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
336 # this is the documentation tracing decorator:
337 def use_case_deco(use_case, trace=trace): lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
338 def use_case_(*args, _dest=dest, _fmt=fmt, **kwargs): lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
339 n_mod = use_case.__module__ lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
340 if os.path.isfile(_dest): 340 ↛ 345line 340 didn't jump to line 345, because the condition on line 340 was never falselp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
341 fn_md_into = _dest lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
342 _d_dest = _dest.rsplit('.', 1)[0] lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
343 os.makedirs(_d_dest, exist_ok=True) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
344 else:
345 raise NotImplementedError(
346 'derive autodocs dir when no mod was documented'
347 )
348 # _d_dest = autodoc_dir(n_mod, dest)
349 # fn_md_into = (d_usecase(_d_dest, use_case) + '.md',)
351 n = n_func(use_case) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
352 fn_chart = n lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
353 blackl = to_list(blacklist) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
354 if max_trace and trace: 354 ↛ 357line 354 didn't jump to line 357, because the condition on line 354 was never falselp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
355 [trace_object(t, blacklist=blackl) for t in to_list(trace)] lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
356 trace_func(use_case) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
357 flg = {} lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
358 if flags: 358 ↛ 359line 358 didn't jump to line 359, because the condition on line 358 was never truelp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
359 for k, v in flags.items():
360 flg[k] = v() if callable(v) else v
361 set_flags(flg)
362 try: lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
363 throw = None lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
364 # set_flags(flags)
365 value = use_case(*args, **kwargs) # <-------- the actual original call lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
366 except SystemExit as ex:
367 throw = ex
369 set_flags(flags, unset=True) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
370 doc_msgs = list(ILS.doc_msgs) # will be cleared in next call: lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
372 write.flow_chart(_d_dest, use_case, clear=True, with_chart=fn_chart) lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
373 T = fmts[fmt]
374 src = deindent(inspect.getsource(use_case)).splitlines()
375 while True:
376 # remove decorators:
377 if src[0].startswith('@'):
378 src.pop(0)
379 else:
380 break
381 src = '\n'.join(src)
383 # if the test is within a class we write the class as header and add the
384 # funcs within:
385 min_header_level, doc = 4, ''
386 # n .e.g. test_01_sock_comm
387 if '.' in n:
388 min_header_level = 5
389 section, n = n.rsplit('.', 1)
390 have = written_sect_headers.setdefault(n_mod, {})
391 if not section in have:
392 doc += '\n\n#### ' + section
393 have[section] = True
395 doc += '\n\n'
396 n_pretty = n.split('_', 1)[1] if n.startswith('test') else n
397 n_pretty = humanize(n_pretty)
399 # set a jump mark for the ops reference page (doc pp):
400 _ = pytest_plot_dest(_dest).split('/autodocs/', 1)[1].rsplit('/', 1)[0]
401 doc += '\n<span id="%s"></span>\n' % _
403 ud = deindent(strip_no_ad(use_case.__doc__ or ''))
404 _ = to_min_header_level
405 doc += _(min_header_level, deindent('\n\n#### %s\n%s' % (n_pretty, ud)))
407 n = use_case.__qualname__
408 f = flg.get('plot_destination')
409 if f:
410 doc += add_flow_json_and_graph_easy_links(f, T)
412 _ = T.closed_admon
413 doc += _(n + ' source code', T.code('python', strip_no_ad(src)))
415 # call flow plots only when we did trace anythning:
416 if max_trace and trace:
417 # m = {'fn': fn_chart}
418 # svg_ids[0] += 1
419 # m['id'] = svg_ids[0]
420 # v = '[s.svg)](?src=%(fn)s&sequence_details=true)' % m
421 # v = 's.svg)' % m # ](?src=%(fn)s&sequence_details=true)' % m
422 # v = (
423 # '''
424 # <span class="std_svg" id="std_svg_%(id)s">
425 # <img src="../%(fn)s.svg"></img>
426 # </span>'''
427 # % m
428 # )
429 #s.svg)' % m # ](?src=%(fn)s&sequence_details=true)' % m
430 tr = [name(t) for t in to_list(trace)]
431 tr = [t for t in tr if t]
432 tr = ', '.join(tr)
434 _ = call_seq_closed
435 adm = T.closed_admon if _ else T.clsabl_admon
436 # they are rendered and inserted at doc pre_process, i.e. later:
437 fn = '%s/%s/call_flow.svg' % (fn_md_into.rsplit('.', 1)[0], fn_chart)
438 fn = fn.rsplit('/autodocs/', 1)[1]
439 svg_placeholder = '[svg_placeholder:%s]' % fn
441 # svg = (
442 # read_file('%s/%s.svg' % (os.path.dirname(fn_md_into), fn_chart))
443 # .split('>', 1)[1]
444 # .strip()
445 # .split('<!--MD5', 1)[0]
446 # )
447 # if not svg.endswith('</svg>'):
448 # svg += '</g></svg>'
449 # id = '<svg id="%s" class="call_flow_chart" ' % fn_chart
450 # svg = svg.replace('<svg ', id)
451 doc += adm('Call Sequence `%s` (%s)' % (n, tr), svg_placeholder, 'info')
453 # had doc_msgs been produced during running the test fuction? Then append:
454 for d in doc_msgs:
455 if not d[1]:
456 doc += '\n\n%s\n\n' % d[0]
457 else:
458 doc += '\n%s%s' % (d[0], T.code('js', json.dumps(d[1], indent=4)))
460 # append our stuffs:
461 s = read_file(fn_md_into, dflt='')
462 if s:
463 doc = s + '\n\n' + doc
464 write_file(fn_md_into, doc)
466 # had the function been raising? Then throw it now:
467 if throw:
468 raise throw
470 return value
472 return use_case_ lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
474 return use_case_deco lp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one
477# svg_ids = [0]
480def strip_no_ad(s, sep='# /--'): lp|features/lp/python/call_flow_logging/index.mdpytest
481 if not sep in s:
482 return s
483 r, add, lines = [], True, s.splitlines()
484 while lines:
485 l = lines.pop(0)
486 if l.strip() == sep:
487 add = not add
488 continue
489 if not add:
490 continue
491 r.append(l)
492 return ('\n'.join(r)).strip()
495written_sect_headers = {} lp|features/lp/python/call_flow_logging/index.mdpytest
498def import_mod(test_mod_file): lp|features/lp/python/call_flow_logging/index.mdpytest
499 """
500 """
501 if not 'pytest' in sys.argv[0]:
502 return
504 if not '/' in test_mod_file:
505 test_mod_file = os.getcwd() + '/' + test_mod_file
506 d, fn = test_mod_file.rsplit('/', 1)
507 sys.path.insert(0, d) if not d in sys.path else 0
508 mod = import_module(fn.rsplit('.py', 1)[0])
510 fn_md = mod_doc(mod, dest='auto')
511 return fn_md
514def init_mod_doc(fn_mod): lp|features/lp/python/call_flow_logging/index.mdpytest
515 """convenience function for test modules"""
516 # cannot be done via FLG, need to parse too early:
517 if not os.environ.get('make_autodocs'):
518 return None, lambda: None
519 fn_md = import_mod(fn_mod)
520 plot = lambda: plot_build_flow(fn_md)
521 return fn_md, plot
524def add_flow_json_and_graph_easy_links(fn, T): lp|features/lp/python/call_flow_logging/index.mdpytest
525 """
526 The flags had been causing operators.build to create .graph_easy files before and after build
527 and also flow.json files for the before phase.
529 Now add those into the markdown.
530 """
531 # fn like '/home/gk/repos/lc-python/build/autodocs/tests/operators/test_op_ax_socket/test01_sock_comm/_plot_tag_.graph_easy.src'
532 d = os.path.dirname(fn)
533 ext = '.graph_easy.src'
534 # pre = fn.rsplit('/')[-1].split('_plot_tag_', 1)[0]
536 def link(f, d=d):
537 """Process one plot"""
538 fn, l = d + '/' + f + '.flow.json', ''
539 s = read_file(fn, dflt='')
540 if s:
541 p = '\n\n> copy & paste into Node-RED\n\n'
542 l = T.closed_admon('Flow Json', p + T.code('js', s), 'info')
543 # os.unlink(fn) # will be analysed by ops refs doc page
544 s = f.replace(ext, '')
545 dl = d.rsplit('/', 1)[-1]
546 r = '\n\n\n\n' % (dl, s)
547 # when s = 'test_build.py::Sharing::test_share_deep_copy.tests.post_build.svg'
548 # then n = 'tests.post_build':
549 if l:
550 return l + r
551 else:
552 return T.closed_admon(s, r, 'note')
554 ge = [link(f) for f in sorted(os.listdir(d)) if f.endswith(ext)]
555 return '\n'.join(ge)
558n_func = lambda func: func.__qualname__.replace('.<locals>', '') lp|features/lp/python/call_flow_logging/index.mdpytestpytest|tests.test_cfl.test_one
559d_usecase = lambda d_dest, use_case: d_dest + '/' + n_func(use_case) 559 ↛ exitline 559 didn't run the lambda on line 559lp|features/lp/python/call_flow_logging/index.mdpytest
562def name(obj): lp|features/lp/python/call_flow_logging/index.mdpytest
563 qn = getattr(obj, '__name__', None)
564 if qn:
565 return qn
566 qn = str(qn)
567 return obj.replace('<', '<').replace('>', '>')
570class write: lp|features/lp/python/call_flow_logging/index.mdpytest
571 def flow_chart(d_dest, use_case, clear=True, with_chart=False): lp|features/lp/python/call_flow_logging/index.mdpytest
572 if clear:
573 sys.settrace(None)
575 root_ = project.root()
576 # we log all modules:
577 mods = {}
578 # and all func sources:
579 sources = {}
580 # to find input and output
581 have = set()
582 os.makedirs(d_usecase(d_dest, use_case), exist_ok=True)
583 _ = write.arrange_frame_and_write_infos
584 flow_w = [
585 _(call, use_case, mods, sources, root_, d_dest, have)
586 for call in ILS.call_chain
587 ]
588 _ = make_call_flow_chart
589 fn_chart = _(flow_w, d_dest, fn=with_chart, ILS=ILS) if with_chart else 0
590 # post write, clear for next use_case:
591 ILS.clear() if clear else 0
592 return fn_chart
594 def arrange_frame_and_write_infos(call, use_case, mods, sources, root, d_dest, have): lp|features/lp/python/call_flow_logging/index.mdpytest
595 """Creates
596 [<callspec>, <input>, None] if input
597 [<callspec>, None, <output>, None] if output
598 (for the flow chart)
600 and writes module and func/linenr files plus the data jsons
601 """
602 l = [call]
603 frame = call.get('frame')
604 if frame in have:
605 # the output one, all other infos added already
606 have.add(frame)
607 l.extend([None, call.pop('output', '-')])
608 return l
609 have.add(frame)
610 fn = call['fn_mod']
611 if fn == 'note':
612 return [call, 'note', None]
613 d_mod = mods.get(fn)
614 if not d_mod:
615 d_mod = mods[fn] = write.module_filename_relative_to_root(fn, root)
616 os.makedirs(d_dest, exist_ok=True)
617 shutil.copyfile(fn, d_dest + '/' + d_mod[0])
618 d = d_mod[0]
619 call['pth'] = d_mod[1]
620 call['fn_mod'] = d.rsplit('/', 1)[-1]
621 func = sources.get(frame)
622 if not func:
623 src = inspect.getsource(frame)
624 src = deindent(src)
625 func_name = frame.f_code.co_name
626 if 'lambda' in func_name:
627 func_name = src.split(' = ', 1)[0].strip()
628 fn_func = '%s.%s.func.py' % (d, call['line'])
629 write_file(d_dest + '/' + fn_func, src)
630 sources[frame] = func = (fn_func, func_name)
631 fn_func, func_name = func
632 d = d_usecase(d_dest, use_case)
634 m = {
635 'line': call['line'],
636 'fn_func': fn_func,
637 'fn_mod': d_mod[0],
638 'dt': call['dt'],
639 'name': func_name,
640 'mod': '.'.join(d_mod[1]),
641 }
642 spec = [json.dumps(m), call['input'], call['output']]
643 fn = call['fn'] = '%s/%s.json' % (d, call['counter'])
644 write_file(fn, '\n-\n'.join(spec))
645 # s = json.dumps(call, default=str)
646 l.extend([call.pop('input'), None])
647 return l
649 def module_filename_relative_to_root(fn, root): lp|features/lp/python/call_flow_logging/index.mdpytest
650 d = (fn.split(root, 1)[-1]).replace('/', '__').rsplit('.py', 1)[0][2:]
651 pth = d.split('__')
652 return (d + '.mod.py'), pth