mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-05 08:15:58 +08:00
firestar5683
86 lines
1.8 KiB
Python
86 lines
1.8 KiB
Python
import http.server
|
|
import socket
|
|
|
|
|
|
class MockResponse:
|
|
def __init__(self, json, status_code):
|
|
self.json = json
|
|
self.text = json
|
|
self.status_code = status_code
|
|
|
|
|
|
class EchoSocket:
|
|
def __init__(self, port):
|
|
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
self.socket.bind(('127.0.0.1', port))
|
|
self.socket.listen(1)
|
|
|
|
def run(self):
|
|
conn, _ = self.socket.accept()
|
|
conn.settimeout(5.0)
|
|
|
|
try:
|
|
while True:
|
|
try:
|
|
data = conn.recv(4096)
|
|
except TimeoutError:
|
|
break
|
|
if data:
|
|
print(f'EchoSocket got {data}')
|
|
conn.sendall(data)
|
|
else:
|
|
break
|
|
finally:
|
|
try:
|
|
conn.shutdown(0)
|
|
except OSError:
|
|
pass
|
|
conn.close()
|
|
try:
|
|
self.socket.shutdown(0)
|
|
except OSError:
|
|
pass
|
|
self.socket.close()
|
|
|
|
|
|
class MockApi:
|
|
def __init__(self, dongle_id):
|
|
pass
|
|
|
|
def get_token(self):
|
|
return "fake-token"
|
|
|
|
|
|
class MockWebsocket:
|
|
def __init__(self, recv_queue, send_queue):
|
|
self.recv_queue = recv_queue
|
|
self.send_queue = send_queue
|
|
self.sock, self._signal_sock = socket.socketpair()
|
|
|
|
def queue_recv(self, data):
|
|
self.recv_queue.put_nowait(data)
|
|
self._signal_sock.send(b"1")
|
|
|
|
def recv(self):
|
|
self.sock.recv(1)
|
|
data = self.recv_queue.get()
|
|
if isinstance(data, Exception):
|
|
raise data
|
|
return data
|
|
|
|
def send(self, data, opcode):
|
|
self.send_queue.put_nowait((data, opcode))
|
|
|
|
def close(self):
|
|
self.sock.close()
|
|
self._signal_sock.close()
|
|
|
|
|
|
class HTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
|
|
def do_PUT(self):
|
|
length = int(self.headers['Content-Length'])
|
|
self.rfile.read(length)
|
|
self.send_response(201, "Created")
|
|
self.end_headers()
|