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