| Total Complexity | 10 |
| Total Lines | 71 |
| Duplicated Lines | 0 % |
| Changes | 0 | ||
| 1 | # -*- coding: utf-8 -*- |
||
|
|
|||
| 2 | |||
| 3 | import time |
||
| 4 | from collections import OrderedDict |
||
| 5 | import requests |
||
| 6 | from defusedxml import ElementTree |
||
| 7 | |||
| 8 | class TclCheckMixin: |
||
| 9 | def do_check(self, https=True, timeout=10, max_tries=5): |
||
| 10 | protocol = "https://" if https else "http://" |
||
| 11 | url = protocol + self.g2master + "/check.php" |
||
| 12 | params = OrderedDict() |
||
| 13 | params["id"] = self.serid |
||
| 14 | params["curef"] = self.curef |
||
| 15 | params["fv"] = self.fv |
||
| 16 | params["mode"] = self.mode.value |
||
| 17 | params["type"] = self.ftype |
||
| 18 | params["cltp"] = self.cltp.value |
||
| 19 | params["cktp"] = self.cktp.value |
||
| 20 | params["rtd"] = self.rtd.value |
||
| 21 | params["chnl"] = self.chnl.value |
||
| 22 | #params["osvs"] = self.osvs |
||
| 23 | #params["ckot"] = self.ckot.value |
||
| 24 | |||
| 25 | last_response = None |
||
| 26 | for num_try in range(0, max_tries): |
||
| 27 | try: |
||
| 28 | reqtime_start = time.perf_counter() |
||
| 29 | req = self.sess.get(url, params=params, timeout=timeout) |
||
| 30 | reqtime = time.perf_counter() - reqtime_start |
||
| 31 | reqtime_avg = self.check_time_avg() |
||
| 32 | self.check_time_add(reqtime) |
||
| 33 | last_response = req |
||
| 34 | if req.status_code == 200: |
||
| 35 | self.master_server_vote_on_time(reqtime, reqtime_avg) |
||
| 36 | req.encoding = "utf-8" # Force encoding as server doesn't give one |
||
|
1 ignored issue
–
show
|
|||
| 37 | self.write_dump(req.text) |
||
| 38 | return req.text |
||
| 39 | elif req.status_code == 204: |
||
| 40 | self.master_server_vote_on_time(reqtime, reqtime_avg) |
||
| 41 | raise requests.exceptions.HTTPError("No update available.", response=req) |
||
|
1 ignored issue
–
show
|
|||
| 42 | elif req.status_code == 404: |
||
| 43 | self.master_server_vote_on_time(reqtime, reqtime_avg) |
||
| 44 | raise requests.exceptions.HTTPError("No data for requested CUREF/FV combination.", response=req) |
||
|
1 ignored issue
–
show
|
|||
| 45 | elif req.status_code not in [500, 502, 503]: |
||
| 46 | self.master_server_downvote() |
||
| 47 | req.raise_for_status() |
||
| 48 | raise requests.exceptions.HTTPError("HTTP {}.".format(req.status_code), response=req) |
||
|
1 ignored issue
–
show
|
|||
| 49 | except requests.exceptions.Timeout: |
||
| 50 | pass |
||
| 51 | # Something went wrong, try a different server |
||
| 52 | self.master_server_downvote() |
||
| 53 | self.g2master = self.get_master_server() |
||
| 54 | protocol = "https://" if https else "http://" |
||
| 55 | url = protocol + self.g2master + "/check.php" |
||
| 56 | raise requests.exceptions.RetryError("Max tries ({}) reached.".format(max_tries), response=last_response) |
||
|
1 ignored issue
–
show
|
|||
| 57 | |||
| 58 | @staticmethod |
||
| 59 | def parse_check(xmlstr): |
||
| 60 | root = ElementTree.fromstring(xmlstr) |
||
| 61 | curef = root.find("CUREF").text |
||
| 62 | fv = root.find("VERSION").find("FV").text |
||
| 63 | tv = root.find("VERSION").find("TV").text |
||
| 64 | fw_id = root.find("FIRMWARE").find("FW_ID").text |
||
| 65 | fileinfo = root.find("FIRMWARE").find("FILESET").find("FILE") |
||
| 66 | fileid = fileinfo.find("FILE_ID").text |
||
| 67 | filename = fileinfo.find("FILENAME").text |
||
| 68 | filesize = fileinfo.find("SIZE").text |
||
| 69 | filehash = fileinfo.find("CHECKSUM").text |
||
| 70 | return curef, fv, tv, fw_id, fileid, filename, filesize, filehash |
||
| 71 |
The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods:
If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions.