MCPcopy Create free account
hub / github.com/GlobalTechInfo/MEGA-MD / set_mini_server

Method set_mini_server

lib/speed.py:1366–1418  ·  view source on GitHub ↗

Instead of querying for a list of servers, set a link to a speedtest mini server

(self, server)

Source from the content-addressed store, hash-verified

1364 return self.servers
1365
1366 def set_mini_server(self, server):
1367 """Instead of querying for a list of servers, set a link to a
1368 speedtest mini server
1369 """
1370
1371 urlparts = urlparse(server)
1372
1373 name, ext = os.path.splitext(urlparts[2])
1374 if ext:
1375 url = os.path.dirname(server)
1376 else:
1377 url = server
1378
1379 request = build_request(url)
1380 uh, e = catch_request(request, opener=self._opener)
1381 if e:
1382 raise SpeedtestMiniConnectFailure('Failed to connect to %s' %
1383 server)
1384 else:
1385 text = uh.read()
1386 uh.close()
1387
1388 extension = re.findall('upload_?[Ee]xtension: "([^"]+)"',
1389 text.decode())
1390 if not extension:
1391 for ext in ['php', 'asp', 'aspx', 'jsp']:
1392 try:
1393 f = self._opener.open(
1394 '%s/speedtest/upload.%s' % (url, ext)
1395 )
1396 except Exception:
1397 pass
1398 else:
1399 data = f.read().strip().decode()
1400 if (f.code == 200 and
1401 len(data.splitlines()) == 1 and
1402 re.match('size=[0-9]', data)):
1403 extension = [ext]
1404 break
1405 if not urlparts or not extension:
1406 raise InvalidSpeedtestMiniServer('Invalid Speedtest Mini Server: '
1407 '%s' % server)
1408
1409 self.servers = [{
1410 'sponsor': 'Speedtest Mini',
1411 'name': urlparts[1],
1412 'd': 0,
1413 'url': '%s/speedtest/upload.%s' % (url.rstrip('/'), extension[0]),
1414 'latency': 0,
1415 'id': 0
1416 }]
1417
1418 return self.servers
1419
1420 def get_closest_servers(self, limit=5):
1421 """Limit servers to the closest speedtest.net servers based on

Callers 1

shellFunction · 0.95

Calls 6

build_requestFunction · 0.85
catch_requestFunction · 0.85
readMethod · 0.80
closeMethod · 0.80

Tested by

no test coverage detected