Track device connection and capture gaps; surface sniff error flags

tests/test_capture.py

@@ -57,6 +57,36 @@ class TestCaptureManager(unittest.TestCase):
         sn._session = _DisconnectingSession()
         self.assertEqual(list(sn.iter_messages()), [])
 
+    def test_capture_gap_message_inserted(self):
+        # A 'gap' marker yielded by the backend is stored as a message.
+        mgr, db = self._manager(_FakeBackend([
+            ('gap', b'', 0),
+            ('tpdu', b'\x80\xf2\x00\x00\x00', 0),
+        ]))
+        sid = mgr.start_session()
+        self.assertEqual(mgr.stop_session(), sid)
+        msgs = db.get_messages(sid)
+        self.assertEqual([m['type'] for m in msgs], ['gap', 'tpdu'])
+
+    def test_direct_sniffer_connected(self):
+        from simtrace2_pysniff.server.capture import DirectSniffer
+
+        class _FakeSession:
+            connected = True
+
+        sn = DirectSniffer()
+        sn._session = _FakeSession()
+        self.assertTrue(sn.connected)
+        _FakeSession.connected = False
+        self.assertFalse(sn.connected)
+
+    def test_device_connected_gsmtap_none(self):
+        from simtrace2_pysniff.server.capture import GsmtapListener
+        listener = GsmtapListener(bind_port=0)
+        self.addCleanup(listener.stop)
+        mgr, _ = self._manager(listener)
+        self.assertIsNone(mgr.device_connected)
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_decode.py

@@ -149,6 +149,19 @@ class TestChangeAndFidiDecoding(unittest.TestCase):
         self.assertEqual(r['fi_val'], 512)
         self.assertEqual(r['di_val'], 64)
 
+    def test_tpdu_error_flags_surfaced(self):
+        r = decode_sniff_msg(bytes.fromhex('00a40000023f009000'), 'tpdu',
+                             flags=(1 << 7) | (1 << 6))
+        self.assertEqual(r['errors'], ['malformed', 'checksum error'])
+
+    def test_tpdu_no_error_flags(self):
+        r = decode_sniff_msg(bytes.fromhex('00a40000023f009000'), 'tpdu', flags=0)
+        self.assertNotIn('errors', r)
+
+    def test_atr_error_flags_surfaced(self):
+        r = decode_sniff_msg(bytes.fromhex('3b00'), 'atr', flags=(1 << 5))
+        self.assertEqual(r['errors'], ['incomplete'])
+
 
 class TestTerminalResponse(unittest.TestCase):
     def test_tr_setup_event_list(self):
@@ -1192,6 +1205,49 @@ class TestSelectionTracking(unittest.TestCase):
             msgs = db.get_messages(sid)
             self.assertNotIn('file', msgs[2]['decoded'])
 
+    def test_timeout_wt_keeps_selection(self):
+        # A waiting-time timeout interrupts one TPDU but the card stays
+        # selected, so the selection must not be reset.
+        import tempfile
+        from simtrace2_pysniff.server.database import Database
+        with tempfile.NamedTemporaryFile(suffix='.db') as tmp:
+            db = Database(tmp.name)
+            sid = db.create_session('capture')
+            db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('a0a40804047fff6f079000'))
+            db.insert_message(sid, 0.1, 'change', b'', 1 << 4)  # TIMEOUT_WT
+            db.insert_message(sid, 0.2, 'tpdu',
+                              bytes.fromhex('00b0000009') + bytes.fromhex('082905102143658709') + bytes.fromhex('9000'))
+            msgs = db.get_messages(sid)
+            self.assertEqual(msgs[2]['decoded']['file']['imsi'], '250011234567890')
+
+    def test_reset_assert_resets_selection(self):
+        import tempfile
+        from simtrace2_pysniff.server.database import Database
+        with tempfile.NamedTemporaryFile(suffix='.db') as tmp:
+            db = Database(tmp.name)
+            sid = db.create_session('capture')
+            db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('a0a40804047fff6f079000'))
+            db.insert_message(sid, 0.1, 'change', b'', 1 << 2)  # RESET_ASSERT
+            db.insert_message(sid, 0.2, 'tpdu',
+                              bytes.fromhex('00b0000009') + bytes.fromhex('082905102143658709') + bytes.fromhex('9000'))
+            msgs = db.get_messages(sid)
+            self.assertNotIn('file', msgs[2]['decoded'])
+
+    def test_gap_resets_selection(self):
+        # A capture gap (device disconnected/reconnected) means messages may
+        # have been missed, so the selection must be reset.
+        import tempfile
+        from simtrace2_pysniff.server.database import Database
+        with tempfile.NamedTemporaryFile(suffix='.db') as tmp:
+            db = Database(tmp.name)
+            sid = db.create_session('capture')
+            db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('a0a40804047fff6f079000'))
+            db.insert_message(sid, 0.1, 'gap', b'', 0)
+            db.insert_message(sid, 0.2, 'tpdu',
+                              bytes.fromhex('00b0000009') + bytes.fromhex('082905102143658709') + bytes.fromhex('9000'))
+            msgs = db.get_messages(sid)
+            self.assertNotIn('file', msgs[2]['decoded'])
+
     def test_failed_select_keeps_selection(self):
         # A SELECT that fails (6A82) must not change the current file.
         import tempfile

tests/test_pcap_import.py

@@ -208,6 +208,7 @@ class TestVersion(unittest.TestCase):
         class FakeCapture:
             active = False
             session_id = None
+            device_connected = None
 
         class Fake(RequestHandler):
             def __init__(self, db):