test: minimally functional test_storage

diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py
index 5613bea455e3b2e9d66be24b173a10fddc325b35..0d02b082c18688777aa9fa4a7bbbbc0ae3727276 100755 (executable)
--- a/gps303/gps303proto.py
+++ b/gps303/gps303proto.py
@@ -85,6 +85,10 @@ class DecodeError(Exception):
             setattr(self, k, v)
 
 
+def maybe_int(x: Optional[int]) -> Optional[int]:
+    return None if x is None else int(x)
+
+
 def intx(x: Union[str, int]) -> int:
     if isinstance(x, str):
         x = int(x, 0)
@@ -311,11 +315,17 @@ class LOGIN(GPS303Pkt):
     PROTO = 0x01
     RESPOND = Respond.INL
     # Default response for ACK, can also respond with STOP_UPLOAD
+    IN_KWARGS = (("imei", str, "0000000000000000"), ("ver", int, 0))
 
     def in_decode(self, length: int, payload: bytes) -> None:
-        self.imei = payload[:8].hex()
+        self.imei = payload[:8].ljust(8, b"\0").hex()
         self.ver = payload[8]
 
+    def in_encode(self) -> bytes:
+        return bytes.fromhex(self.imei).ljust(8, b"\0")[:8] + pack(
+            "B", self.ver
+        )
+
 
 class SUPERVISION(GPS303Pkt):
     PROTO = 0x05
@@ -374,6 +384,13 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING):
 class STATUS(GPS303Pkt):
     PROTO = 0x13
     RESPOND = Respond.EXT
+    IN_KWARGS = (
+        ("batt", int, 100),
+        ("ver", int, 0),
+        ("timezone", int, 0),
+        ("intvl", int, 0),
+        ("signal", maybe_int, None),
+    )
     OUT_KWARGS = (("upload_interval", int, 25),)
 
     def in_decode(self, length: int, payload: bytes) -> None:
@@ -385,6 +402,13 @@ class STATUS(GPS303Pkt):
         else:
             self.signal = None
 
+    def in_encode(self) -> bytes:
+        return (
+            pack("BBBB", self.batt, self.ver, self.timezone, self.intvl) + b""
+            if self.signal is None
+            else pack("B", self.signal)
+        )
+
     def out_encode(self) -> bytes:  # Set interval in minutes
         return pack("B", self.upload_interval)
 
@@ -392,6 +416,9 @@ class STATUS(GPS303Pkt):
 class HIBERNATION(GPS303Pkt):  # Server can send to send devicee to sleep
     PROTO = 0x14
 
+    def in_encode(self) -> bytes:
+        return b""
+
 
 class RESET(GPS303Pkt):
     # Device sends when it got reset SMS

diff --git a/gps303/termconfig.py b/gps303/termconfig.py
index 831397201a4521c9c3fd160c0fdbb3bb8eb0b39a..723e8f62766ef542dbdb337ae015c66203ff1b9c 100644 (file)
--- a/gps303/termconfig.py
+++ b/gps303/termconfig.py
@@ -44,8 +44,10 @@ def runserver(conf: ConfigParser) -> None:
                 )
             if zmsg.imei is not None and conf.has_section(zmsg.imei):
                 termconfig = common.normconf(conf[zmsg.imei])
-            else:
+            elif conf.has_section("termconfig"):
                 termconfig = common.normconf(conf["termconfig"])
+            else:
+                termconfig = {}
             kwargs = {}
             if isinstance(msg, STATUS):
                 kwargs = {

diff --git a/test/test_storage.py b/test/test_storage.py
index 889446de24d5d941bc75f9165b2c6d716b4f28fe..7101c1aae422ffa0225f40e40d907933d56ff170 100644 (file)
--- a/test/test_storage.py
+++ b/test/test_storage.py
@@ -6,6 +6,7 @@ from sqlite3 import connect, Row
 from time import sleep
 import unittest
 from .common import send_and_drain, TestWithServers
+from gps303.gps303proto import *
 
 
 class Storage(TestWithServers):
@@ -26,16 +27,26 @@ class Storage(TestWithServers):
         super().tearDown()
 
     def test_storage(self) -> None:
-        buf = b"xx\r\x01\x03Y3\x90w\x19q\x85\x05\r\n"
-        send_and_drain(self.sock, buf)
+        for buf in (
+            LOGIN.In(imei="9999123456780000", ver=9).packed,
+            STATUS.In().packed,
+            HIBERNATION.In().packed,
+        ):
+            send_and_drain(self.sock, b"xx" + buf + b"\r\n")
         self.sock.close()
-        # TODO: make a proper sequence
         sleep(1)
-        print("checking database")
+        got = set()
         with connect(self.conf.get("storage", "dbfn")) as db:
             db.row_factory = Row
-            for row in db.execute("select * from events"):
-                print(dict(row))
+            for is_incoming, packet in db.execute(
+                "select is_incoming, packet from events"
+            ):
+                msg = parse_message(packet, is_incoming=is_incoming)
+                # print(msg)
+                got.add(type(msg))
+        self.assertEqual(
+            got, {LOGIN.Out, HIBERNATION.In, LOGIN.In, STATUS.Out, STATUS.In}
+        )
 
 
 if __name__ == "__main__":