main.py

@@ -1,8 +1,6 @@
 import socket
-import urllib3
+from urllib.request import urlopen, Request
 
-import logging
-import unittest
 from unittest import TestCase
 
 from fixtures import device
@@ -24,73 +22,29 @@ class TestTCP(TestCase):
         self._wait_for_ok()
 
     def test_json(self) -> None:
-        self.sock.sendall(device.json_req_data().encode())
+        raise NotImplemented
-        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.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(
+        self._send_and_check(Request(
-            'GET',
             HTTP_URL + '/get?' + device.http_req_data(),
         ))
 
     def test_post(self) -> None:
-        self._check_status(self.http.request(
+        self._send_and_check(Request(
-            'POST',
             HTTP_URL + '/post',
-            body=device.http_req_data(),
+            data=device.http_req_data().encode(),
         ))
 
     def test_json(self) -> None:
-        self._check_status(self.http.request(
+        raise NotImplemented
-            'POST',
-            HTTP_URL + '/json',
-            body=device.json_req_data(),
-        ))
 
-    def _check_status(self, resp: urllib3.HTTPResponse) -> None:
+    def _send_and_check(self, req: Request) -> None:
-        logging.debug(resp.headers.items())
+        with urlopen(req) as resp:
-        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()

models.py

@@ -1,8 +1,7 @@
-import json
 from urllib.parse import quote
 
 from dataclasses import dataclass
-from typing import Collection, Any
+from typing import Collection
 
 
 type Numeric = int | float
@@ -72,18 +71,11 @@ class Device:
 
         for s in self.sensors:
             res += f'&{s.mac}={s.value}'
-            # if s.time is not None:
-            #     res += f'&time={s.time}'
-            # if s.name is not None:
-            #     res += f'&name={quote(s.name)}'
-
-        if self.sensors:
-            s = self.sensors[0]
             if s.time is not None:
                 res += f'&time={s.time}'
+            if s.name is not None:
+                res += f'&name={quote(s.name)}'
 
-        if self.name is not None:
-            res += '&name=' + quote(self.name)
         if self.owner is not None:
             res += '&owner=' + quote(self.owner)
 
@@ -97,45 +89,4 @@ class Device:
         return res
 
     def json_req_data(self) -> str:
-        '''Generate a JSON format request body'''
+        raise NotImplemented
-
-        obj = {
-            'devices': [
-                self._json_device_obj(),
-            ]
-        }
-
-        return json.dumps(obj)
-
-    def _json_device_obj(self) -> dict[str, Any]:
-        '''Create a device JSON object for json_req_data().
-        Moved to a separate function to allow performing tests
-        with multiple devices in a request at once.'''
-
-        dev = {
-            'mac': self.mac,
-            'sensors': [],
-        }
-
-        if self.name is not None:
-            dev['name'] = self.name
-        if self.owner is not None:
-            dev['owner'] = self.owner
-        if self.geopos is not None:
-            dev['lat'] = self.geopos.lat
-            dev['lon'] = self.geopos.lon
-            dev['alt'] = self.geopos.alt
-
-        sl: list[dict[str, Any]] = dev['sensors']
-        for s in self.sensors:
-            sobj = {
-                'id': s.mac,
-                'value': s.value,
-            }
-            if s.time is not None:
-                sobj['time'] = s.time
-            if s.name is not None:
-                sobj['name'] = s.name
-            sl.append(sobj)
-
-        return dev

requirements.txt

@@ -1 +0,0 @@
-urllib3