Coverage for src/lcdoc/lp.py: 80.40%

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

1#!/usr/bin/env python

2"""

3Literate Programming / Language of Babel style

5Edit: We support this now not only from emacs but also from normal mkdocs markdown.

6See the documentation.

10Old emacs centric docs follow:

12Org Mode / Mkdocs Helper Script.

14Usage: See e.g. repos/lc/docs/.../getting_started.org

16Basics:

17 In you org file have this:

20#+BEGIN_SRC python :exports none :results silent :session pyroot

21from sys import modules as m; m.pop('lp') if 'lp' in m else 0; import lp

22#lp.alias('lcm', '$d_repos/lc/build/make')

23run = lp.p(lp.run, fmt='mk_cmd_out')

24root = lp.p(run, root=True)

25#+END_SRC

27which you execute at any change of the module (reloads)

29Set this:

31:PROPERTIES:

32:header-args:python: :session pyroot :exports results :results html

33:END:

35Then using is like

37#+BEGIN_SRC python

38root([

39 {

40 'cmd': 'm -l run lnr -d',

41 'expect': '\n/opt/repos/lc/node-red#',

42 'timeout': 15,

43 },

44 {

45 'cmd': "ax/start -p ''",

46 'expect': 'Started flows',

47 'timeout': 10,

48 'cmt': "-p '': no project"

49 },

50 ], new_session='root_nr')

51#+END_SRC

54"""

55import hashlib 2 ctxlppytest

56import html 2 ctxlppytest

57import json 2 ctxlppytest

58import os 2 ctxlppytest

59import string 2 ctxlppytest

60import subprocess as sp 2 ctxlppytest

61import sys 2 ctxlppytest

62import time 2 ctxlppytest

63from functools import partial as p 2 ctxlppytest

64from importlib import import_module 2 ctxlppytest

66import pycond 2 ctxlppytest

68from lcdoc.mkdocs.tools import page_dir 2 ctxlppytest

69from lcdoc.tools import app, dirname, write_file 2 ctxlppytest

71# important (colorize)

72I = lambda s: s if not sys.stdout.isatty() else '\x1b[1;32m%s\x1b[0m' % s 72 ↛ exitline 72 didn't run the lambda on line 722 ctxlppytest

75env = os.environ 2 ctxlppytest

76wait = time.sleep 2 ctxlppytest

77now = time.time 2 ctxlppytest

78exists = os.path.exists 2 ctxlppytest

79user = env['USER'] 2 ctxlppytest

80env['PATH'] = 'bin:%s' % env.get('PATH', '') 2 ctxlppytest

84dbg = p(_, app.debug) 2 ctxlppytest

85nfo = p(_, app.info) 2 ctxlppytest

88# ---------------------------------------------------------------------------- Utilities

89def check_assert(ass, res): 2 ctxlppytest

91 msg = 'Assertion failed: Expected "%s" not found in result' % ass

92 app.error(msg, asserts=ass, json={'result': res})

93 raise Exception(msg)

100 is_pycond = True 3 ctxlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/parameters.md

101 elif isinstance(ass, list) and ass[1] in {'and', 'or'}: 101 ↛ 102line 101 didn't jump to line 102, because the condition on line 101 was never true12 ctxlp|features/lp/examples.mdlp|features/lp/examples.mdlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/parameters.mdlp|features/lp/parameters.mdlp|features/lp/parameters.mdlp|features/lp/parameters.mdlp|features/lp/syntax.mdlp|features/lp/syntax.mdlp|features/lp/syntax.mdlp|features/lp/tips.md

102 is_pycond = True

104

105 def f(k, v, state, **kw): 3 ctxlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/parameters.md

106 return (k in state['res'], v) 3 ctxlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/parameters.md

107

108 r = pycond.pycond(ass, f) 3 ctxlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/parameters.md

109 a = r(state={'res': s}) 3 ctxlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/parameters.md

110 if not a: 110 ↛ 111line 110 didn't jump to line 111, because the condition on line 110 was never true3 ctxlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/parameters.md

111 raise_()

112 return 3 ctxlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/parameters.md

120 raise_()

123# aliases = {}

127def repl_dollar_var_with_env_val(s, die_on_fail=True): 2 ctxlppytest

130 for k in env:

131 if k in s:

132 s = s.replace('$' + k, env[k])

133 if '$' in s:

134 dbg('Not defined in environ: $%s' % s)

135 return s

136

137

138def spresc(cmd): 2 ctxlppytest

139 """subprocess run - with escape sequences escaped (for tmux)"""

141

142

143def sprun(*a, no_fail=False, report=False, **kw): 2 ctxlppytest

144 """Running a command as bash subprocess

145 W/o executable we crash on ubuntu's crazy dash, which cannot even echo -e

146 """

148 if not kw and len(a) == 1:

149 report('', a[0])

150 else:

151 report('', a, kw) # for the user (also in view messages)

152

156 except Exception as ex:

157 if not no_fail:

158 raise

159 app.error('Failure', args=a, kw=kw, exc=ex)

160 return err(str(ex))

161

162

163# ----------------------------------------------------------------------- Plugins(modes)

164

165our_plugs = 'lcdoc.mkdocs.lp.plugs' 2 ctxlppytest

166

167

168class plugs: 2 ctxlppytest

169 pass 2 ctxlppytest

170

171

172def get_or_import_plug(mode): 2 ctxlppytest

178 if not hasattr(p, 'run'):

179 raise ModuleNotFoundError('')

183 except ModuleNotFoundError as ex:

184 msg = ' - If this is a custom module add its directory to your PYTHONPATH'

185 raise type(ex)(str(ex) + msg)

188

189

190# ------------------------------------------------------------------------------ Formats

191# seems we don't need the stupid dot anymore for the javascript xterm part (?)

192# mk_cmd_out = '''

193

194# === "Cmd"

195

196# ```console

197# %(cmds)s

198# ```

200# .

202# === "Output"

204# %(ress)s

206# '''

208mk_console = '''

209```%(lang)s

210%(cmd)s

211%(res)s

212```

213'''

216mk_cmd_out = '''

218=== "Cmd"

220 ```console

221 %(cmds)s

222 ```

224=== "Output"

226 %(ress)s

228'''

231class mk_cmd_out_fetch: 2 ctxlppytest

232 org = '''

233

234

235#+begin_tab:Command

236

237#+BEGIN_SRC console :eval no

239%(cmds)s

241#+END_SRC

243#+end_tab:Command

246#+begin_tab:Output

247

248%(ress)s

249

250#+end_tab:Output

251

252'''

253 mkdocs = mk_cmd_out 2 ctxlppytest

254

255

256xt_flat = '''

257<xterm %(root)s/>

258

259 %(ress)s

260

261'''

262

263xt_flat_fetch = '''

264 <xterm />

265

266 remote_content

267

268 ![](%(fn_frm)s)

269'''

270

271

272# xt_flat_fetch_test = """

273##+begin_tab:foo

274

275# <xtf %(root)s xtfrm=file:%(fn_frm)s></xtf>

276

277##+end_tab:foo

278# """

279

280# this not yet sane:

281##+begin_tab:foo

282

283# <xterm></xterm>

284

285# ansiremotecontent

286

287# [ ](file:./media/info_flow_adm.ansi)

288

289##+end_tab:foo

290

291

292dflt_fmt = 'mk_cmd_out' 2 ctxlppytest

293

294

295class T: 2 ctxlppytest

296 def xt_flat(resl, **kw): 2 ctxlppytest

303 res = cmdstr + '\n' + res

308 fn_frm = kw['fn_frm'] = file_.write_fetchable(ress, **kw) 1 ctxlp|features/lp/parameters.md

309 t = xt_flat_fetch 1 ctxlp|features/lp/parameters.md

310 ress = t % locals() 1 ctxlp|features/lp/parameters.md

311 else:

315

316 def mk_cmd_out(res, **kw): 2 ctxlppytest

318

320 """

321 We don't do it since console rendering requires $ or # only for syn hilite

322 """

329 if len(c) + len(cmt) > 80:

330 c = '%s # %s:\n%s' % (p, cmt, c)

331 else:

332 c = '%s # %s' % (c, cmt)

334

338 return ress 1 ctxlp|features/lp/parameters.md

342 t = getattr(mk_cmd_out_fetch, kw['fetched_block_fmt']) 1 ctxlp|features/lp/parameters.md

343 r = t % locals() 1 ctxlp|features/lp/parameters.md

344 else:

345 # indent one in:

350

351 def mk_console(res, root=False, **kw): 2 ctxlppytest

362 # is the command part of the res? then skip print:

364 cmd = 'SKIP-PRINT-OUT' 4 ctxlp|features/lp/parameters.mdlp|features/lp/parameters.mdlp|features/lp/plugs/make_file/index.mdlp|features/lp/syntax.md

366

370

371

372T.xtf = T.xt_flat 2 ctxlppytest

373

374

375def get_cmd(res): 2 ctxlppytest

378

379

380# def alias(k, v):

381# aliases[k + ' '] = v + ' '

382

383

384def prepare_and_fmt(res, orig_cmd, **kw): 2 ctxlppytest

385 # allow plugs to do their formatting:

394 c = res[i] = {}

396 app.debug('Command missing', res=o, cmd=orig_cmd) 1 ctxlp|features/lp/python/_tech.md

397 c['cmd'] = orig_cmd 1 ctxlp|features/lp/python/_tech.md

399 app.error('Result missing', res=o, cmd=orig_cmd)

400 c['res'] = 'Not available (%s)' % orig_cmd

404 else:

405 c['res'] = json.dumps(r, indent=4)

407 c['cmd'] = '' 1 ctxlp|features/lp/parameters.md

408

412 raise Exception(

413 'Format not found (%s) - have: %s'

414 % (f, [k for k in dir(T) if not k[0] == '_'])

415 )

417

418

419letters = string.ascii_letters + string.digits + '_' 2 ctxlppytest

420

421

422# ----------------------------------------------------------------------- header parsing

423def cast(v): 2 ctxlppytest

425 return json.loads(v) 1 ctxlp|features/lp/examples.md

433

434

435apos = ["'", '"'] 2 ctxlppytest

436neutrl_chars = [[' ', '\x01'], [',', '\x02'], ['=', '\x04']] 2 ctxlppytest

437

438

439def neutralize(s, mode=None): 2 ctxlppytest

445

446

447def parse_kw_str(kws, header_kws=None, try_json=True): 2 ctxlppytest

448 """for kw via cli"""

450 return {} 8 ctxlp|features/lp/examples.mdlp|features/lp/parameters.mdlp|features/lp/plugs/chart/index.mdlp|features/lp/plugs/column/index.mdlp|features/lp/plugs/flowchart/index.mdlp|features/lp/plugs/kroki/index.mdlp|features/lp/sessions.mdlp|index.md

453 if kws and kws[0] in ('{', '['):

454 try:

455 return json.loads(kws)

456 except:

457 pass

458 # neutralize all within apos:

467 msg = 'Unparsable apostrophes: %s - %s'

468 raise Exception(msg % (kws, apo))

471

472 # quoted values possible:

474 raise Exception('No comma allowed') 1 ctxlp|features/lp/examples.md

478 {

479 p[0]: cast(neutralize(p[1], 'rev'))

480 for p in [(k if '=' in k else k + '=true').split('=') for k in parts]

481 }

482 )

483 # breakpoint() # FIXME BREAKPOINT

484 # if '"' in kws or "'" in kws:

485 # r = []

486 # while kw:

487 # k, v = kw.pop(0)

488 # for apo in apos:

489 # if v[0] == apo:

490 # v = v[1:]

491 # if v.endswith(apo):

492 # v = v[:-1]

493 # break

494 # while kw:

495 # nk, _ = kw.pop(0)

496 # if nk.endswith(apo):

497 # v += ' ' + nk[:-1]

498 # break

499 # else:

500 # v += ' ' + nk

501 # break

502 # r.append([k, v])

503 # kw = r

504 # kw = {k: cast(v) for k, v in kw}

505

508

509

510def parse_header_args(header, **ctx): 2 ctxlppytest

511 """Parsing either python or easy format

512 CAUTION: This is used

513 - caused here, to parse inline headers ( # lp: ....)

514 - in the plugin code as well, to parse markdown headers

515 """

516 # ctx['dir_repo'] = ctx['fn_lp'].split('/docs/', 1)[0]

518

520 # if header.rstrip().endswith('"'):

521 # raise Exception("Not tried (apostrophe at end)")

523 except Exception as ex: 2 ctxlp|features/lp/examples.mdlp|features/lp/examples.md

524 ex1 = ex 2 ctxlp|features/lp/examples.mdlp|features/lp/examples.md

525 # evaling now.

526 # still - we supply only a minimum eval ctx and prevent

527 # imports. But still this won't be totally safe.

528 # BUT: Hey - we are about to run code *anyway*, this is LP in the end!

529 if not 'import' in header: 529 ↛ 535line 529 didn't jump to line 535, because the condition on line 529 was never false2 ctxlp|features/lp/examples.mdlp|features/lp/examples.md

530 try: 2 ctxlp|features/lp/examples.mdlp|features/lp/examples.md

531 m = eval('get_args(%s)' % header, ctx, {}) 2 ctxlp|features/lp/examples.mdlp|features/lp/examples.md

532 return 0, (m['args'], m['kw']) 2 ctxlp|features/lp/examples.mdlp|features/lp/examples.md

533 except Exception as ex:

534 ex2 = ex

535 return 1, (ex2, ex1)

536

537

538# ----------------------------------------------------------------------------- sessions

539

540

541def pb(s): 2 ctxlppytest

542 try:

543 s = s.decode('utf-8')

544 except Exception as ex:

545 pass

546 dbg(s)

547

548

549def get_args(*a, **kw): 2 ctxlppytest

550 return {'args': a, 'kw': kw} 2 ctxlp|features/lp/examples.mdlp|features/lp/examples.md

551

552

553def root(cmd, **kw): 2 ctxlppytest

554 return run(cmd, root=True, **kw)

555

556

557class file_: 2 ctxlppytest

558 def write_fetchable(cont, fetch, **kw): 2 ctxlppytest

559 """write .ansi XTF files"""

560 d, fn = kw['fn_doc'].rsplit('/', 1) 1 ctxlp|features/lp/parameters.md

561 lnk = '/media/%s_%s.ansi' % (fn, fetch) 1 ctxlp|features/lp/parameters.md

562 fn = d + lnk 1 ctxlp|features/lp/parameters.md

563 if not exists(d + '/media'): 563 ↛ 565line 563 didn't jump to line 565, because the condition on line 563 was never false1 ctxlp|features/lp/parameters.md

564 os.makedirs(d + '/media') 1 ctxlp|features/lp/parameters.md

565 s = rpl(cont, kw) 1 ctxlp|features/lp/parameters.md

566 write_file(fn, s, only_on_change=True) 1 ctxlp|features/lp/parameters.md

567 return '.' + lnk 1 ctxlp|features/lp/parameters.md

568

569

570def rpl(s, kw): 2 ctxlppytest

571 'global replacement'

574 if not isinstance(rpl[0], (list, tuple)):

575 rpl = [rpl]

576 for r in rpl:

577 s = s.replace(r[0], r[1])

579

580

581def repl_dollar_var_with_env_vals(kw, *keys): 2 ctxlppytest

586

587

588def multi_line_to_list(cmd): 2 ctxlppytest

589 """

590

591 ```bash lp session=DO asserts=loadbalancer.tf

592 ip () { echo 1.2.3.4; }

593 cat << FOO > loadbalancer.tf

594 > resource "digitalocean_loadbalancer" "www-lb" {

595 > ipv4 = $(ip)

596 > (...)

597 >}

598 >FOO

599 ls -l

600 ```

601

602 """

604 return cmd 2 ctxlp|features/lp/parameters.mdlp|features/lp/parameters.md

606 # when some lines start with ' ' we send the whole cmd as one string:

608 # echo 'foo

609 # bar' > baz

610 return [{'cmd': cmd}] 1 ctxlp|features/lp/examples.md

616 l = lines.pop(0) 3 ctxlp|features/lp/examples.mdlp|features/lp/syntax.mdlp|features/lp/tips.md

617 r[-1] += '\n' + l[2:] 3 ctxlp|features/lp/examples.mdlp|features/lp/syntax.mdlp|features/lp/tips.md

620

621

622def check_inline_lp(cmd): 2 ctxlppytest

624 breakpoint() # FIXME BREAKPOINT

625 return

631 msg = 'Inline lp construct wrong: %s. Valid e.g.: "ls -lta /etc # lp: '

632 msg += 'expect=hosts timeout=10". Got: %s %s'

633 raise Exception(msg % (cmd, res[0], res[1]))

636

637

638def os_system_if_param_present(kw, if_present): 2 ctxlppytest

639 """utility for a frequent use case"""

641 return

644 if os.system(cmd):

645 raise Exception('%s run failed: %s. kw: %s' % (if_present, cmd, str(kw)))

646

647

648def err(msg, **kw): 2 ctxlppytest

649 app.error(msg, **kw)

650 return {'res': {'lp err': msg, 'kw': kw}}

651

652

653class SessionNS: 2 ctxlppytest

654 """Namespace for methods related to session parameters

655

656 Including methods when there is no session support

657 (but different handling when there is)

658 """

659

660 @classmethod 2 ctxlppytest

661 def init(cls, cmd, kw):

665

666 @classmethod 2 ctxlppytest

667 def handle_new_session_param(cls, kw):

672 raise Exception(msg)

674 ns = kw.pop('session')

675 if not ns:

676 raise Exception(msg)

680

681 @classmethod 2 ctxlppytest

682 def delete(cls, session_name, kw):

683 raise Exception('No session support', mode=kw.get('mode'))

684

685 @classmethod 2 ctxlppytest

686 def pre(cls, session_name, kw):

687 # pre, if not overwritten here should just os.system it:

689

690 @classmethod 2 ctxlppytest

691 def post(cls, session_name, kw):

693

694

695def eval_lp(cmd, kw): 2 ctxlppytest

696 # deprecated alias, not any more docued, did not work for py style args:

703 res = cmd = res.get('result') or res.get('res') 2 ctxlp|features/lp/plugs/chart/index.mdlp|features/lp/plugs/flowchart/index.md

704 if isinstance(res, dict) and res.get('header'): 704 ↛ 705line 704 didn't jump to line 705, because the condition on line 704 was never true2 ctxlp|features/lp/plugs/chart/index.mdlp|features/lp/plugs/flowchart/index.md