Coverage for selenible/drivers/base.py: 62%

1import copy

2import functools

3import getpass

4import importlib.resources

5import inspect

6import io

7import json

8import os

9import subprocess

10import sys

11import time

12from logging import getLogger

13from threading import Lock

14from typing import ClassVar

16import jsonpath_rw

17import selenium.common.exceptions

18import toml

19import yaml

20from jinja2 import Template

21from lxml import etree

22from PIL import Image

23from selenium.webdriver.common.by import By

25from ..version import VERSION

28class Base:

29 passcmd = "pass"

30 schema = yaml.safe_load(

31 importlib.resources.files("selenible.schema").joinpath("base.yaml").read_text()

32 )

34 def __init__(self):

35 self.lock = Lock()

36 self.step = False

37 self.save_every = False

38 self._driver = None

39 self.variables = {

40 "selenible_version": VERSION,

41 }

42 self.funcs = {}

43 self.log = getLogger(self.__class__.__name__)

44 self.browser_args = {}

46 @property

47 def driver(self):

48 if self._driver is None:

49 self._driver = self.boot_driver()

50 self.log.info("driver started")

51 self.variables["driver"] = self._driver.name

52 self.variables["desired_capabilities"] = self._driver.desired_capabilities

53 return self._driver

55 def get_options(self):

56 return {}

58 def boot_driver(self):

59 raise Exception("please implement")

61 def shutdown_driver(self):

62 if hasattr(self, "_driver") and self._driver is not None:

63 self._driver.close()

64 self._driver.quit()

65 self._driver = None

67 def printpdf(self, output_fn):

68 raise Exception("please implement")

70 def __del__(self):

71 self.shutdown_driver()

73 @classmethod

74 def load_modules(cls, modname):

75 log = getLogger(cls.__name__)

76 log.debug("load module %s", modname)

77 pfx = cls.__name__ + "_"

78 try:

79 mm = modname.rsplit(".", 1)

80 if len(mm) == 1: 80 ↛ 84line 80 didn't jump to line 84 because the condition on line 80 was always true

81 modfirst = "selenible.modules"

82 modlast = mm[0]

83 else:

84 modfirst = mm[0]

85 modlast = mm[1]

86 mod1 = __import__(modfirst, globals(), locals(), [modlast], 0)

87 mod = getattr(mod1, modlast)

88 except AttributeError:

89 log.debug("cannot import %s from %s", modlast, modfirst)

90 return

91 log.debug("names: %s", dir(mod))

92 mtd = []

93 for m in filter(lambda f: f.startswith(pfx), dir(mod)):

94 fn = getattr(mod, m)

95 if callable(fn): 95 ↛ 107line 95 didn't jump to line 107 because the condition on line 95 was always true

96 log.debug("register method: %s", m[len(pfx) :])

97 name = f"do_{m[len(pfx) :]}"

98 setattr(cls, name, fn)

99 funcname = m[len(pfx) :]

100 mtd.append(funcname)

101 scmname = f"{funcname}_schema"

102 if hasattr(mod, scmname):

103 scm = getattr(mod, scmname)

104 if isinstance(scm, dict): 104 ↛ 93line 104 didn't jump to line 93 because the condition on line 104 was always true

105 cls.schema["items"]["properties"][funcname] = scm

106 else:

107 log.warning("%s is not callable", fn)

108 if len(mtd) != 0:

109 log.debug("register methods: %s", "/".join(mtd))

110

111 def load_vars(self, fp):

112 self.variables.update(yaml.safe_load(fp))

113

114 def render(self, s):

115 return Template(s).render(self.variables)

116

117 def render_dict(self, d):

118 if isinstance(d, dict):

119 res = {}

120 for k, v in d.items():

121 res[k] = self.render_dict(v)

122 return res

123 elif isinstance(d, (list, tuple)):

124 return [self.render_dict(x) for x in d]

125 elif isinstance(d, str):

126 return self.render(d)

127 return d

128

129 def run(self, prog):

130 res = None

131 for cmd in prog:

132 self.log.debug("cmd %s", cmd)

133 res = self.run1(cmd)

134 if self.step: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 ans = input("step(q=exit, s=screenshot, c=continue, other=continue):")

136 if ans == "q":

137 break

138 elif ans == "s":

139 self.saveshot_image().show()

140 elif ans == "c":

141 self.step = False

142 if self.save_every: 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true

143 self.do_screenshot({})

144 return res

145

146 def run1(self, cmd):

147 withitem = self.render_dict(cmd.pop("with_items", None))

148 delay = cmd.pop("delay", 0)

149 if withitem is not None:

150 loopctl = self.render_dict(cmd.pop("loop_control", {}))

151 loopvar = loopctl.get("loop_var", "item")

152 loopiter = loopctl.get("loop_iter", "iter")

153 start = time.time()

154 if isinstance(withitem, dict) and "range" in withitem: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true

155 rg = withitem.get("range")

156 if isinstance(rg, (list, tuple)):

157 withitem = range(*rg)

158 else:

159 withitem = range(int(rg))

160 elif isinstance(withitem, str): 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true

161 withitem = self.variables.get(withitem, None)

162 self.log.info("start loop: %d times", len(withitem))

163 for i, j in enumerate(withitem):

164 self.variables[loopvar] = j

165 self.variables[loopiter] = i

166 self.log.info("loop by %d: %s", i, j)

167 res = self.run1(cmd.copy())

168 time.sleep(delay)

169 self.variables.pop(loopvar)

170 self.variables.pop(loopiter)

171 self.log.info("finish loop: %f second", time.time() - start)

172 return res

173 # cmd = self.render_dict(cmd)

174 name = self.render_dict(cmd.pop("name", ""))

175 condition = self.render_dict(cmd.pop("when", True))

176 ncondition = self.render_dict(cmd.pop("when_not", False))

177 if not self.eval_param(condition): 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true

178 self.log.info("skip(when) %s", repr(name))

179 return

180 if self.eval_param(ncondition): 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true

181 self.log.info("skip(when_not) %s", repr(name))

182 return

183 register = self.render_dict(cmd.pop("register", None))

184 ignoreerr = self.render_dict(cmd.pop("ignore_error", False))

185 if len(cmd) != 1: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true

186 raise Exception(f"too many parameters: {cmd.keys()}")

187 self.variables["env"] = os.environ

188 if self._driver is not None: 188 ↛ 190line 188 didn't jump to line 190 because the condition on line 188 was never true

189 # set driver related variables

190 for v in (

191 "current_url",

192 "page_source",

193 "title",

194 "window_handles",

195 "session_id",

196 "current_window_handle",

197 "capabilities",

198 "log_types",

199 "w3c",

200 ):

201 try:

202 self.variables[v] = getattr(self.driver, v)

203 except selenium.common.exceptions.WebDriverException:

204 self.log.info("cannot get attribute %s", v)

205 for v in ("cookies", "window_size", "window_position"):

206 try:

207 self.variables[v] = getattr(self.driver, "get_" + v)()

208 except selenium.common.exceptions.WebDriverException:

209 self.log.info("cannot get attribute %s", v)

210 self.variables["log"] = {}

211 try:

212 for logtype in self.driver.log_types:

213 self.variables["log"][logtype] = self.driver.get_log(logtype)

214 # phantomjs case

215 try:

216 if logtype == "har":

217 logdata = json.loads(

218 self.variables["log"][logtype][0]["message"]

219 )

220 self.variables["log"][logtype][0]["message"] = logdata

221 except (KeyError, IndexError, json.decoder.JSONDecodeError):

222 self.log.debug("log.har.0.message does not exists or not json")

223 except selenium.common.exceptions.WebDriverException:

224 self.log.info("cannot get log types")

225 for c in cmd: 225 ↛ exitline 225 didn't return from function 'run1' because the loop on line 225 didn't complete

226 mtdname = f"do_{c}"

227 mtdname2 = f"do2_{c}"

228 if hasattr(self, mtdname):

229 mtd = getattr(self, mtdname)

230 param = self.render_dict(cmd.get(c))

231 self.log.debug("%s %s %s", name, c, param)

232 self.log.info("start %s", repr(name))

233 start = time.time()

234 try:

235 with self.lock:

236 res = mtd(param)

237 except Exception as e:

238 if ignoreerr: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true

239 self.log.info("error(ignored): %s", e)

240 else:

241 self.log.error("error: %s", e)

242 raise

243 if register is not None:

244 self.log.debug("register %s = %s", register, res)

245 self.variables[register] = res

246 self.log.info("finish %s %f second", repr(name), time.time() - start)

247 elif hasattr(self, mtdname2):

248 # 1st class module

249 mtd = getattr(self, mtdname2)

250 param = cmd.get(c)

251 self.log.debug("%s %s %s", name, c, param)

252 with self.lock:

253 res = mtd(c, param)

254 if register is not None:

255 self.log.debug("register %s = %s", register, res)

256 self.variables[register] = res

257 else:

258 raise Exception(f"module not found: {c}")

259 time.sleep(delay)

260 return res

261

262 def do2_defun(self, funcname, params):

263 """

264 - name: define func1

265 defun:

266 name: func1

267 args: [a1, a2]

268 return: r

269 progn:

270 - name: hello

271 echo: "{{a1}} is {{a2}}"

272 - name: set-retval

273 var:

274 r: "hello"

275 - name: call func1

276 func1:

277 a1: xyz

278 a2: abc

279 register: rval1

280 - name: return value

281 echo: "{{rval1}}"

282 """

283 funcname = params.get("name")

284 args = params.get("args", [])

285 retvar = params.get("return", None)

286 progn = params.get("progn", [])

287 self.funcs[funcname] = (args, retvar, progn)

288 setattr(self, "do2_" + funcname, self.run_func)

289

290 def run_func(self, funcname, params):

291 params = self.render_dict(params)

292 args, retvar, progn = self.funcs.get(funcname, ([], None, None))

293 oldvars = self.variables

294 self.variables = copy.deepcopy(self.variables)

295 for a in args:

296 self.variables[a] = params.get(a)

297 self.log.debug("running %s", progn)

298 self.lock.release()

299 res = self.run(progn)

300 self.lock.acquire()

301 newvars = self.variables

302 self.variables = oldvars

303 if retvar is not None: 303 ↛ 306line 303 didn't jump to line 306 because the condition on line 303 was always true

304 self.log.debug("return val %s -> %s", retvar, newvars.get(retvar))

305 return newvars.get(retvar)

306 return res

307

308 @classmethod

309 def listmodule(cls):

310 pfx = ["do_", "do2_"]

311 res = {}

312 for x in dir(cls):

313 for p in pfx:

314 if x.startswith(p):

315 doc = inspect.getdoc(getattr(cls, x))

316 if doc is None:

317 doc = "(no document)"

318 res[x[len(p) :]] = doc

319 return res

320

321 def execute(self, script, args):

322 self.driver.execute_script(script, args)

323

324 def runcmd(

325 self, cmd, encoding="utf-8", stdin=subprocess.DEVNULL, stderr=subprocess.DEVNULL

326 ):

327 flag = False

328 if isinstance(cmd, str):

329 flag = True

330 self.log.debug("run(%s) %s", flag, cmd)

331 ret = subprocess.check_output(

332 cmd, stdin=stdin, stderr=stderr, shell=flag

333 ).decode(encoding)

334 self.log.debug("result: %s", ret)

335 return ret

336

337 def saveshot_image(self):

338 return Image.open(io.BytesIO(self.saveshot()))

339

340 def saveshot(self, fp=None):

341 data = self.driver.get_screenshot_as_png()

342 if fp is None: 342 ↛ 343line 342 didn't jump to line 343 because the condition on line 342 was never true

343 return data

344 elif isinstance(fp, str): 344 ↛ 348line 344 didn't jump to line 348 because the condition on line 344 was always true

345 with open(fp, "wb") as f:

346 f.write(data)

347 else:

348 fp.write(data)

349 return data

350

351 findmap: ClassVar = {

352 "id": By.ID,

353 "xpath": By.XPATH,

354 "linktext": By.LINK_TEXT,

355 "partlinktext": By.PARTIAL_LINK_TEXT,

356 "name": By.NAME,

357 "tag": By.TAG_NAME,

358 "class": By.CLASS_NAME,

359 "select": By.CSS_SELECTOR,

360 }

361

362 def removelocator(self, param):

363 res = copy.deepcopy(param)

364 res.pop("nth", None)

365 for k, v in self.findmap.items():

366 res.pop(k, None)

367 for v in filter(lambda f: not f.startswith("_"), dir(By)):

368 res.pop(v, None)

369 res.pop(v.lower(), None)

370 if getattr(By, v) in res:

371 res.pop(getattr(By, v), None)

372 return res

373

374 def getlocator(self, param):

375 for k, v in self.findmap.items():

376 if k in param:

377 return (v, param.get(k))

378 for v in filter(lambda f: not f.startswith("_"), dir(By)):

379 if v in param: 379 ↛ 380line 379 didn't jump to line 380 because the condition on line 379 was never true

380 return (getattr(By, v), param.get(v))

381 if v.lower() in param:

382 return (getattr(By, v), param.get(v.lower()))

383 if getattr(By, v) in param: 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true

384 return (getattr(By, v), param.get(getattr(By, v)))

385 return (None, None)

386

387 def findone(self, param):

388 k, v = self.getlocator(param)

389 if k is not None: 389 ↛ 391line 389 didn't jump to line 391 because the condition on line 389 was always true

390 return self.driver.find_element(k, v)

391 if param.get("active", False):

392 return self.driver.switch_to.active_element

393 return None

394

395 def findmany2one(self, param):

396 ret = self.findmany(param)

397 nth = param.get("nth", 0)

398 if isinstance(ret, (list, tuple)): 398 ↛ 404line 398 didn't jump to line 404 because the condition on line 398 was always true

399 self.log.debug("found %d elements. choose %d-th", len(ret), nth)

400 if len(ret) > nth:

401 return ret[nth]

402 else:

403 return None

404 return ret

405

406 def findmany(self, param):

407 k, v = self.getlocator(param)

408 if k is not None:

409 return self.driver.find_elements(k, v)

410 if param.get("active", False): 410 ↛ 411line 410 didn't jump to line 411 because the condition on line 410 was never true

411 return [self.driver.switch_to.active_element]

412 return []

413

414 def getvalue(self, param):

415 if isinstance(param, str): 415 ↛ 416line 415 didn't jump to line 416 because the condition on line 415 was never true

416 return param

417 encoding = param.get("encoding", "utf-8")

418 if "text" in param:

419 return param.get("text")

420 elif "password" in param: 420 ↛ 421line 420 didn't jump to line 421 because the condition on line 420 was never true

421 label = param.get("password")

422 return self.runcmd([self.passcmd, label], encoding).strip()

423 elif "pipe" in param: 423 ↛ 424line 423 didn't jump to line 424 because the condition on line 423 was never true

424 cmd = param.get("pipe")

425 return self.runcmd(cmd, encoding).strip()

426 elif "yaml" in param: 426 ↛ 427line 426 didn't jump to line 427 because the condition on line 426 was never true

427 p = param.get("yaml")

428 with open(p.get("file")) as f:

429 data = yaml.safe_load(f)

430 return jsonpath_rw.parse(p.get("path", "*")).find(data)[0].value

431 elif "json" in param: 431 ↛ 432line 431 didn't jump to line 432 because the condition on line 431 was never true

432 p = param.get("json")

433 with open(p.get("file")) as f:

434 data = json.load(f)

435 return jsonpath_rw.parse(p.get("path", "*")).find(data)[0].value

436 elif "toml" in param: 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true

437 p = param.get("toml")

438 with open(p.get("file")) as f:

439 data = toml.load(f)

440 return jsonpath_rw.parse(p.get("path", "*")).find(data)[0].value

441 elif "input" in param: 441 ↛ 442line 441 didn't jump to line 442 because the condition on line 441 was never true

442 return input(param.get("input"))

443 elif "input_password" in param: 443 ↛ 444line 443 didn't jump to line 444 because the condition on line 443 was never true

444 return getpass.getpass(param.get("input_password"))

445 elif "input_multiline" in param: 445 ↛ 446line 445 didn't jump to line 446 because the condition on line 445 was never true

446 print(param.get("input_multiline"))

447 res = sys.stdin.read()

448 sys.stdin.seek(0)

449 return res

450 return None

451

452 def eval_param(self, param):

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

454 return [self.eval_param(x) for x in param]

455 elif isinstance(param, dict):

456 res = []

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

458 if k in ("eq", "equals", "==", "is"):

459 res.append(len(set(self.eval_param(v))) == 1)

460 elif k in ("neq", "not_equals", "!=", "is_not"): 460 ↛ 462line 460 didn't jump to line 462 because the condition on line 460 was always true

461 res.append(len(set(self.eval_param(v))) >= 2)

462 elif k in ("not"):

463 res.append(not self.eval_param(v))

464 elif k in ("and", "&", "&&"):

465 res.append(

466 functools.reduce(lambda a, b: a and b, self.eval_param(v))

467 )

468 elif k in ("or", "|", "||"):

469 res.append(

470 functools.reduce(lambda a, b: a or b, self.eval_param(v))

471 )

472 elif k in ("xor", "^"):

473 res.append(

474 functools.reduce(

475 lambda a, b: bool(a) ^ bool(b), self.eval_param(v)

476 )

477 )

478 elif k in ("add", "sum", "plus", "+"):

479 res.append(functools.reduce(lambda a, b: a + b, self.eval_param(v)))

480 elif k in ("sub", "minus", "-"):

481 res.append(functools.reduce(lambda a, b: a - b, self.eval_param(v)))

482 elif k in ("mul", "times", "*"):

483 res.append(functools.reduce(lambda a, b: a * b, self.eval_param(v)))

484 elif k in ("div", "/"):

485 res.append(functools.reduce(lambda a, b: a / b, self.eval_param(v)))

486 elif k in ("selected",):

487 for e in self.findmany(v):

488 res.append(e.is_selected())

489 elif k in ("not_selected", "unselected"):

490 for e in self.findmany(v):

491 res.append(not e.is_selected())

492 elif k in ("enabled",):

493 for e in self.findmany(v):

494 res.append(e.is_enabled())

495 elif k in ("not_enabled", "disabled"):

496 for e in self.findmany(v):

497 res.append(not e.is_enabled())

498 elif k in ("displayed",):

499 for e in self.findmany(v):

500 res.append(e.is_displayed())

501 elif k in ("not_displayed", "undisplayed"):

502 for e in self.findmany(v):

503 res.append(not e.is_displayed())

504 elif k in ("defined",):

505 if isinstance(v, (tuple, list)):

506 res.extend([x in self.variables for x in v])

507 elif isinstance(v, str):

508 res.append(v in self.variables)

509 else:

510 raise Exception(f"invalid argument: {v}")

511 elif k in ("not_defined", "undefined"):

512 if isinstance(v, (tuple, list)):

513 res.extend([x not in self.variables for x in v])

514 elif isinstance(v, str):

515 res.append(v not in self.variables)

516 else:

517 raise Exception(f"invalid argument: {v}")

518 else:

519 raise Exception(f"operator not supported: {k} ({v})")

520 return functools.reduce(lambda a, b: a and b, res)

521 return param

522

523 def return_element(self, param, elem):

524 if elem is None: 524 ↛ 525line 524 didn't jump to line 525 because the condition on line 524 was never true

525 return elem

526 if not param.get("parseHTML", False): 526 ↛ 528line 526 didn't jump to line 528 because the condition on line 526 was always true

527 return elem

528 if isinstance(elem, (tuple, list)):

529 return [etree.fromstring(x.get_attribute("outerHTML")) for x in elem]

530 elif isinstance(elem, str):

531 return etree.fromstring(elem)

532 return etree.fromstring(elem.get_attribute("outerHTML"))