105 lines
2.7 KiB
Python
105 lines
2.7 KiB
Python
import socket
|
|
import urllib3
|
|
|
|
import logging
|
|
import unittest
|
|
from unittest import TestCase
|
|
|
|
from fixtures import device
|
|
|
|
|
|
TCP_ADDR = ('127.0.0.1', 8283)
|
|
UDP_ADDR = ('127.0.0.1', 8283)
|
|
HTTP_URL = 'http://127.0.0.1:8080'
|
|
|
|
|
|
class TestTCP(TestCase):
|
|
|
|
def setUp(self) -> None:
|
|
self.skipTest('Not implemented yet')
|
|
self.sock = socket.socket()
|
|
self.sock.connect(TCP_ADDR)
|
|
|
|
def test_nm(self) -> None:
|
|
self.sock.sendall(device.nm_req_data().encode())
|
|
self._wait_for_ok()
|
|
|
|
def test_json(self) -> None:
|
|
self.sock.sendall(device.json_req_data().encode())
|
|
self._wait_for_ok()
|
|
|
|
def _wait_for_ok(self) -> None:
|
|
data = self.sock.recv(1024)
|
|
self.assertEqual(data, b'OK')
|
|
|
|
def tearDown(self) -> None:
|
|
self.sock.close()
|
|
|
|
|
|
class TestUDP(TestCase):
|
|
|
|
def setUp(self) -> None:
|
|
self.skipTest('Useless until no client API')
|
|
self.sock = socket.socket(type=socket.SOCK_DGRAM)
|
|
|
|
def test_nm(self) -> None:
|
|
self.sock.sendto(device.nm_req_data().encode(), UDP_ADDR)
|
|
# aaand we can not check if server accepted the data
|
|
# so this TestCase is absolutely useless
|
|
# until client API is implemented
|
|
|
|
def test_json(self) -> None:
|
|
self.sock.sendto(device.json_req_data().encode(), UDP_ADDR)
|
|
|
|
def tearDown(self) -> None:
|
|
self.sock.close()
|
|
|
|
|
|
class TestHTTP(TestCase):
|
|
|
|
def setUp(self) -> None:
|
|
self.http = urllib3.PoolManager()
|
|
|
|
def test_get(self) -> None:
|
|
self._check_status(self.http.request(
|
|
'GET',
|
|
HTTP_URL + '/get?' + device.http_req_data(),
|
|
))
|
|
|
|
def test_post(self) -> None:
|
|
self._check_status(self.http.request(
|
|
'POST',
|
|
HTTP_URL + '/post',
|
|
body=device.http_req_data(),
|
|
headers={
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
},
|
|
))
|
|
|
|
def test_json(self) -> None:
|
|
self.skipTest('Not implemented yet')
|
|
self._check_status(self.http.request(
|
|
'POST',
|
|
HTTP_URL + '/json',
|
|
body=device.json_req_data(),
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
},
|
|
))
|
|
|
|
def _check_status(self, resp: urllib3.HTTPResponse) -> None:
|
|
logging.debug(resp.headers.items())
|
|
logging.debug(resp.data)
|
|
if resp.status == 400:
|
|
logging.warning(' x-full-error = %s', resp.getheader('x-full-error'))
|
|
self.assertEqual(resp.status, 200)
|
|
|
|
def tearDown(self) -> None:
|
|
self.http.clear()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import os
|
|
if os.getenv('DEBUG', '0') not in {'0', 'false'}:
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
unittest.main()
|