(self, key: str, reply: QtNetwork.QNetworkReply)
| 177 | reply.finished.connect(lambda key=key, reply=reply: self._on_reply_finished(key, reply)) |
| 178 | |
| 179 | def _on_reply_finished(self, key: str, reply: QtNetwork.QNetworkReply): |
| 180 | try: |
| 181 | if reply.error() != QtNetwork.QNetworkReply.NetworkError.NoError: |
| 182 | # keep the first network-ish error but remain non-fatal overall |
| 183 | if self._result["error"] is None: |
| 184 | self._result["error"] = reply.errorString() |
| 185 | return |
| 186 | |
| 187 | payload = bytes(reply.readAll()) |
| 188 | latest_version = json.loads(payload.decode("utf-8"))["info"]["version"] |
| 189 | |
| 190 | if key == "dlc": |
| 191 | self._result["latest_version"] = latest_version |
| 192 | self._result["is_latest"] = self._is_up_to_date(self._dlc_version, latest_version) |
| 193 | else: |
| 194 | self._result["latest_plugin_version"] = latest_version |
| 195 | self._result["is_latest_plugin"] = self._is_up_to_date(self._plugin_version, latest_version) |
| 196 | except Exception as e: |
| 197 | if self._result["error"] is None: |
| 198 | self._result["error"] = str(e) |
| 199 | finally: |
| 200 | reply.deleteLater() |
| 201 | self._replies.pop(key, None) |
| 202 | |
| 203 | if self._running and not self._replies: |
| 204 | self._finish() |
| 205 | |
| 206 | def _on_timeout(self): |
| 207 | if not self._running: |
no test coverage detected