Coverage for src/lcdoc/tools.py: 61.39%

1"""

2Common tools for all modules

4"""

5import collections 2 ctxlppytest

6import hashlib 2 ctxlppytest

7import json 2 ctxlppytest

8import os 2 ctxlppytest

9import socket 2 ctxlppytest

10import sys 2 ctxlppytest

11import time 2 ctxlppytest

13import toml 2 ctxlppytest

15from lcdoc.log import app, now 2 ctxlppytest

17exists = os.path.exists 2 ctxlppytest

19# fastest algo:

20def file_hash(fn, algo='blake2b'): 2 ctxlppytest

21 return getattr(hashlib, algo)(open(fn, 'rb').read()).hexdigest() 1 ctxlp|features/lp/plugs/drawio/index.md

24def dirname(fn, create=False): 2 ctxlppytest

27 os.makedirs(d, exist_ok=True)

31def require(cmd, name=None, die=True, _have=set()): 2 ctxlppytest

32 '''cmd e.g. "rg --version"'''

36 if os.system(cmd + ' >/dev/null') == 0: 36 ↛ 40line 36 didn't jump to line 40, because the condition on line 36 was never false2 ctxlplp|features/lp/eval.md

37 _have.add(cmd) 2 ctxlplp|features/lp/eval.md

38 return True 2 ctxlplp|features/lp/eval.md

40 msg = 'Missing required command: %s' % name

41 if die:

42 app.die(msg, tried=cmd)

43 else:

44 app.warning(msg, tried=cmd)

45 return False

48def hostname(c=[0]): 2 ctxlppytest

49 i = c[0]

50 if i:

51 return i

52 c[0] = socket.gethostname()

53 return c[0]

56def into(d, k, v): 2 ctxlppytest

57 """the return is important for e.g. rx"""

58 d[k] = v

59 return d

62def to_list(o): 2 ctxlppytest

63 o = [] if o is None else o 2 ctxlp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one

64 t = type(o) 2 ctxlp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one

65 return o if t == list else list(o) if t == tuple else [o] 2 ctxlp|features/lp/python/call_flow_logging/index.mdpytest|tests.test_cfl.test_one

68# ----------------------------------------------------------------------------- file ops

69def walk_dir(directory, crit=None): 2 ctxlppytest

70 crit = (lambda *a: True) if crit is None else crit

71 files = []

72 j = os.path.join

73 for (dirpath, dirnames, filenames) in os.walk(directory):

74 files += [j(dirpath, file) for file in filenames if crit(dirpath, file)]

75 return files

78def read_file(fn, dflt=None, bytes=-1, strip_comments=False): 2 ctxlppytest

79 """

80 API function.

81 read a file - return a default if it does not exist"""

85 raise Exception(fn, 'does not exist')

87 # no idea why but __etc__hostname always contains a linesep at end

88 # not present in source => rstrip(), hope this does not break templs

92 lines = res.splitlines()

93 res = '\n'.join([l for l in lines if not l.startswith('#')])

97def insert_file(fn, content, sep): 2 ctxlppytest

98 s = read_file(fn).split(sep) 1 ctxlp|index.md

99 if not len(s) == 3: 99 ↛ 100line 99 didn't jump to line 100, because the condition on line 99 was never true1 ctxlp|index.md

100 raise Exception(f'No 2 times occurrance of sep in file. sep: {sep}, fn: {fn}')

101 write_file(fn, s[0] + sep + content + sep + s[2]) 1 ctxlp|index.md

102

103

104def write_file(fn, s, log=0, mkdir=0, chmod=None, mode='w', only_on_change=False): 2 ctxlppytest

105 'API: Write a file. chmod e.g. 0o755 (as octal integer)'

106

108

110 s = '\n'.join(s)

112 so = read_file(fn, dflt='xxx') 3 ctxlp|features/lp/parameters.mdlp|features/lp/plugs/make_badges/index.mdlp|index.md

113 if s == so: 3 ctxlp|features/lp/parameters.mdlp|features/lp/plugs/make_badges/index.mdlp|index.md

114 return 1 ctxlp|index.md

115

117 app.info('Writing file', fn=fn)

119 sep = '\n----------------------\n'

120 ps = (

121 s

122 if not 'key' in fn and not 'assw' in fn and not 'ecret' in fn

123 else '<hidden>'

124 )

125 app.debug('Content', content=sep + ps + sep[:-1])

132 if not isinstance(chmod, (list, tuple)):

133 chmod = [int(chmod)]

134 for s in chmod:

135 os.chmod(fn, s)

137 except IOError as ex: 137 ↛ 144line 137 didn't jump to line 1441 ctxlp|features/lp/python/call_flow_logging/index.md

138 if mkdir: 138 ↛ 143line 138 didn't jump to line 143, because the condition on line 138 was never false1 ctxlp|features/lp/python/call_flow_logging/index.md

139 mkdir = 0 1 ctxlp|features/lp/python/call_flow_logging/index.md

140 d = os.path.dirname(fn) 1 ctxlp|features/lp/python/call_flow_logging/index.md

141 os.makedirs(d) 1 ctxlp|features/lp/python/call_flow_logging/index.md

142 continue 1 ctxlp|features/lp/python/call_flow_logging/index.md

143 e = ex

144 except Exception as ex:

145 e = ex

146 app.die('Could not write file', filename=fn, exc=e)

147

148

149have_tty = lambda: sys.stdin.isatty() and sys.stdout.isatty() 149 ↛ exitline 149 didn't run the lambda on line 1492 ctxlppytest

150

151

152def flatten(d, sep='_', tpljoin=None): 2 ctxlppytest

153 """when tpljoin is given we detect them as keys and join

154 (enables json serialization)"""

155

156 obj = collections.OrderedDict() 1 ctxlp

157

158 def recurse(t, parent_key=''): 1 ctxlp

159

160 if isinstance(t, list): 1 ctxlp

161 for i in range(len(t)): 1 ctxlp

162 recurse(t[i], parent_key + sep + str(i) if parent_key else str(i)) 1 ctxlp

163 elif isinstance(t, dict): 1 ctxlp

164 for k, v in t.items(): 1 ctxlp

165 if isinstance(k, tuple): 165 ↛ 166line 165 didn't jump to line 166, because the condition on line 165 was never true1 ctxlp

166 k = tpljoin.join(k)

167 recurse(v, parent_key + sep + k if parent_key else k) 1 ctxlp

168 else:

169 obj[parent_key] = t 1 ctxlp

170

171 recurse(d) 1 ctxlp

172

173 return obj 1 ctxlp

174

175

176from collections import OrderedDict as OD 2 ctxlppytest

177

178

179def unflatten(dictionary, sep='_'): 2 ctxlppytest

180 resultDict = OD() 1 ctxlp

181 for key, value in dictionary.items(): 1 ctxlp

182 parts = key.split(sep) 1 ctxlp

183 d = resultDict 1 ctxlp

184 for part in parts[:-1]: 1 ctxlp

185 if part not in d: 1 ctxlp

186 d[part] = OD() 1 ctxlp

187 if isinstance(d[part], str): 1 ctxlp

188 d[part] = OD({None: d[part]}) 1 ctxlp

189 d = d[part] 1 ctxlp

190

191 d[parts[-1]] = value 1 ctxlp

192 return resultDict 1 ctxlp

193

194

195def deep_update(o, n, sep): 2 ctxlppytest

196 of = flatten(o, sep)

197 nf = flatten(n, sep)

198 of.update(nf)

199 return unflatten(of, sep)

200

201

202class project: 2 ctxlppytest

203 """loads the project config

204

205 Main app config put into project.cfg['app']

206 An optional tool.lc.app section will be merged into that

207

208 TODO: other formats than pyproject.toml

209

210 """

211

212 config, dir_home, fn_cfg = {}, None, None 2 ctxlppytest

213

214 def root(config=None, root=None, c=[0]): 2 ctxlppytest

215 """

216 config must given first time. Understood currently: mkdocs config

217 """

219 c[0] = root # for tests, to force it 1 ctxpytest

222 ret = 0 2 ctxlppytest

223 try: 2 ctxlppytest

224 if config and hasattr(config, '__getitem__'): 224 ↛ 229line 224 didn't jump to line 229, because the condition on line 224 was never false2 ctxlppytest

225 dd = config.get('docs_dir') 2 ctxlppytest

226 if dd: 226 ↛ 229line 226 didn't jump to line 229, because the condition on line 226 was never false2 ctxlppytest

227 ret = c[0] = dirname(dd) 2 ctxlppytest

228 return ret 2 ctxlppytest

229 try:

230 # if we run in pytest *and* have devapp, we can create call flow logs

231 # for devapps:

232 from devapp.tools import project

233

234 ret = c[0] = project.root()

235 return ret

236 except:

237 pass

238 # todo: look at git here

239 raise Exception('Cannot derive project root')

240 finally:

241 c[0] = ret 2 ctxlppytest

242

243 fn_config = lambda: project.root() + '/pyproject.toml' 4 ctxlplp|features/lp/plugs/make_badges/index.mdlp|index.mdpytest

244 d_autodocs = lambda: project.root() + '/build/autodocs' 244 ↛ exitline 244 didn't run the lambda on line 2442 ctxlppytest

245

246 def abs_path(fn, config=None, mkdirs=False): 2 ctxlppytest

247 r = project.root(config)

248 if not fn:

249 return r

250 if fn[0] != '/':

251 fn = r + '/' + fn

252 if mkdirs:

253 os.makedirs(dirname(fn), exist_ok=True)

254 return fn

255

256 # TODO: understand also poetry and piptools:

257 def load_config(): 2 ctxlppytest

258

259 fn = project.fn_config() 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

260 if not exists(fn): 260 ↛ 261line 260 didn't jump to line 261, because the condition on line 260 was never true2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

261 app.die('no config found', fn=fn)

262

263 cfg = toml.load(fn) 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

264 app.info('loaded config', filename=fn) 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

265 c = project.config 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

266 c.update(cfg) 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

267 if not 'project' in c: 267 ↛ 268line 267 didn't jump to line 268, because the condition on line 267 was never true2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

268 c['project'] = {'urls': {}}

269 if 'tool' in c and 'poetry' in c['tool']: 269 ↛ 274line 269 didn't jump to line 274, because the condition on line 269 was never false2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

270 c['project'].update(c['tool']['poetry']) 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

271 c['project']['urls']['homepage'] = c['project']['homepage'] 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

272 c['project']['urls']['repository'] = c['project']['repository'] 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

273

274 t = cfg['tool'] 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

275 # breakpoint() # FIXME BREAKPOINT

276 # c['app'] = t['poetry']

277 # c['app'].update(t.get('lc', {}).get('app', {}))

278 project.fn_cfg = fn 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

279 # app.die('Did not find a pyproject.toml file with badges declared')

280 return project.config 2 ctxlp|features/lp/plugs/make_badges/index.mdlp|index.md

281

282 def conf(): 2 ctxlppytest

283 return project.config or project.load_config() 2 ctxlp|about/credits.mdlp|index.md

284

285 def name(): 2 ctxlppytest

286 p = project.conf() 2 ctxlp|about/credits.mdlp|index.md

287 return p['project']['name'] 2 ctxlp|about/credits.mdlp|index.md

288

289 def urls(): 2 ctxlppytest

290 p = project.conf()['project'] 1 ctxlp|index.md

291 if 'urls' in p: 291 ↛ 293line 291 didn't jump to line 293, because the condition on line 291 was never false1 ctxlp|index.md

292 return p['urls'] 1 ctxlp|index.md

293 urls = 'packagehome', 'discusshome', 'homepage', 'repository'

294 return {k: p.get(k, '') for k in urls}

295

296 def homepage(): 2 ctxlppytest

297 return project.urls().get('homepage', 'n.a.')

298

299 def repository(): 2 ctxlppytest

300 return project.urls().get('repository', 'n.a.')

301

302 def packagehome(): 2 ctxlppytest

303 return (

304 project.urls().get('packagehome')

305 or 'https://pypi.org/project/%s/_VERSION_/' % project.name()

306 ).replace('_VERSION_', project.version())

307

308 def version(): 2 ctxlppytest

309 p = project.conf()['project'] 1 ctxlp|index.md

310 v = p['version'] 1 ctxlp|index.md

311 if isinstance(v, dict) and v.get('use_scm'): 311 ↛ 312line 311 didn't jump to line 312, because the condition on line 311 was never true1 ctxlp|index.md