[Revert if incompatible] ?time=&name= is for device, not sensor

main.py

@@ -1,6 +1,8 @@
 import socket
-from urllib.request import urlopen, Request
+import urllib3
 
+import logging
+import unittest
 from unittest import TestCase
 
 from fixtures import device
@@ -22,29 +24,73 @@ class TestTCP(TestCase):
         self._wait_for_ok()
 
     def test_json(self) -> None:
-        raise NotImplemented
+        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.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._send_and_check(Request(
+        self._check_status(self.http.request(
+            'GET',
             HTTP_URL + '/get?' + device.http_req_data(),
         ))
 
     def test_post(self) -> None:
-        self._send_and_check(Request(
+        self._check_status(self.http.request(
+            'POST',
             HTTP_URL + '/post',
-            data=device.http_req_data().encode(),
+            body=device.http_req_data(),
         ))
 
     def test_json(self) -> None:
-        raise NotImplemented
+        self._check_status(self.http.request(
+            'POST',
+            HTTP_URL + '/json',
+            body=device.json_req_data(),
+        ))
 
-    def _send_and_check(self, req: Request) -> None:
+    def _check_status(self, resp: urllib3.HTTPResponse) -> None:
-        with urlopen(req) as resp:
+        logging.debug(resp.headers.items())
-            self.assertEqual(resp.status, 200)
+        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,7 +1,8 @@
+import json
 from urllib.parse import quote
 
 from dataclasses import dataclass
-from typing import Collection
+from typing import Collection, Any
 
 
 type Numeric = int | float
@@ -71,11 +72,18 @@ 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)
 
@@ -89,4 +97,45 @@ class Device:
         return res
 
     def json_req_data(self) -> str:
-        raise NotImplemented
+        '''Generate a JSON format request body'''
+
+        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

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