| 1259 | |
| 1260 | |
| 1261 | class HTTPApi: |
| 1262 | def __init__(self, addr: Tuple[str, int]): |
| 1263 | self.host, self.port = addr |
| 1264 | self.api = "http://%s:%d" % (self.host, self.port) |
| 1265 | |
| 1266 | @classmethod |
| 1267 | def from_target(cls, target): |
| 1268 | ws_url = urllib.parse.urlparse(target.websocket_url) |
| 1269 | inst = cls((ws_url.hostname, ws_url.port)) |
| 1270 | return inst |
| 1271 | |
| 1272 | async def get(self, endpoint: str): |
| 1273 | return await self._request(endpoint) |
| 1274 | |
| 1275 | async def post(self, endpoint, data): |
| 1276 | return await self._request(endpoint, data) |
| 1277 | |
| 1278 | async def _request(self, endpoint, method: str = "GET", data: dict = None): |
| 1279 | url = urllib.parse.urljoin( |
| 1280 | self.api, f"json/{endpoint}" if endpoint else "/json" |
| 1281 | ) |
| 1282 | if data and method.lower() == "get": |
| 1283 | raise ValueError("GET requests cannot contain data") |
| 1284 | if not url: |
| 1285 | url = self.api + endpoint |
| 1286 | request = urllib.request.Request(url) |
| 1287 | request.method = method |
| 1288 | request.data = None |
| 1289 | if data: |
| 1290 | request.data = json.dumps(data).encode("utf-8") |
| 1291 | |
| 1292 | response = await asyncio.get_running_loop().run_in_executor( |
| 1293 | None, lambda: urllib.request.urlopen(request, timeout=10) |
| 1294 | ) |
| 1295 | return json.loads(response.read()) |
| 1296 | |
| 1297 | |
| 1298 | atexit.register(deconstruct_browser) |
no outgoing calls
no test coverage detected
searching dependent graphs…