feat: merge Survey into Scanning and harden live recon
- Delete the Survey stack (sampler, handlers, routes, view, recordings); Scanning gains AP compare (cap 6), whole-page selection filter, channel map (20 MHz lobes, the radio does not report width), and auto-follow that prefers the newest non-empty scan - Richer HTML scan reports: stat cards, band/encryption breakdowns, channel occupancy, color-coded signal cells, GPS line only on a fix - Harden live recon: bounded + retried scan-detail reads (503 on lock), serialize recon starts (409 + scan_remaining while running), GPS and archive discovery move to a 30s poll, GPS_GET skipped when gpsd is down - Read-only history for pineapd-rotated databases (error-*/diagnostic-*): archives list/detail/download endpoints, Previous Scans optgroup, delete disabled, traversal-guarded - Status flags: hopper-radio-offline and history-reset banners; empty scans greyed in the picker - Tests: recon suite grows to 99 cases; all 14 modules green Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
This commit is contained in:
co-authored by
factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
parent
251b1f6261
commit
d8a7074e1f
+255
-180
@@ -158,6 +158,10 @@ class FakeSock:
|
||||
|
||||
|
||||
class DaemonSockTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# h_recon_start now reads shared scan state; keep these isolated.
|
||||
server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0}
|
||||
|
||||
def test_socket_call_posts_json_to_sock(self):
|
||||
server.DAEMON_SOCK = '/tmp/api.sock'
|
||||
fake = FakeSock(b'HTTP/1.1 200 OK\r\nContent-Length: 17\r\n\r\n{"success":true}')
|
||||
@@ -299,6 +303,19 @@ class ReconScanStateTest(unittest.TestCase):
|
||||
status, data = self._status()
|
||||
self.assertFalse(data['scanning'])
|
||||
|
||||
def test_start_while_scanning_returns_409_without_restart(self):
|
||||
server.time.time = lambda: 1000.0
|
||||
self._start(scan_time=30)
|
||||
calls = []
|
||||
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p)) or (200, {'success': True})
|
||||
status, data = server.h_recon_start(
|
||||
type('C', (), {'args': (), 'body': {'scan_time': 30}})())
|
||||
self.assertEqual(status, 409)
|
||||
self.assertEqual(data['scan_remaining'], 30)
|
||||
self.assertEqual(calls, [])
|
||||
status, data = self._status()
|
||||
self.assertTrue(data['scanning'])
|
||||
|
||||
def test_watchdog_stops_expired_timed_scan(self):
|
||||
server.time.time = lambda: 1000.0
|
||||
self._start(scan_time=10)
|
||||
@@ -339,6 +356,25 @@ class ReconExtrasTest(unittest.TestCase):
|
||||
status, data = server.h_recon_status(type('C', (), {'args': ()})())
|
||||
self.assertFalse(data['active'])
|
||||
|
||||
def test_status_includes_hopper_and_history_flags(self):
|
||||
with mock.patch.object(server, '_hopper_online', return_value=False), \
|
||||
mock.patch.object(server, '_recon_history_reset', return_value=True):
|
||||
status, data = server.h_recon_status(type('C', (), {'args': ()})())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertFalse(data['hopper_online'])
|
||||
self.assertTrue(data['history_reset'])
|
||||
|
||||
def test_hopper_online_cached(self):
|
||||
server._hopper_cache.update({'updated': 0, 'online': None})
|
||||
with mock.patch.object(server, 'wifi_ifaces',
|
||||
return_value=['wlan0mon', 'wlan1mon', 'wlan2mon']):
|
||||
self.assertTrue(server._hopper_online())
|
||||
# Second call within the cache window must not re-run iwinfo.
|
||||
with mock.patch.object(server, 'wifi_ifaces',
|
||||
side_effect=AssertionError('cached, must not call iwinfo')):
|
||||
self.assertTrue(server._hopper_online())
|
||||
server._hopper_cache['updated'] = 0 # let other tests start fresh
|
||||
|
||||
def test_delete_cascades(self):
|
||||
server.recon_delete_scan(1)
|
||||
rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM wifi_device')
|
||||
@@ -637,27 +673,6 @@ class HandshakeRoutesTest(unittest.TestCase):
|
||||
self.assertEqual(os.listdir(self.dir), ['.hidden'])
|
||||
|
||||
|
||||
def make_survey_db():
|
||||
"""Single-scan recon DB so the newest scan carries the AP data."""
|
||||
fd, db = tempfile.mkstemp(suffix='.db')
|
||||
os.close(fd)
|
||||
conn = sqlite3.connect(db)
|
||||
conn.executescript(SCHEMA)
|
||||
conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u1', 1786466531, 'pager')")
|
||||
conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (1, 1, 'AE77C0EB3141', 1786466531, -71, 2412, 5)")
|
||||
conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (2, 1, 'C89E43648080', 1786466532, -76, 5745, 9)")
|
||||
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
|
||||
"VALUES (10, 2, 1, 8, 'C89E43648080', X'416E646572736F6E2D35', 0, 1786466532, -76, 5745, 149, 0x400400108)")
|
||||
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
|
||||
"VALUES (11, 2, 1, 8, '506F9A010000', X'', 1, 1786466532, -64, 5745, 149, 0)")
|
||||
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
|
||||
"VALUES (12, 1, 1, 4, NULL, X'5A6E6574', NULL, 1786466531, -40, 2412, NULL, NULL)")
|
||||
conn.execute("INSERT INTO handshake (hash, scan, stahash, aphash, time) VALUES (20, 1, 1, 2, 1786466600)")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return db
|
||||
|
||||
|
||||
class OuiVendorTest(unittest.TestCase):
|
||||
def test_oui_prefix_forms(self):
|
||||
self.assertEqual(server._oui_prefix('C8:9E:43:64:80:80'), 'C89E43')
|
||||
@@ -746,6 +761,30 @@ class ReconEnrichmentTest(unittest.TestCase):
|
||||
self.assertEqual(a['band'], '2.4')
|
||||
self.assertEqual(a['vendor'], 'Google')
|
||||
|
||||
def test_scan_detail_handler_is_bounded_and_retried(self):
|
||||
seen = {}
|
||||
orig = server.recon_scan_data
|
||||
def fake(scan_id, _timeout=20, _limit=None, db=None):
|
||||
seen['limit'] = _limit
|
||||
seen['db'] = db
|
||||
return orig(scan_id, _timeout=_timeout, _limit=_limit, db=db)
|
||||
with mock.patch.object(server, 'recon_scan_data', side_effect=fake), \
|
||||
mock.patch.object(server.time, 'sleep'):
|
||||
status, data = server.h_recon_scan_detail(type('C', (), {'args': ('1',)})())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(seen['limit'], 300)
|
||||
# The live handler relies on the default, which resolves to RECON_DB.
|
||||
self.assertIsNone(seen['db'])
|
||||
self.assertEqual(data['scan']['id'], 1)
|
||||
|
||||
def test_scan_detail_handler_503_on_lock(self):
|
||||
with mock.patch.object(server, 'recon_scan_data',
|
||||
side_effect=RuntimeError('sqlite read failed: locked')), \
|
||||
mock.patch.object(server.time, 'sleep'):
|
||||
status, data = server.h_recon_scan_detail(type('C', (), {'args': ('1',)})())
|
||||
self.assertEqual(status, 503)
|
||||
self.assertIn('temporarily unavailable', data['error'])
|
||||
|
||||
|
||||
class ReconReportTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
@@ -769,7 +808,8 @@ class ReconReportTest(unittest.TestCase):
|
||||
self.assertIn('C8:9E:43:64:80:80', text)
|
||||
|
||||
def test_html_download_contains_stats(self):
|
||||
status, payload = server.h_recon_scan_download_html(self._ctx(('1',)))
|
||||
with mock.patch.object(server, '_gps_status_data', return_value={'lock': False}):
|
||||
status, payload = server.h_recon_scan_download_html(self._ctx(('1',)))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload.ctype, 'text/html')
|
||||
self.assertEqual(payload.filename, 'scan-1.html')
|
||||
@@ -777,7 +817,32 @@ class ReconReportTest(unittest.TestCase):
|
||||
self.assertIn('Scan #1', text)
|
||||
self.assertIn('Anderson-5', text)
|
||||
self.assertIn('WPA3 WPA2', text)
|
||||
# stat cards
|
||||
self.assertIn('stat-card', text)
|
||||
self.assertIn('Unassociated', text)
|
||||
# band + encryption breakdowns
|
||||
self.assertIn('Band Breakdown', text)
|
||||
self.assertIn('Encryption Breakdown', text)
|
||||
self.assertIn('WPA2', text)
|
||||
# channel occupancy table
|
||||
self.assertIn('Channel Occupancy', text)
|
||||
self.assertIn('5 GHz', text)
|
||||
self.assertIn('>149<', text)
|
||||
# color-coded signal cells
|
||||
self.assertIn('sig-weak', text)
|
||||
self.assertIn('sig-good', text)
|
||||
self.assertIn('-76 dBm', text)
|
||||
# no GPS line without a fix
|
||||
self.assertNotIn('GPS:', text)
|
||||
|
||||
def test_html_report_includes_gps_when_locked(self):
|
||||
with mock.patch.object(server, '_gps_status_data',
|
||||
return_value={'lock': True, 'lat': 37.7,
|
||||
'lon': -122.4, 'satellites': 8}):
|
||||
status, payload = server.h_recon_scan_download_html(self._ctx(('1',)))
|
||||
text = payload.data.decode('utf-8')
|
||||
self.assertIn('GPS: 37.70000, -122.40000', text)
|
||||
self.assertIn('8 sats', text)
|
||||
|
||||
def test_download_404_for_missing_scan(self):
|
||||
status, payload = server.h_recon_scan_download_csv(self._ctx(('999',)))
|
||||
@@ -795,6 +860,135 @@ class ReconReportTest(unittest.TestCase):
|
||||
self.assertEqual(status, 503)
|
||||
|
||||
|
||||
class ReconArchivesTest(unittest.TestCase):
|
||||
"""Read-only history from pineapd-rotated databases (error-*-recon.db)."""
|
||||
|
||||
def setUp(self):
|
||||
self.dir = tempfile.mkdtemp()
|
||||
self.addCleanup(shutil.rmtree, self.dir)
|
||||
self.archive_name = 'error-2026-08-18-19:21:42Z-recon.db'
|
||||
tmp = make_db()
|
||||
self.addCleanup(os.unlink, tmp)
|
||||
server.RECON_DB = os.path.join(self.dir, 'recon.db')
|
||||
shutil.copy(tmp, server.RECON_DB)
|
||||
archive = os.path.join(self.dir, self.archive_name)
|
||||
shutil.copy(tmp, archive)
|
||||
conn = sqlite3.connect(archive)
|
||||
conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u3', 1786467000, 'pager')")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def tearDown(self):
|
||||
server._recon_archives_cache.update({'dir': None, 'updated': 0, 'data': None})
|
||||
|
||||
def _ctx(self, args=()):
|
||||
return type('C', (), {'args': args, 'body': {}})()
|
||||
|
||||
def test_archives_list_finds_rotated_db(self):
|
||||
data = server.recon_archives_data()
|
||||
self.assertEqual(len(data['archives']), 1)
|
||||
a = data['archives'][0]
|
||||
self.assertEqual(a['id'], self.archive_name)
|
||||
self.assertEqual(a['min_id'], 1)
|
||||
self.assertEqual(a['max_id'], 3)
|
||||
self.assertEqual(a['scans_count'], 3)
|
||||
self.assertEqual([s['id'] for s in a['scans']], [3, 2, 1])
|
||||
|
||||
def test_archives_list_is_cached_per_directory(self):
|
||||
orig = server.recon_scans_data
|
||||
calls = []
|
||||
def fake(limit=50, _timeout=20, db=None):
|
||||
calls.append(db)
|
||||
return orig(limit=limit, _timeout=_timeout, db=db)
|
||||
with mock.patch.object(server, 'recon_scans_data', side_effect=fake):
|
||||
server.recon_archives_data()
|
||||
server.recon_archives_data()
|
||||
# Second call within the cache window must not re-scan the archives.
|
||||
self.assertEqual(len(calls), 1)
|
||||
self.assertTrue(calls[0].endswith(self.archive_name))
|
||||
|
||||
def test_archive_path_rejects_traversal(self):
|
||||
self.assertIsNone(server._recon_archive_path('..%2F..%2Fetc%2Fpasswd'))
|
||||
self.assertIsNone(server._recon_archive_path('../etc/passwd'))
|
||||
self.assertIsNone(server._recon_archive_path('error-x-recon.db/../evil'))
|
||||
self.assertIsNone(server._recon_archive_path('random.db'))
|
||||
self.assertIsNone(server._recon_archive_path(''))
|
||||
self.assertIsNotNone(server._recon_archive_path(self.archive_name))
|
||||
|
||||
def test_archive_scan_detail_reads_archive_not_live(self):
|
||||
status, data = server.h_recon_archive_scan_detail(
|
||||
self._ctx((self.archive_name, '3')))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data['scan']['id'], 3)
|
||||
# The live db has no scan 3; the archive must be the source.
|
||||
self.assertIsNone(server.recon_scan_data(3))
|
||||
|
||||
def test_archive_scan_detail_is_bounded(self):
|
||||
seen = {}
|
||||
orig = server.recon_scan_data
|
||||
def fake(scan_id, _timeout=20, _limit=None, db=None):
|
||||
seen.update({'limit': _limit, 'db': db})
|
||||
return orig(scan_id, _timeout=_timeout, _limit=_limit, db=db)
|
||||
with mock.patch.object(server, 'recon_scan_data', side_effect=fake), \
|
||||
mock.patch.object(server.time, 'sleep'):
|
||||
status, data = server.h_recon_archive_scan_detail(
|
||||
self._ctx((self.archive_name, '1')))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(seen['limit'], 300)
|
||||
self.assertTrue(seen['db'].endswith(self.archive_name))
|
||||
|
||||
def test_archive_scans_list(self):
|
||||
status, data = server.h_recon_archive_scans(self._ctx((self.archive_name,)))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual([s['id'] for s in data['scans']], [3, 2, 1])
|
||||
|
||||
def test_archive_handlers_404_for_missing(self):
|
||||
status, data = server.h_recon_archive_scans(self._ctx(('nope.db',)))
|
||||
self.assertEqual(status, 404)
|
||||
status, data = server.h_recon_archive_scan_detail(self._ctx(('nope.db', '1')))
|
||||
self.assertEqual(status, 404)
|
||||
status, data = server.h_recon_archive_scan_detail(
|
||||
self._ctx((self.archive_name, '999')))
|
||||
self.assertEqual(status, 404)
|
||||
|
||||
def test_archive_downloads(self):
|
||||
status, payload = server.h_recon_archive_scan_download(
|
||||
self._ctx((self.archive_name, '1')))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload.filename, 'scan-1.json')
|
||||
self.assertIn(b'Anderson-5', payload.data)
|
||||
status, payload = server.h_recon_archive_scan_download_csv(
|
||||
self._ctx((self.archive_name, '1')))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload.ctype, 'text/csv')
|
||||
self.assertIn(b'C8:9E:43:64:80:80', payload.data)
|
||||
status, payload = server.h_recon_archive_scan_download_html(
|
||||
self._ctx((self.archive_name, '1')))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload.ctype, 'text/html')
|
||||
text = payload.data.decode('utf-8')
|
||||
self.assertIn('archived history', text)
|
||||
self.assertIn('Anderson-5', text)
|
||||
|
||||
def test_archive_html_omits_current_gps(self):
|
||||
# An archive is historical; attaching a current fix would mislead.
|
||||
with mock.patch.object(server, '_gps_status_data',
|
||||
return_value={'lock': True, 'lat': 37.7, 'lon': -122.4}):
|
||||
status, payload = server.h_recon_archive_scan_download_html(
|
||||
self._ctx((self.archive_name, '1')))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertNotIn('GPS:', payload.data.decode('utf-8'))
|
||||
|
||||
def test_archive_detail_503_on_lock(self):
|
||||
with mock.patch.object(server, 'recon_scan_data',
|
||||
side_effect=RuntimeError('sqlite read failed: locked')), \
|
||||
mock.patch.object(server.time, 'sleep'):
|
||||
status, data = server.h_recon_archive_scan_detail(
|
||||
self._ctx((self.archive_name, '1')))
|
||||
self.assertEqual(status, 503)
|
||||
self.assertIn('temporarily unavailable', data['error'])
|
||||
|
||||
|
||||
class GpsTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
server._gps_cache.update({'updated': 0, 'data': None})
|
||||
@@ -821,6 +1015,31 @@ class GpsTest(unittest.TestCase):
|
||||
self.assertTrue(data['present'])
|
||||
self.assertTrue(data['wigle'])
|
||||
|
||||
def test_gps_status_skips_hak5cmd_when_gpsd_down(self):
|
||||
# GPS_GET can block for seconds when gpsd is down; it must not run.
|
||||
with mock.patch.object(server, '_uci_gps_get', return_value=None), \
|
||||
mock.patch.object(server, '_gps_serial_candidates', return_value=[]), \
|
||||
mock.patch.object(server, '_wigle_config', return_value={'logwigle': False}), \
|
||||
mock.patch.object(server, '_gpsd_running', return_value=False), \
|
||||
mock.patch.object(server, '_gps_from_gpspipe', return_value=None), \
|
||||
mock.patch.object(server, '_gps_from_hak5cmd',
|
||||
side_effect=AssertionError('GPS_GET must not run when gpsd is down')) as hak5:
|
||||
data = server._gps_status_data_nocache()
|
||||
self.assertFalse(data['gpsd_running'])
|
||||
hak5.assert_not_called()
|
||||
|
||||
def test_gps_status_falls_back_to_hak5cmd_when_gpsd_up(self):
|
||||
with mock.patch.object(server, '_uci_gps_get', return_value=None), \
|
||||
mock.patch.object(server, '_gps_serial_candidates', return_value=[]), \
|
||||
mock.patch.object(server, '_wigle_config', return_value={'logwigle': False}), \
|
||||
mock.patch.object(server, '_gpsd_running', return_value=True), \
|
||||
mock.patch.object(server, '_gps_from_gpspipe', return_value=None), \
|
||||
mock.patch.object(server, '_gps_from_hak5cmd',
|
||||
return_value={'fix': 3, 'lat': 37.7, 'lon': -122.4, 'satellites': 8}):
|
||||
data = server._gps_status_data_nocache()
|
||||
self.assertEqual(data['lat'], 37.7)
|
||||
self.assertEqual(data['satellites'], 8)
|
||||
|
||||
def test_configure_binds_preferred_device_and_locks(self):
|
||||
candidates = [('1.2_1-1.2:1.0', '/dev/1.2'), ('1.3_2-1.3:1.0', '/dev/1.3')]
|
||||
with mock.patch.object(server, '_gps_serial_candidates', return_value=candidates), \
|
||||
@@ -937,152 +1156,6 @@ class WigleTest(unittest.TestCase):
|
||||
hak5.assert_called_with('WIGLE_STOP', timeout=10)
|
||||
|
||||
|
||||
class SurveyTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.db = make_survey_db()
|
||||
server.RECON_DB = self.db
|
||||
self.dir = tempfile.mkdtemp()
|
||||
self._orig = {
|
||||
'SURVEY_DIR': server.SURVEY_DIR,
|
||||
'SURVEY_MAX_SAMPLES': server.SURVEY_MAX_SAMPLES,
|
||||
'SURVEY_SAMPLE_INTERVAL': server.SURVEY_SAMPLE_INTERVAL,
|
||||
}
|
||||
server.SURVEY_DIR = self.dir
|
||||
server.SURVEY_MAX_SAMPLES = 3
|
||||
server.SURVEY_SAMPLE_INTERVAL = 0.0
|
||||
server._survey_state = {'active': False, 'id': None, 'name': None, 'path': None,
|
||||
'started': 0, 'samples': 0, 'last_sample': 0}
|
||||
server._gps_cache.update({'updated': 0, 'data': None})
|
||||
|
||||
def tearDown(self):
|
||||
server.SURVEY_DIR = self._orig['SURVEY_DIR']
|
||||
server.SURVEY_MAX_SAMPLES = self._orig['SURVEY_MAX_SAMPLES']
|
||||
server.SURVEY_SAMPLE_INTERVAL = self._orig['SURVEY_SAMPLE_INTERVAL']
|
||||
server._survey_state = {'active': False, 'id': None, 'name': None, 'path': None,
|
||||
'started': 0, 'samples': 0, 'last_sample': 0}
|
||||
server._gps_cache.update({'updated': 0, 'data': None})
|
||||
shutil.rmtree(self.dir)
|
||||
os.unlink(self.db)
|
||||
|
||||
def _ctx(self, args=(), body=None):
|
||||
return type('C', (), {'args': args, 'body': body or {}})()
|
||||
|
||||
def test_start_creates_meta_file(self):
|
||||
status, data = server.h_recon_survey_start(self._ctx(body={'name': 'Kitchen Walk'}))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(data['ok'])
|
||||
path = os.path.join(self.dir, data['id'] + '.jsonl')
|
||||
self.assertTrue(os.path.isfile(path))
|
||||
with open(path) as f:
|
||||
first = f.readline()
|
||||
self.assertIn('"meta"', first)
|
||||
self.assertIn('Kitchen Walk', first)
|
||||
|
||||
def test_start_rejects_duplicate(self):
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
status, data = server.h_recon_survey_start(self._ctx(body={'name': 'B'}))
|
||||
self.assertEqual(status, 409)
|
||||
|
||||
def test_sample_via_watchdog_and_cap(self):
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
server._recon_watchdog_tick()
|
||||
server._recon_watchdog_tick()
|
||||
self.assertEqual(server._survey_state['samples'], 2)
|
||||
status, data = server.h_recon_survey_stop(self._ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data['samples'], 2)
|
||||
self.assertFalse(server._survey_state['active'])
|
||||
|
||||
def test_sample_cap_stops_recording(self):
|
||||
server.SURVEY_MAX_SAMPLES = 2
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
for _ in range(4):
|
||||
server._recon_watchdog_tick()
|
||||
self.assertFalse(server._survey_state['active'])
|
||||
self.assertEqual(server._survey_state['samples'], 2)
|
||||
|
||||
def test_live_reports_scan_unassociated_and_recording(self):
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
server._recon_watchdog_tick()
|
||||
status, data = server.h_recon_survey_live(self._ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data['scan']['id'], 1)
|
||||
self.assertEqual(len(data['aps']), 2)
|
||||
self.assertEqual(data['unassociated'], 1)
|
||||
self.assertTrue(data['recording']['active'])
|
||||
self.assertEqual(data['recording']['samples'], 1)
|
||||
self.assertIn('wigle', data['gps'])
|
||||
|
||||
def test_live_recording_none_when_stopped(self):
|
||||
status, data = server.h_recon_survey_live(self._ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertIsNone(data['recording'])
|
||||
|
||||
def test_surveys_list_counts_samples(self):
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
server._recon_watchdog_tick()
|
||||
server.h_recon_survey_stop(self._ctx())
|
||||
status, data = server.h_recon_surveys(self._ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(len(data['surveys']), 1)
|
||||
survey = data['surveys'][0]
|
||||
self.assertEqual(survey['name'], 'A')
|
||||
self.assertEqual(survey['samples'], 1)
|
||||
self.assertGreater(survey['size'], 0)
|
||||
|
||||
def test_detail_aggregates_signal(self):
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
server._recon_watchdog_tick()
|
||||
server._recon_watchdog_tick()
|
||||
sid = server._survey_state['id']
|
||||
server.h_recon_survey_stop(self._ctx())
|
||||
status, data = server.h_recon_survey_detail(self._ctx((sid,)))
|
||||
self.assertEqual(status, 200)
|
||||
aps = {a['bssid']: a for a in data['aps']}
|
||||
a = aps['C8:9E:43:64:80:80']
|
||||
self.assertEqual(a['min'], -76)
|
||||
self.assertEqual(a['max'], -76)
|
||||
self.assertEqual(a['avg'], -76)
|
||||
self.assertEqual(a['samples'], 2)
|
||||
self.assertEqual(a['band'], '5')
|
||||
self.assertEqual(a['channel'], 149)
|
||||
self.assertEqual(a['first_seen'], a['last_seen'])
|
||||
self.assertEqual(data['gps_fixes'], 0)
|
||||
|
||||
def test_downloads_all_formats(self):
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
server._recon_watchdog_tick()
|
||||
sid = server._survey_state['id']
|
||||
server.h_recon_survey_stop(self._ctx())
|
||||
for fmt, ctype in [('json', 'application/json'), ('csv', 'text/csv'), ('html', 'text/html')]:
|
||||
status, payload = server.h_recon_survey_download(self._ctx((sid, fmt)))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload.ctype, ctype)
|
||||
self.assertEqual(payload.filename, 'survey-%s.%s' % (sid, fmt))
|
||||
status, payload = server.h_recon_survey_download(self._ctx((sid, 'csv')))
|
||||
self.assertIn('C8:9E:43:64:80:80', payload.data.decode('utf-8'))
|
||||
|
||||
def test_delete_removes_file_then_404(self):
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
sid = server._survey_state['id']
|
||||
server.h_recon_survey_stop(self._ctx())
|
||||
status, data = server.h_recon_survey_delete(self._ctx((sid,)))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertFalse(os.listdir(self.dir))
|
||||
status, data = server.h_recon_survey_delete(self._ctx((sid,)))
|
||||
self.assertEqual(status, 404)
|
||||
|
||||
def test_delete_blocks_active_survey(self):
|
||||
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
|
||||
sid = server._survey_state['id']
|
||||
status, data = server.h_recon_survey_delete(self._ctx((sid,)))
|
||||
self.assertEqual(status, 409)
|
||||
|
||||
def test_detail_404_missing(self):
|
||||
status, data = server.h_recon_survey_detail(self._ctx(('nope',)))
|
||||
self.assertEqual(status, 404)
|
||||
|
||||
|
||||
class ReconRoutesTest(unittest.TestCase):
|
||||
def test_new_routes_registered(self):
|
||||
expected = [
|
||||
@@ -1093,24 +1166,26 @@ class ReconRoutesTest(unittest.TestCase):
|
||||
('POST', '/api/recon/wigle', 'h_recon_wigle'),
|
||||
('GET', '/api/recon/wigle/files', 'h_recon_wigle_files'),
|
||||
('GET', '/api/recon/wigle/files/x.csv', 'h_recon_wigle_file'),
|
||||
('GET', '/api/recon/survey/live', 'h_recon_survey_live'),
|
||||
('POST', '/api/recon/survey/start', 'h_recon_survey_start'),
|
||||
('POST', '/api/recon/survey/stop', 'h_recon_survey_stop'),
|
||||
('GET', '/api/recon/surveys', 'h_recon_surveys'),
|
||||
('GET', '/api/recon/surveys/20260818-120000-A', 'h_recon_survey_detail'),
|
||||
('GET', '/api/recon/surveys/20260818-120000-A/download/csv', 'h_recon_survey_download'),
|
||||
('GET', '/api/recon/surveys/20260818-120000-A/download/json', 'h_recon_survey_download'),
|
||||
('GET', '/api/recon/surveys/20260818-120000-A/download/html', 'h_recon_survey_download'),
|
||||
('DELETE', '/api/recon/surveys/20260818-120000-A', 'h_recon_survey_delete'),
|
||||
('GET', '/api/recon/archives', 'h_recon_archives'),
|
||||
('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans', 'h_recon_archive_scans'),
|
||||
('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans/1', 'h_recon_archive_scan_detail'),
|
||||
('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans/1/download/json', 'h_recon_archive_scan_download'),
|
||||
('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans/1/download/csv', 'h_recon_archive_scan_download_csv'),
|
||||
('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans/1/download/html', 'h_recon_archive_scan_download_html'),
|
||||
]
|
||||
for method, path, handler in expected:
|
||||
h, args = server.ROUTER.dispatch(method, path)
|
||||
self.assertIsNotNone(h, '%s %s' % (method, path))
|
||||
self.assertEqual(h.__name__, handler, '%s %s' % (method, path))
|
||||
|
||||
def test_survey_download_route_captures_format(self):
|
||||
h, args = server.ROUTER.dispatch('GET', '/api/recon/surveys/abc/download/csv')
|
||||
self.assertEqual(args, ('abc', 'csv'))
|
||||
def test_survey_routes_are_gone(self):
|
||||
for method, path in [('GET', '/api/recon/survey/live'),
|
||||
('POST', '/api/recon/survey/start'),
|
||||
('GET', '/api/recon/surveys'),
|
||||
('GET', '/api/recon/surveys/abc'),
|
||||
('DELETE', '/api/recon/surveys/abc')]:
|
||||
h, args = server.ROUTER.dispatch(method, path)
|
||||
self.assertIsNone(h, '%s %s should not be registered' % (method, path))
|
||||
|
||||
def test_original_recon_routes_unchanged(self):
|
||||
for path in ['/api/recon/start', '/api/recon/status', '/api/recon/scans',
|
||||
|
||||
Reference in New Issue
Block a user