bzuccaro/Mark-VIII
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: repair recon scanning and add macOS deployment

Browse Source 
Start scans via the Pager's native /api/pineap/recon/new so history
appends instead of rotating. Enforce finite durations (1-86400s,
default 30s), remove the unsupported Continuous mode, and refuse the
unsafe manual stop that left recon.db locked.

Rework scan list and detail reads into single-pass aggregate SQL,
shorten lock retries, and serve cached results during short exclusive
lock windows. Fall back to immutable read-only access when the
firmware leaves a stale lock after native completion.

Frontend auto-follows new scans, queues a single in-flight detail
refresh, keeps previous tables visible while a scan starts, and shows
completion toasts and daemon error details.

Add scripts/deploy.sh for macOS/Linux (zip packaging, scp/ssh install,
atomic payload replacement, service restart, portal refresh) with
README instructions, plus regression coverage for native start,
safe stop semantics, aggregate queries, and stale-lock fallback.

Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
This commit is contained in:
bzuccaro
2026-08-17 17:46:54 -05:00
co-authored by factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
parent 3e1805dab8
commit 2bf39ecb9d
5 changed files with 431 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_recon.py
+58 -20
View File
@@ -182,9 +182,8 @@ class DaemonSockTest(unittest.TestCase):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True})
server.h_recon_start(type('C', (), {'args': ()})())
server.h_recon_stop(type('C', (), {'args': ()})())
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {}))
self.assertEqual(calls[1], ('POST', '/api/pineap/log/recon/stop', {}))
self.assertEqual(calls, [
('POST', '/api/pineap/recon/new', {'scan_time': 30})])
def test_start_forwards_scan_time(self):
calls = []
@@ -192,19 +191,36 @@ class DaemonSockTest(unittest.TestCase):
ctx = type('C', (), {'args': (), 'body': {'scan_time': 60}})()
status, data = server.h_recon_start(ctx)
self.assertEqual(status, 200)
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {'scan_time': 60}))
self.assertEqual(calls[0], ('POST', '/api/pineap/recon/new', {'scan_time': 60}))
def test_start_defaults_empty_body(self):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True})
server.h_recon_start(type('C', (), {'args': ()})())
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {}))
self.assertEqual(calls[0], ('POST', '/api/pineap/recon/new', {'scan_time': 30}))
def test_start_rejects_invalid_scan_time(self):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body))
ctx = type('C', (), {'args': (), 'body': {'scan_time': 'forever'}})()
status, data = server.h_recon_start(ctx)
self.assertEqual(status, 400)
self.assertIn('scan_time', data['error'])
self.assertEqual(calls, [])
def test_start_reports_native_failure(self):
server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0}
server.daemon_sock_call = lambda m, p, body=None: (500, {'error': 'no radio'})
status, data = server.h_recon_start(
type('C', (), {'args': (), 'body': {'scan_time': 30}})())
self.assertEqual(status, 502)
self.assertEqual(data['error'], 'native recon scan failed')
self.assertEqual(data['detail'], {'error': 'no radio'})
self.assertFalse(server._recon_scan_state['active'])
class ReconScanStateTest(unittest.TestCase):
"""The daemon ignores scan_time and scans continuously until 'stop'. The webui
must track the requested duration itself so timed scans actually end and the
toggle can reflect real scan state."""
"""The webui mirrors the duration of the Pager's native timed scan."""
def setUp(self):
self.db = make_db()
@@ -248,26 +264,33 @@ class ReconScanStateTest(unittest.TestCase):
self.assertFalse(data['scanning'])
self.assertEqual(data['scan_remaining'], 0)
def test_continuous_scan_has_no_remaining(self):
def test_zero_duration_is_rejected(self):
server.time.time = lambda: 1000.0
self._start(scan_time=0)
status, data = self._status()
self.assertTrue(data['scanning'])
self.assertIsNone(data['scan_remaining'])
status, data = self._start(scan_time=0)
self.assertEqual(status, 400)
self.assertFalse(server._recon_scan_state['active'])
def test_default_start_is_continuous(self):
def test_default_start_uses_thirty_seconds(self):
server.time.time = lambda: 1000.0
self._start()
status, data = self._status()
self.assertTrue(data['scanning'])
self.assertIsNone(data['scan_remaining'])
self.assertEqual(data['scan_remaining'], 30)
def test_stop_clears_scanning(self):
def test_stop_rejects_active_native_scan(self):
server.time.time = lambda: 1000.0
self._start(scan_time=30)
self._stop()
status, data = self._stop()
self.assertEqual(status, 409)
self.assertIn('finish automatically', data['error'])
self.assertEqual(data['scan_remaining'], 30)
status, data = self._status()
self.assertFalse(data['scanning'])
self.assertTrue(data['scanning'])
def test_stop_is_idempotent_when_inactive(self):
status, data = self._stop()
self.assertEqual(status, 200)
self.assertEqual(data, {'ok': True})
def test_start_failure_does_not_mark_scanning(self):
server.time.time = lambda: 1000.0
@@ -277,13 +300,13 @@ class ReconScanStateTest(unittest.TestCase):
self.assertFalse(data['scanning'])
def test_watchdog_stops_expired_timed_scan(self):
calls = []
server.time.time = lambda: 1000.0
self._start(scan_time=10)
server.time.time = lambda: 1012.0
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p)) or (200, {'success': True})
server._recon_watchdog_tick()
self.assertEqual(calls, [('POST', '/api/pineap/log/recon/stop')])
self.assertEqual(calls, [])
self.assertFalse(server._recon_scan_state['active'])
def test_watchdog_leaves_active_scan_alone(self):
@@ -454,6 +477,21 @@ class CliFallbackTest(unittest.TestCase):
server.RECON_DB = '/nonexistent.db'
self.assertEqual(server.decode_ssid('casaalicia\\x00.\\xde_'), 'casaalicia\x00.\ufffd_')
def test_completed_recon_lock_uses_immutable_read(self):
calls = []
server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0}
def locked_then_read(args, timeout=20):
calls.append(args)
if len(calls) == 1:
return 5, '', 'Error: database is locked'
return 0, '[{"id": 2}]', ''
server.device_run = locked_then_read
rows = server._db_rows(self.db, 'SELECT MAX(id) AS id FROM scan')
self.assertEqual(rows, [{'id': 2}])
self.assertEqual(calls[1][-2], 'file:%s?immutable=1' % self.db)
def make_hs_db():
db = make_db()