tests/main.py

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()