Show status in trace empty-state; discard empty captures; polish capture-mode
Frontend: - trace empty-state now shows connection status + capture mode (boxed, colored) in the center, updated live via checkConnection - capture-mode uses amber warning style (gsmtap/disabled), green for direct - stopCapture clears currentSessionId Server: - gsmtap read_packet catches OSError on socket close (no more EBADF traceback on stop) - stop_session joins the loop and discards a session with zero messages; /api/capture/stop reports discarded - startup prints server version Tests: 122 (gsmtap close, empty/non-empty capture lifecycle).
This commit is contained in:
1 parent
0bb3bc82c7
commit
97a7e6034a
7 files changed
+147
-29
No files matched your search
@@ -0,0 +1,49 @@
|
||||
"""Tests for CaptureManager session lifecycle (empty-session discard)."""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from simtrace2_pysniff.server.capture import CaptureManager
|
||||
from simtrace2_pysniff.server.database import Database
|
||||
|
||||
|
||||
class _FakeBackend:
|
||||
def __init__(self, messages=()):
|
||||
self._messages = list(messages)
|
||||
|
||||
def start(self):
|
||||
pass
|
||||
|
||||
def stop(self):
|
||||
pass
|
||||
|
||||
def iter_messages(self):
|
||||
yield from self._messages
|
||||
|
||||
|
||||
class TestCaptureManager(unittest.TestCase):
|
||||
def _manager(self, backend):
|
||||
tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
||||
tmp.close()
|
||||
db = Database(tmp.name)
|
||||
self.addCleanup(lambda: (db._conn.close(), os.remove(tmp.name)))
|
||||
return CaptureManager(backend, db), db
|
||||
|
||||
def test_empty_capture_discarded(self):
|
||||
mgr, db = self._manager(_FakeBackend())
|
||||
sid = mgr.start_session()
|
||||
self.assertIsNotNone(sid)
|
||||
self.assertIsNone(mgr.stop_session())
|
||||
self.assertIsNone(db.get_session(sid))
|
||||
|
||||
def test_non_empty_capture_kept(self):
|
||||
mgr, db = self._manager(_FakeBackend([('tpdu', b'\x80\xf2\x00\x00\x00', 0)]))
|
||||
sid = mgr.start_session()
|
||||
self.assertEqual(mgr.stop_session(), sid)
|
||||
self.assertIsNotNone(db.get_session(sid))
|
||||
self.assertEqual(db.count_messages(sid), 1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,16 @@
|
||||
"""Tests for GSMTAP receiver shutdown behavior."""
|
||||
|
||||
import unittest
|
||||
|
||||
from simtrace2_pysniff.gsmtap import GsmtapReceiver
|
||||
|
||||
|
||||
class TestGsmtapReceiver(unittest.TestCase):
|
||||
def test_read_after_close_returns_none(self):
|
||||
r = GsmtapReceiver(bind_port=0)
|
||||
r.close()
|
||||
self.assertEqual(r.read_packet(), (None, None))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user