Coverage for selenible/modules/ctrl.py: 52%

1import json

2import logging.config

3import tempfile

4import time

5import urllib.parse

6from contextlib import ExitStack

7from subprocess import DEVNULL

9import requests

10import toml

11import yaml

12from lxml import etree

14progn_schema = yaml.safe_load("""

15type: array

16items: {type: object}

17""")

20def Base_progn(self, param):

21 """

22 - name: subroutine

23 progn:

24 - name: debug1

25 echo: hello world

26 - name: debug2

27 echo: good-bye world

28 """

29 self.lock.release()

30 ret = self.run(param)

31 self.lock.acquire()

32 return ret

35var_schema = {"type": "object"}

38def Base_var(self, param):

39 """

40 - name: set variables

41 var:

42 key1: value1

43 key2:

44 - value2.1

45 - value2.2

46 """

47 self.variables.update(param)

50var_if_not_schema = var_schema

53def Base_var_if_not(self, param):

54 """

55 - name: set variables

56 var_if_not:

57 key1: value1

58 key2:

59 - value2.1

60 - value2.2

61 """

62 for k, v in param.items():

63 if k not in self.variables:

64 self.variables[k] = v

67var_from_schema = yaml.safe_load("""

68type: object

69properties:

70 yaml: {type: string}

71 json: {type: string}

72 toml: {type: string}

73""")

76def Base_var_from(self, param):

77 """

78 - name: set variables from file

79 var_from:

80 yaml: filename

81 json: filename

82 toml: filename

83 """

84 if "yaml" in param:

85 with open(param.get("yaml")) as f:

86 self.do_var(yaml.safe_load(f))

87 if "json" in param:

88 with open(param.get("json")) as f:

89 self.do_var(json.load(f))

90 if "toml" in param:

91 with open(param.get("toml")) as f:

92 self.do_var(toml.load(f))

95var_from_if_not_schema = var_from_schema

98def Base_var_from_if_not(self, param):

99 """

100 - name: set variables from file

101 var_from_if_not:

102 yaml: filename

103 json: filename

104 toml: filename

105 """

106 if "yaml" in param:

107 with open(param.get("yaml")) as f:

108 self.do_var_if_not(yaml.safe_load(f))

109 if "json" in param:

110 with open(param.get("json")) as f:

111 self.do_var_if_not(json.load(f))

112 if "toml" in param:

113 with open(param.get("toml")) as f:

114 self.do_var_if_not(toml.load(f))

115

116

117def Base_runcmd(self, param):

118 """

119 - name: run shell command

120 runcmd: echo hello

121 - name: word count

122 runcmd:

123 stdin: "{{page_source}}"

124 cmd: wc

125 """

126 if isinstance(param, (list, tuple, str)):

127 self.log.debug("run: %s", param)

128 out = self.runcmd(param)

129 self.log.debug("result: %s", out)

130 return out

131 elif isinstance(param, dict):

132 cmd = param.get("cmd", None)

133 stdin = param.get("stdin", None)

134 stdout = param.get("stdout", None)

135 stderr = param.get("stderr", None)

136 if cmd is None:

137 raise Exception(f"missing cmd: {param}")

138 with ExitStack() as stack:

139 if stdin is not None:

140 sin = stack.enter_context(tempfile.TemporaryFile())

141 sin.write(stdin.encode("utf-8"))

142 sin.seek(0)

143 else:

144 sin = DEVNULL

145 if stderr is not None:

146 serr = stack.enter_context(open(stderr))

147 else:

148 serr = DEVNULL

149 out = self.runcmd(cmd, stdin=sin, stderr=serr)

150 self.log.info("result: %s", out)

151 if stdout is not None:

152 with open(stdout, "w") as f:

153 f.write(out)

154 return out

155 else:

156 raise Exception(f"runcmd: param not supported: {param}")

157

158

159echo_schema = yaml.safe_load("""

160oneOf:

161 - type: string

162 - "$ref": "#/definitions/common/textvalue"

163""")

164

165

166def Base_echo(self, param):

167 """

168 - name: debug message

169 echo:

170 text: hello world

171 """

172 if isinstance(param, str):

173 self.log.info("echo %s", param)

174 return param

175 else:

176 txt = self.getvalue(param)

177 self.log.info("echo %s", txt)

178 return txt

179

180

181sleep_schema = {"type": "integer"}

182

183

184def Base_sleep(self, param):

185 """

186 - name: wait 10 sec

187 sleep: 10

188 """

189 self.lock.release()

190 time.sleep(int(param))

191 self.lock.acquire()

192

193

194include_schema = yaml.safe_load("""

195oneOf:

196 - type: string

197 - type: array

198 items: {type: string}

199""")

200

201

202def Base_include(self, param):

203 """

204 - name: run other file

205 include: filename.yaml

206 - name: run other files

207 include:

208 - file1.yaml

209 - file2.yaml

210 """

211 if isinstance(param, (list, tuple)):

212 for fname in param:

213 self.log.info("loading %s", fname)

214 with open(fname) as f:

215 self.lock.release()

216 ret = self.run(yaml.safe_load(f))

217 self.lock.acquire()

218 return ret

219 elif isinstance(param, str):

220 self.log.info("loading %s", param)

221 with open(param) as f:

222 self.lock.release()

223 ret = self.run(yaml.safe_load(f))

224 self.lock.acquire()

225 return ret

226 else:

227 raise Exception(f"cannot load: {param}")

228

229

230def Base_config(self, param):

231 """

232 - name: configuration

233 config:

234 wait: 10

235 cookie:

236 var1: val1

237 window:

238 width: 600

239 height: 480

240 """

241 if "wait" in param:

242 self.log.debug("implicitly wait %s sec", param.get("wait"))

243 self.driver.implicitly_wait(param.get("wait"))

244 if "cookie" in param:

245 self.log.debug("cookie update: {}".format(param.get("cookie", {}).keys()))

246 self.driver.add_cookie(param.get("cookie"))

247 if "window" in param:

248 self.log.info("window size update: {}".format(param.get("window", {})))

249 win = param.get("window")

250 if win.get("maximize", False):

251 self.driver.maximize_window()

252 else:

253 x = win.get("x")

254 y = win.get("y")

255 if x is not None and y is not None:

256 self.log.debug("set window pos: x=%d, y=%d", x, y)

257 self.driver.set_window_position(x, y)

258 width = win.get("width")

259 height = win.get("height")

260 if width is not None and height is not None:

261 self.log.debug("set window size: width=%d, height=%d", width, height)

262 self.driver.set_window_size(width, height)

263 if "log" in param:

264 logconf = param.get("log")

265 if isinstance(logconf, str):

266 logging.config.fileConfig(logconf)

267 elif isinstance(logconf, dict):

268 logging.config.dictConfig(logconf)

269 else:

270 raise Exception("config.log must be filename or dict: %s", param.get("log"))

271 if "page_load_timeout" in param:

272 self.driver.set_page_load_timeout(param.get("page_load_timeout"))

273 if "implicitly_wait" in param:

274 self.driver.implicitly_wait(param.get("implicitly_wait"))

275 if "script_timeout" in param:

276 self.driver.set_script_timeout(param.get("script_timeout"))

277

278

279assert_schema = {"$ref": "#/definitions/common/condition"}

280

281

282def Base_assert(self, param):

283 """

284 - name: check condition

285 assert:

286 eq:

287 - "{{selenible_version}}"

288 - "0.1"

289 """

290 if not self.eval_param(param):

291 self.log.error("condition failed: %s", param)

292 raise Exception(f"condition failed: {param}")

293

294

295assert_not_schema = assert_schema

296

297

298def Base_assert_not(self, param):

299 """

300 - name: check condition

301 assert_not:

302 eq:

303 - "{{selenible_version}}"

304 - "0.3"

305 """

306 if self.eval_param(param):

307 self.log.error("condition(not) failed: %s", param)

308 raise Exception(f"condition(not) failed: {param}")

309

310

311xslt_schema = yaml.safe_load("""

312type: object

313properties:

314 proc: {type: string}

315 output: {type: string}

316""")

317

318

319def Base_xslt(self, param):

320 """

321 - name: transform

322 xslt:

323 proc: |

324 <xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform">

325 <xsl:template match="/">

326 <xsl:value-of select="//a/@href" />

327 </xsl:template>

328 </xsl:stylesheet>

329 output: outfile.txt

330 """

331 if isinstance(param, dict):

332 proc = etree.XSLT(etree.XML(param.get("proc", "")))

333 output = param.get("output", None)

334 elem = self.findmany(param)

335 if elem == [None]:

336 p = etree.parse(self.driver.page_source)

337 rst = [str(proc(p))]

338 else:

339 rst = []

340 for e in elem:

341 p = etree.parse(e.get_attribute("innerHTML"))

342 rst.append(str(proc(p)))

343 if output is not None:

344 with open(output, "w") as f:

345 f.writelines(str(x) for x in rst)

346 return rst

347 else:

348 raise Exception(f"invalid parameter: {param}")

349

350

351download_schema = yaml.safe_load("""

352type: object

353properties:

354 url: {type: string}

355 method: {type: string}

356 query: {type: object}

357 headers: {type: object}

358 json: {type: boolean}

359 output: {type: string}

360""")

361

362

363def Base_download(self, param):

364 """

365 - name: download file using python-requests

366 download:

367 url: "{{current_url}}/file1"

368 method: get

369 query:

370 var1: val1

371 timeout: 10

372 json: false

373 output: outfile.txt

374 """

375 url = param.get("url", None)

376 if url is None:

377 raise Exception(f"url mut set: {param}")

378 parsed_url = urllib.parse.urlparse(url)

379 self.log.debug("URL parsed: %s", parsed_url)

380 method = param.get("method", "get")

381 query = param.get("query", None)

382 cookies = self.driver.get_cookies()

383 headers = param.get("headers", None)

384 timeout = param.get("timeout", None)

385 is_json = param.get("json", False)

386 sess = requests.Session()

387 output = param.get("output", None)

388 for ck in cookies:

389 sess.cookies.set(

390 ck.get("name"),

391 ck.get("value"),

392 path=ck.get("path", "/"),

393 domain=ck.get("domain", ""),

394 secure=ck.get("secure", False),

395 )

396 resp = sess.request(method, url, params=query, headers=headers, timeout=timeout)

397 if output is not None:

398 with open(output, "wb") as f:

399 f.write(resp.content)

400 if is_json:

401 return resp.json()

402 return resp.text

403

404

405set_schema = yaml.safe_load("""

406anyOf:

407 - "$ref": "#/definitions/common/locator"

408 - "$ref": "#/definitions/common/textvalue"

409 - type: object

410 properties:

411 parseHTML: {type: boolean}

412""")

413

414

415def Base_set(self, param):

416 """

417 - name: set variable (v1="blabla")

418 register: v1

419 set:

420 text: blabla

421 - name: set variable v2

422 register: v2

423 set:

424 xpath: '//a'

425 parseHTML: true

426 - name: echo

427 echo: 'v1={{v1}}, v2={%for x in v2%}{{x.get("href")}},{%endfor%}'

428 """

429 res = self.getvalue(param)

430 if res is not None: 430 ↛ 432line 430 didn't jump to line 432 because the condition on line 430 was always true

431 return self.return_element(param, res)

432 return self.return_element(param, self.findmany(param))