Skip to content

Commit 09d1553

Browse files
KS-HTKKS-HTK
committed
added a mock_server for testing
 without a real module fixed minor issues in fluepdot.py with set and unset pixel fixed issue in fluepdot.py where get_pixel was not returning a bool
1 parent 0c16e88 commit 09d1553

2 files changed

Lines changed: 185 additions & 5 deletions

File tree

‎src/fluepdot/fluepdot.py‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -68,9 +68,9 @@ def get_frame() -> List[str]:
6868

6969

7070
def get_pixel(x: int = 0, y: int = 0) -> bool:
71-
r = _get(pixelURL, get={x, y})
72-
print(r)
73-
return False
71+
r = _get(pixelURL, get={"x": x, "y": y})
72+
rtn = True if r.text == "X" else False if r.text == " " else None
73+
return rtn
7474

7575

7676
def get_fonts() -> None:
@@ -105,11 +105,11 @@ def post_frame(frame: List[List[bool]]) -> Response:
105105

106106

107107
def set_pixel(x: int = 0, y: int = 0) -> Response:
108-
return _post(pixelURL, get={x, y})
108+
return _post(pixelURL, get={"x": x, "y": y})
109109

110110

111111
def unset_pixel(x: int = 0, y: int = 0) -> Response:
112-
return _delete(pixelURL, get={x, y})
112+
return _delete(pixelURL, get={"x": x, "y": y})
113113

114114

115115
def set_mode(mode: Mode = Mode.FULL) -> Response:

‎src/fluepdot/mock_server.py‎

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
# mock server for the fluepdot api
2+
from fluepdot import Mode
3+
from http.server import BaseHTTPRequestHandler, HTTPServer
4+
from urllib.parse import urlparse, parse_qs
5+
import socket
6+
import logging
7+
from threading import Thread
8+
import re
9+
10+
# Third-party imports...
11+
import requests
12+
13+
class MockServerRequestHandler(BaseHTTPRequestHandler):
14+
FRAME_PATTERN = re.compile(r'/framebuffer')
15+
PIXEL_PATTERN = re.compile(r'/pixel')
16+
TEXT_PATTERN = re.compile(r'/framebuffer/text')
17+
FONT_PATTERN = re.compile(r'/fonts')
18+
MODE_PATTERN = re.compile(r'/rendering/mode')
19+
TIMEING_PATTERN = re.compile(r'/rendering/timings')
20+
21+
framebuffer = [115*[False] for _ in range(16)]
22+
rendermode = Mode.FULL
23+
24+
def do_GET(self):
25+
# /framebuffer
26+
# /pixel x y
27+
# /fonts
28+
# /rendering/mode
29+
# /rendering/timings
30+
# Process an HTTP GET request and return a response with an HTTP 200 status.
31+
if re.search(self.FRAME_PATTERN, self.path):
32+
if re.search(self.TEXT_PATTERN, self.path):
33+
# invalid endpoint (text does not support get)
34+
self.send_response(requests.codes.not_found)
35+
self.end_headers()
36+
else:
37+
msg = '/n'.join(''.join('X' if x else ' ' for x in i) for i in self.framebuffer)
38+
self.send_response(requests.codes.ok)
39+
self.end_headers()
40+
self.wfile.write(msg.encode('utf-8'))
41+
elif re.search(self.PIXEL_PATTERN, self.path):
42+
query = parse_qs(urlparse(self.path).query)
43+
try:
44+
x = int(query['x'][0])
45+
y = int(query['y'][0])
46+
msg = 'X' if self.framebuffer[y][x] else ' '
47+
self.send_response(requests.codes.ok)
48+
self.end_headers()
49+
self.wfile.write(msg.encode('utf-8'))
50+
except Exception:
51+
logging.debug(' invalid pixel get %s', self.path)
52+
self.send_response(requests.codes.bad)
53+
self.end_headers()
54+
elif re.search(self.FONT_PATTERN, self.path):
55+
msg = "mock font for fluepdot\nmock_font\nsecond font for mock server\nmock_font2"
56+
self.send_response(requests.codes.ok)
57+
self.end_headers()
58+
self.wfile.write(msg.encode('utf-8'))
59+
elif re.search(self.MODE_PATTERN, self.path):
60+
logging.debug(' mode GET endpoint %s', self.path)
61+
self.send_response(requests.codes.ok)
62+
self.end_headers()
63+
self.wfile.write(self.rendermode.value.encode('ascii'))
64+
else:
65+
logging.debug(' invalid GET endpoint %s', self.path)
66+
self.send_response(requests.codes.bad)
67+
self.end_headers()
68+
return
69+
70+
def do_PUT(self):
71+
# Process an HTTP PUT request and return a response with an HTTP 200 status.
72+
# /rendering/mode
73+
if re.search(self.MODE_PATTERN, self.path):
74+
self.rendermode = Mode(int(self.rfile.read(int(self.headers.get('content-length')))))
75+
logging.debug(' mode set to %s', self.rendermode)
76+
self.send_response(requests.codes.ok)
77+
self.end_headers()
78+
else:
79+
logging.debug(' invalid PUT endpoint %s', self.path)
80+
self.send_response(requests.codes.bad)
81+
self.end_headers()
82+
return
83+
84+
def do_POST(self):
85+
# Process an HTTP POST request and return a response with an HTTP 200 status.
86+
# /framebuffer
87+
# /framebuffer/text p=text g=x,y,font
88+
# /pixel x y
89+
# /rendering/timings
90+
logging.debug(' POST endpoint %s', self.path)
91+
if re.search(self.FRAME_PATTERN, self.path):
92+
if re.search(self.TEXT_PATTERN, self.path):
93+
query = parse_qs(urlparse(self.path).query)
94+
try:
95+
x = int(query['x'][0])
96+
y = int(query['y'][0])
97+
font = query['font'][0]
98+
text = self.rfile.read(int(self.headers.get('content-length'))).decode('utf-8')
99+
logging.debug(' text %s at %d,%d with font %s', text, x, y, font)
100+
self.send_response(requests.codes.ok)
101+
self.end_headers()
102+
except Exception:
103+
logging.debug(' invalid POST endpoint %s', self.path)
104+
self.send_response(requests.codes.bad)
105+
self.end_headers()
106+
else:
107+
logging.debug(' framebuffer endpoint %s', self.path)
108+
data = self.rfile.read(int(self.headers.get('content-length')))
109+
logging.debug(' data %s', data)
110+
self.framebuffer = [[True if x == 'X' else False for x in i] for i in data.decode('utf-8').split('\n')]
111+
self.send_response(requests.codes.ok)
112+
self.end_headers()
113+
elif re.search(self.PIXEL_PATTERN, self.path):
114+
query = parse_qs(urlparse(self.path).query)
115+
try:
116+
x = int(query['x'][0])
117+
y = int(query['y'][0])
118+
self.framebuffer[y][x] = True
119+
logging.debug(' pixel set %s, %s', x, y)
120+
self.send_response(requests.codes.ok)
121+
self.end_headers()
122+
except Exception:
123+
logging.debug(' invalid pixel set %s', self.path)
124+
self.send_response(requests.codes.bad)
125+
self.end_headers()
126+
elif re.search(self.TIMEING_PATTERN, self.path):
127+
# TODO: implement this correctly
128+
print('timing endpoint', self.path)
129+
self.send_response(requests.codes.unsupported)
130+
self.end_headers()
131+
else:
132+
logging.debug(' invalid POST endpoint %s', self.path)
133+
self.send_response(requests.codes.bad)
134+
self.end_headers()
135+
return
136+
137+
def do_DELETE(self):
138+
# Process an HTTP DELETE request and return a response with an HTTP 200 status.
139+
# /pixel x y
140+
if re.search(self.PIXEL_PATTERN, self.path):
141+
query = parse_qs(urlparse(self.path).query)
142+
try:
143+
x = int(query['x'][0])
144+
y = int(query['y'][0])
145+
self.framebuffer[y][x] = False
146+
logging.debug(' pixel unset %s, %s', x, y)
147+
self.send_response(requests.codes.ok)
148+
self.end_headers()
149+
except Exception:
150+
logging.debug(' invalid pixel unset %s', self.path)
151+
self.send_response(requests.codes.bad)
152+
self.end_headers()
153+
else:
154+
logging.debug(' invalid DELETE endpoint %s', self.path)
155+
self.send_response(requests.codes.bad)
156+
self.end_headers()
157+
return
158+
159+
160+
class TestMockServer(object):
161+
@classmethod
162+
def setup_class(cls):
163+
# Configure mock server.
164+
cls.mock_server = HTTPServer(('localhost', 8080), MockServerRequestHandler)
165+
166+
# Start running mock server in a separate thread.
167+
# Daemon threads automatically shut down when the main process exits.
168+
cls.mock_server_thread = Thread(target=cls.mock_server.serve_forever, daemon=True)
169+
cls.mock_server_thread.start()
170+
171+
if __name__ == "__main__":
172+
logging.basicConfig(level=logging.DEBUG)
173+
logging.debug(' starting mock server')
174+
server = TestMockServer().setup_class()
175+
try:
176+
while True:
177+
pass
178+
except KeyboardInterrupt:
179+
logging.debug(' stopping mock server')
180+

0 commit comments

Comments
 (0)