452 lines
24 KiB
Python
452 lines
24 KiB
Python
import argparse
|
|
import base64
|
|
import contextlib
|
|
import hashlib
|
|
import importlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from unittest.mock import patch
|
|
|
|
from whatsapp_suite import exports, history, live, qr_memory, ram_search
|
|
from whatsapp_suite.common import BASE, asset_bytes, resource_bytes, cli_command
|
|
from whatsapp_suite.analyzer import endpoint
|
|
|
|
|
|
class SavedExportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.folder = Path(self.temp.name)
|
|
self.rows = [
|
|
{'id': '1', 'chat_id': 'a', 'body': 'Cafe\u0301 STRASSE', 'type': 'chat', 'nested': {'secret': 'private'}},
|
|
{'id': '2', 'chatId': 'b', 'caption': 'CAFÉ Straße', 'type': 'image'},
|
|
{'id': '3', 'chat_id': 'a', 'indexed_text': 'other', 'type': 'chat'},
|
|
]
|
|
self.path = self.folder / 'messages.json'
|
|
self.path.write_text(json.dumps(self.rows), encoding='utf-8')
|
|
|
|
def invoke(self, *args):
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out):
|
|
self.assertEqual(exports.main(list(args)), 0)
|
|
return json.loads(out.getvalue())
|
|
|
|
def test_json_report_wrappers(self):
|
|
for key in ('messages', 'models', 'results'):
|
|
self.path.write_text(json.dumps({key: self.rows}), encoding='utf-8')
|
|
self.assertEqual(list(exports.messages(self.path)), self.rows)
|
|
|
|
def test_jsonl_bom_and_blank_lines(self):
|
|
path = self.folder / 'messages.jsonl'
|
|
path.write_text('\n' + '\n\n'.join(json.dumps(r) for r in self.rows), encoding='utf-8-sig')
|
|
self.assertEqual(list(exports.messages(path)), self.rows)
|
|
|
|
def test_malformed_jsonl_reports_line(self):
|
|
path = self.folder / 'broken.jsonl'
|
|
path.write_text('{}\ninvalid', encoding='utf-8')
|
|
with self.assertRaisesRegex(ValueError, 'line 2'):
|
|
list(exports.messages(path))
|
|
|
|
def test_nonobject_records_rejected(self):
|
|
for data in ([None], {'messages': [42]}, 'wrong', {}):
|
|
self.path.write_text(json.dumps(data), encoding='utf-8')
|
|
with self.assertRaises(ValueError):
|
|
list(exports.messages(self.path))
|
|
|
|
def test_unicode_search_and_total_beyond_limit(self):
|
|
result = self.invoke('search', str(self.path), 'café straße', '--limit', '1')
|
|
self.assertEqual((result['matches'], result['returned']), (2, 1))
|
|
|
|
def test_chat_filter_and_nested_redaction(self):
|
|
result = self.invoke('search', str(self.path), 'café', '--chat-id', 'a', '--redact-text')
|
|
self.assertEqual(result['matches'], 1)
|
|
row = result['results'][0]
|
|
self.assertEqual(row['id'], '1')
|
|
self.assertNotIn('body', row)
|
|
self.assertNotIn('nested', row)
|
|
self.assertGreater(row['text_length'], 0)
|
|
|
|
def test_stats_understands_both_chat_id_fields(self):
|
|
result = self.invoke('stats', str(self.path))
|
|
self.assertEqual((result['messages'], result['messages_with_text'], result['chats']), (3, 3, 2))
|
|
self.assertEqual(result['types'], {'chat': 2, 'image': 1})
|
|
|
|
def test_output_cannot_overwrite_input(self):
|
|
before = self.path.read_bytes()
|
|
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
|
exports.main(['stats', str(self.path), '--output', str(self.path)])
|
|
self.assertEqual(self.path.read_bytes(), before)
|
|
|
|
def test_empty_query_and_invalid_limit_rejected(self):
|
|
for tail in ([' '], ['café', '--limit', '0']):
|
|
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
|
exports.main(['search', str(self.path)] + tail)
|
|
|
|
def test_inventory_empty_and_existing_profile(self):
|
|
self.assertFalse(exports.inventory(self.folder / 'absent')['exists'])
|
|
store = self.folder / 'Local Storage/leveldb'
|
|
store.mkdir(parents=True)
|
|
(store / 'sample.ldb').write_bytes(b'123')
|
|
report = exports.inventory(self.folder)
|
|
self.assertEqual(report['storage_inventory'][0]['files'], [{'name': 'sample.ldb', 'bytes': 3}])
|
|
|
|
|
|
class ProbeContractTests(unittest.TestCase):
|
|
@unittest.skipUnless(shutil.which('node'), 'Node.js is needed for JavaScript fixtures')
|
|
def test_self_message_guards_and_contact_export(self):
|
|
from whatsapp_suite.self_message import PROBE as send
|
|
from whatsapp_suite.contacts_export import PROBE as contacts
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
probe=Path(folder)/'probes.json'
|
|
probe.write_text(json.dumps({'send':send.replace('__BODY__',json.dumps('fixture')),'contacts':contacts}),encoding='utf-8')
|
|
result=subprocess.run(['node','-',str(probe)],input=resource_bytes('tests/self_and_contacts.cjs').decode('utf-8'),capture_output=True,text=True,encoding='utf-8',timeout=20)
|
|
self.assertEqual(result.returncode,0,result.stdout+result.stderr)
|
|
|
|
def test_self_send_refuses_existing_receipt_without_retry(self):
|
|
from whatsapp_suite.self_message import send_self
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path=Path(folder)/'receipt.json'
|
|
path.write_text('original',encoding='utf-8')
|
|
with patch('whatsapp_suite.self_message.run_math') as bridge:
|
|
with self.assertRaises(ValueError):send_self(path)
|
|
bridge.assert_not_called()
|
|
self.assertEqual(path.read_text(),'original')
|
|
|
|
@unittest.skipUnless(shutil.which('node'), 'Node.js is needed for JavaScript fixtures')
|
|
def test_oldest_history_read_without_cdp(self):
|
|
from whatsapp_suite.history_read import PROBE
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
probe = Path(folder) / 'probe.json'
|
|
probe.write_text(json.dumps(PROBE), encoding='utf-8')
|
|
result = subprocess.run(['node', '-', str(probe)], input=resource_bytes('tests/history_read.cjs').decode('utf-8'), capture_output=True, text=True, encoding='utf-8', timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def test_ram_search_requires_verified_membership(self):
|
|
rows = [{'body': 'café', 'chat_id': 'a', 'in_message_collection': valid} for valid in (True, False)]
|
|
result = ram_search.search_models(rows, 'CAFE\u0301', chat_id='a', redact=True)
|
|
self.assertEqual(len(result), 1)
|
|
self.assertNotIn('body', result[0])
|
|
self.assertEqual(result[0]['body_length'], 4)
|
|
self.assertEqual(ram_search.search_models(rows, 'café', chat_id='b'), [])
|
|
with self.assertRaises(ValueError):
|
|
ram_search.search_models(rows, ' ')
|
|
|
|
def test_loopback_endpoint_validation(self):
|
|
self.assertEqual(endpoint('http://127.0.0.1:9223/'), 'http://127.0.0.1:9223')
|
|
self.assertEqual(endpoint('http://[::1]:9223'), 'http://[::1]:9223')
|
|
for url in ('https://example.com', 'file:///tmp/x', 'http://localhost:9223', 'http://user:[email protected]', 'http://127.0.0.1?x=1'):
|
|
with self.assertRaises(argparse.ArgumentTypeError):
|
|
endpoint(url)
|
|
|
|
def test_native_assets_match_verified_hashes(self):
|
|
from whatsapp_suite.native_search import EMBEDDED_PROBE_SHA256
|
|
for filename, expected in (('math_probe.dll', live.EMBEDDED_SHA256), ('search_probe.dll', EMBEDDED_PROBE_SHA256)):
|
|
payload = asset_bytes(filename)
|
|
self.assertEqual(payload[:2], b'MZ')
|
|
self.assertEqual(hashlib.sha256(payload).hexdigest(), expected)
|
|
|
|
def test_math_response_binding_and_float_tolerance(self):
|
|
data = {'probe_id': 'test', 'cases': [{'expression': expression, 'value': expected} for expression, expected in live.CASES]}
|
|
self.assertTrue(all(row['passed'] for row in live.validate_result(json.dumps(data), 'test')))
|
|
with self.assertRaises(ValueError):
|
|
live.validate_result(json.dumps(data), 'other')
|
|
data['cases'][0]['value'] = -1
|
|
self.assertFalse(live.validate_result(json.dumps(data), 'test')[0]['passed'])
|
|
|
|
def test_capability_response_must_match_fixed_checks(self):
|
|
data = {'probe_id': 'test', 'state': 'done', 'checks': [{'id': spec[0]} for spec in live.CAPABILITIES]}
|
|
self.assertEqual(live.validate_capabilities(data, 'test'), data['checks'])
|
|
data['checks'].reverse()
|
|
with self.assertRaises(ValueError):
|
|
live.validate_capabilities(data, 'test')
|
|
|
|
def test_pairing_payload_shape(self):
|
|
key = base64.b64encode(bytes(range(32))).decode()
|
|
value = ','.join(['reference123', key, key, key, '8'])
|
|
self.assertTrue(qr_memory.valid_payload(value))
|
|
self.assertTrue(qr_memory.valid_payload('https://wa.me/settings/linked_devices#' + value))
|
|
self.assertFalse(qr_memory.valid_payload(value.replace(',8', ',7')))
|
|
self.assertFalse(qr_memory.valid_payload('reference,invalid,invalid,invalid,8'))
|
|
|
|
def test_history_refuses_existing_export(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / 'history.jsonl'
|
|
path.write_text('original', encoding='utf-8')
|
|
with self.assertRaisesRegex(RuntimeError, 'already exists'):
|
|
history.export(None, path)
|
|
self.assertEqual(path.read_text(), 'original')
|
|
|
|
def test_desktop_search_escapes_query_and_enforces_bounds(self):
|
|
from whatsapp_suite import native_search
|
|
query = '"; window.unwanted = true; //'
|
|
with patch.object(native_search, 'native_webview_test', return_value=0) as backend:
|
|
self.assertEqual(native_search.run_live_probe(None, Path('unused.json'), query, 10), 0)
|
|
config = backend.call_args.args[2]
|
|
self.assertIn('query=' + json.dumps(query), config['start'])
|
|
self.assertEqual(config['mode'], 'search')
|
|
for query, limit in (('', 10), ('valid', 0), ('valid', 51), ('x' * 201, 10)):
|
|
with self.assertRaises(ValueError):
|
|
native_search.run_live_probe(None, Path('unused.json'), query, limit)
|
|
|
|
@unittest.skipUnless(shutil.which('node'), 'Node.js is needed for the JavaScript history fixtures')
|
|
def test_history_javascript_pagination_decryption_redaction(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
probe = Path(folder) / 'probe.json'
|
|
probe.write_text(json.dumps(history.PROBE), encoding='utf-8')
|
|
result = subprocess.run(['node', '-', str(probe)], input=resource_bytes('tests/history_export.cjs').decode('utf-8'), capture_output=True, text=True, encoding='utf-8', timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
|
|
class CommandTests(unittest.TestCase):
|
|
def test_all_commands_have_help(self):
|
|
from whatsapp_suite.cli import COMMANDS
|
|
for command in ['test'] + list(COMMANDS):
|
|
with self.subTest(command=command):
|
|
result = subprocess.run(cli_command(command, '--help'), cwd=BASE, capture_output=True, text=True, timeout=15)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn('usage:', result.stdout)
|
|
|
|
def test_saved_search_dispatch(self):
|
|
from whatsapp_suite.cli import main
|
|
with patch('whatsapp_suite.exports.main', return_value=0) as target:
|
|
self.assertEqual(main(['search', 'export.json', 'keyword']), 0)
|
|
target.assert_called_once_with(['search', 'export.json', 'keyword'])
|
|
|
|
def test_live_probe_is_explicit(self):
|
|
from whatsapp_suite.cli import main
|
|
with patch('whatsapp_suite.runner.main', return_value=0) as target:
|
|
self.assertEqual(main([]), 0)
|
|
target.assert_called_once_with([])
|
|
|
|
|
|
class ZipExportTests(unittest.TestCase):
|
|
def test_pagination_zip_unicode_and_row_errors(self):
|
|
from whatsapp_suite.zip_export import write_messages
|
|
pages = iter([
|
|
{'count': 3, 'upper': 'c'},
|
|
{'keys': ['a', 'b'], 'last': 'b', 'messages': [{'id': 'a', 'body': 'Grüße 😺'}, {'id': 'b', 'error': 'decode_failed'}]},
|
|
{'keys': ['c'], 'last': 'c', 'messages': [{'id': 'c', 'body': 'old message'}]},
|
|
{'count': 3, 'upper': 'c'},
|
|
])
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / 'test.zip'
|
|
with zipfile.ZipFile(path, 'w') as archive, contextlib.redirect_stdout(io.StringIO()):
|
|
result = write_messages(lambda _: next(pages), archive, 2, False)
|
|
with zipfile.ZipFile(path) as archive:
|
|
self.assertIsNone(archive.testzip())
|
|
rows = [json.loads(line) for line in archive.read('messages.jsonl').decode('utf-8').splitlines()]
|
|
self.assertEqual(rows[0]['body'], 'Grüße 😺')
|
|
self.assertEqual((result['rows'], result['row_errors']), (3, 1))
|
|
self.assertTrue(result['reached_initial_upper'])
|
|
|
|
def test_duplicate_rows_fail(self):
|
|
from whatsapp_suite.zip_export import write_messages
|
|
pages = iter([{'count': 2, 'upper': 'b'}, {'keys': ['a', 'a'], 'last': 'a', 'messages': [{'id': 'a'}, {'id': 'a'}]}])
|
|
with tempfile.TemporaryDirectory() as folder, zipfile.ZipFile(Path(folder) / 'test.zip', 'w') as archive:
|
|
with self.assertRaisesRegex(RuntimeError, 'Duplicate'):
|
|
write_messages(lambda _: next(pages), archive, 2, False)
|
|
|
|
def test_existing_output_preserved(self):
|
|
from whatsapp_suite.zip_export import export_zip
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / 'test.zip'
|
|
path.write_bytes(b'original')
|
|
with self.assertRaises(ValueError):
|
|
export_zip(path)
|
|
self.assertEqual(path.read_bytes(), b'original')
|
|
|
|
def test_empty_database(self):
|
|
from whatsapp_suite.zip_export import write_messages
|
|
with tempfile.TemporaryDirectory() as folder, zipfile.ZipFile(Path(folder) / 'test.zip', 'w') as archive:
|
|
result = write_messages(lambda _: {'count': 0, 'upper': None}, archive, 100, False)
|
|
self.assertEqual(result['rows'], 0)
|
|
self.assertTrue(result['reached_initial_upper'])
|
|
|
|
|
|
class NativeCacheTests(unittest.TestCase):
|
|
@unittest.skipUnless(sys.platform == 'win32', 'Native helper cache requires Windows')
|
|
def test_verified_helper_reuses_and_repairs_cache(self):
|
|
from whatsapp_suite import common
|
|
payload=b'MZ synthetic fixture, never loaded'
|
|
digest=hashlib.sha256(payload).hexdigest()
|
|
with tempfile.TemporaryDirectory() as folder, patch.object(common,'BASE',Path(folder)), patch.object(common,'asset_bytes',return_value=payload):
|
|
path,status=common.verified_helper('fixture.dll',digest,'cache','helper.dll')
|
|
self.assertEqual(status,'extracted')
|
|
self.assertEqual(common.verified_helper('fixture.dll',digest,'cache','helper.dll')[1],'reused')
|
|
path.write_bytes(b'tampered')
|
|
self.assertEqual(common.verified_helper('fixture.dll',digest,'cache','helper.dll')[1],'repaired')
|
|
self.assertEqual(path.read_bytes(),payload)
|
|
with self.assertRaisesRegex(ValueError,'SHA-256'):
|
|
common.verified_helper('fixture.dll','0'*64,'cache','helper.dll')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|
|
|
|
class DesktopCompatibilityTests(unittest.TestCase):
|
|
def test_untested_version_and_core_are_advisory(self):
|
|
from whatsapp_suite import desktop
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path=Path(folder)/'WhatsAppDesktop_9.9.9.0_x64'/'Core.dll'
|
|
path.parent.mkdir();path.write_bytes(b'other core fixture')
|
|
result=desktop.check_core_build(path)
|
|
self.assertEqual(result['detected_build'],'9.9.9.0')
|
|
self.assertFalse(result['matches_tested_configuration'])
|
|
self.assertIn('Untested',result['warning'])
|
|
|
|
def test_tested_configuration_has_no_warning(self):
|
|
from whatsapp_suite import desktop
|
|
payload=b'tested core fixture'
|
|
with tempfile.TemporaryDirectory() as folder, patch.object(desktop,'CORE_SHA256',hashlib.sha256(payload).hexdigest()):
|
|
path=Path(folder)/('WhatsAppDesktop_'+desktop.TESTED_BUILD+'_x64')/'Core.dll'
|
|
path.parent.mkdir();path.write_bytes(payload)
|
|
result=desktop.check_core_build(path)
|
|
self.assertTrue(result['matches_tested_configuration'])
|
|
self.assertIsNone(result['warning'])
|
|
|
|
|
|
class WebhookTests(unittest.TestCase):
|
|
def test_listener_javascript_does_not_receive_transport_credentials(self):
|
|
from whatsapp_suite import webhook
|
|
config={'version':1,'endpoint':'https://example.test/events','token_env':'TEST_TOKEN',
|
|
'protected_token':'synthetic-protected-credential','self_only':True,'since_ms':1,
|
|
'listener_id':'91b17841-2106-4e73-97db-c38be6e9b269'}
|
|
def run(pid,suite,_operation):
|
|
report={'pid':1};_operation(None,None,report);return report
|
|
with patch.object(webhook,'run_math',side_effect=run), \
|
|
patch.object(webhook,'evaluate',return_value={'installed':True}) as evaluate, \
|
|
patch.object(webhook,'native_control',return_value={'running':True}), \
|
|
patch.object(webhook,'configuration_token',return_value='synthetic-token'):
|
|
webhook.manage('install',config)
|
|
script=evaluate.call_args.args[1]
|
|
self.assertNotIn(config['protected_token'],script)
|
|
self.assertNotIn(config['token_env'],script)
|
|
self.assertNotIn(config['endpoint'],script)
|
|
self.assertIn(config['listener_id'],script)
|
|
|
|
def test_listener_filters_decodes_deduplicates_and_stops(self):
|
|
if not shutil.which('node'):
|
|
self.skipTest('Node.js unavailable')
|
|
fixture=resource_bytes('tests/webhook_listener.cjs').decode('utf-8')
|
|
source=asset_bytes('webhook_listener.js').decode('utf-8')
|
|
# Node receives its fixture as argv code and the real producer through stdin.
|
|
result=subprocess.run(['node','-e',fixture],input=source,capture_output=True,
|
|
text=True,encoding='utf-8',timeout=10)
|
|
self.assertEqual(result.returncode,0,result.stderr)
|
|
|
|
def test_receiver_acknowledges_committed_events_and_deduplicates(self):
|
|
import threading
|
|
import urllib.request
|
|
import sqlite3
|
|
from whatsapp_suite.webhook import make_server
|
|
listener='91b17841-2106-4e73-97db-c38be6e9b269'
|
|
payload={'listener_id':listener,'events':[{'seq':1,'event_id':'synthetic-1',
|
|
'type':'message','message':{'id':'synthetic-1','body':'Grüße 🐈'}}]}
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
store=Path(folder)/'events.sqlite3'
|
|
server=make_server(0,store,listener)
|
|
thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
|
|
try:
|
|
url='http://127.0.0.1:'+str(server.server_port)+'/whatsapp-webhook'
|
|
results=[]
|
|
for _ in range(2):
|
|
request=urllib.request.Request(url,data=json.dumps(payload).encode(),
|
|
headers={'Content-Type':'application/json'})
|
|
with urllib.request.urlopen(request,timeout=3) as response:
|
|
results.append(json.load(response)['inserted'])
|
|
self.assertEqual(results,[1,0])
|
|
with contextlib.closing(sqlite3.connect(store)) as db:
|
|
stored=json.loads(db.execute('SELECT payload FROM events').fetchone()[0])
|
|
self.assertEqual(stored['message']['body'],'Grüße 🐈')
|
|
payload['listener_id']='other'
|
|
request=urllib.request.Request(url,data=json.dumps(payload).encode())
|
|
from urllib.error import HTTPError
|
|
with self.assertRaises(HTTPError) as error:
|
|
urllib.request.urlopen(request,timeout=3)
|
|
self.assertEqual(error.exception.code,400)
|
|
finally:
|
|
server.shutdown();server.server_close();thread.join(3)
|
|
|
|
def test_receiver_storage_failure_does_not_acknowledge(self):
|
|
import threading
|
|
import urllib.request
|
|
from urllib.error import HTTPError
|
|
import sqlite3
|
|
from whatsapp_suite import webhook
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
server=webhook.make_server(0,Path(folder)/'events.sqlite3')
|
|
thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
|
|
try:
|
|
with patch.object(webhook,'store_events',side_effect=sqlite3.OperationalError('fixture')):
|
|
request=urllib.request.Request('http://127.0.0.1:'+str(server.server_port)+'/whatsapp-webhook',data=b'{}')
|
|
with self.assertRaises(HTTPError) as error:
|
|
urllib.request.urlopen(request,timeout=3)
|
|
self.assertEqual(error.exception.code,503)
|
|
finally:
|
|
server.shutdown();server.server_close();thread.join(3)
|
|
|
|
def test_restore_loop_installs_missing_listener_without_replacing_another(self):
|
|
import threading
|
|
from whatsapp_suite import webhook
|
|
listener='91b17841-2106-4e73-97db-c38be6e9b269'
|
|
config={'version':1,'port':18763,'self_only':True,'since_ms':1,'listener_id':listener}
|
|
stopped=threading.Event();health={}
|
|
def manage(action,*args):
|
|
if action=='status':return {'passed':True,'pid':42,'listener':{'installed':False},'relay':{'running':False}}
|
|
stopped.set();return {'passed':True,'pid':42}
|
|
with patch.object(webhook,'read_config',return_value=config),patch.object(webhook,'manage',side_effect=manage) as operations:
|
|
webhook.restore_loop('fixture.json',health,stopped)
|
|
self.assertEqual([c.args[0] for c in operations.call_args_list],['status','install'])
|
|
self.assertEqual(health['restores'],1)
|
|
class OneIteration:
|
|
def __init__(self):self.done=False
|
|
def is_set(self):return self.done
|
|
def wait(self,*args):self.done=True
|
|
with patch.object(webhook,'read_config',return_value=config),patch.object(webhook,'manage',return_value={
|
|
'passed':True,'pid':42,'listener':{'installed':True,'listener_id':'other'},'relay':{'running':True}}) as operations:
|
|
webhook.restore_loop('fixture.json',health,OneIteration())
|
|
self.assertEqual(operations.call_count,1)
|
|
self.assertIn('refusing replacement',health['last_error'])
|
|
|
|
def test_receiver_configuration_and_native_asset_integrity(self):
|
|
from whatsapp_suite import webhook
|
|
config={'version':1,'port':18763,'self_only':True,'since_ms':1,
|
|
'listener_id':'91b17841-2106-4e73-97db-c38be6e9b269'}
|
|
self.assertEqual(webhook.validate_config(config),config)
|
|
with self.assertRaises(ValueError):webhook.validate_config(dict(config,port=70000))
|
|
self.assertEqual(hashlib.sha256(asset_bytes('webhook_probe.dll')).hexdigest(),webhook.HELPER_SHA256)
|
|
|
|
def test_direct_https_configuration_rejects_unsafe_routes(self):
|
|
from whatsapp_suite import webhook
|
|
config={'version':1,'self_only':True,'since_ms':1,
|
|
'listener_id':'91b17841-2106-4e73-97db-c38be6e9b269',
|
|
'endpoint':'https://receiver.example/webhook','token_env':'FIXTURE_TOKEN'}
|
|
self.assertEqual(webhook.validate_config(config),config)
|
|
for endpoint in ('http://receiver.example/webhook','https://user:[email protected]/webhook',
|
|
'https://receiver.example/webhook?token=unsafe','https://receiver.example/webhook#fragment'):
|
|
with self.assertRaises(ValueError):webhook.validate_config(dict(config,endpoint=endpoint))
|
|
|
|
@unittest.skipUnless(sys.platform=='win32','Windows DPAPI required')
|
|
def test_windows_token_protection_roundtrip_without_plaintext_in_config(self):
|
|
from whatsapp_suite import webhook
|
|
value='synthetic-webhook-token-1234567890'
|
|
protected=webhook.protect_token(value)
|
|
self.assertNotIn(value,protected)
|
|
self.assertEqual(webhook.protect_token(protected,decrypt=True),value)
|
|
|
|
def test_one_shot_restore_respects_disabled_configuration(self):
|
|
from whatsapp_suite import webhook
|
|
config={'version':1,'port':18763,'self_only':True,'since_ms':1,'enabled':False,
|
|
'listener_id':'91b17841-2106-4e73-97db-c38be6e9b269'}
|
|
with patch.object(webhook,'read_config',return_value=config),patch.object(webhook,'manage') as operation:
|
|
result=webhook.ensure_listener('fixture')
|
|
self.assertTrue(result['disabled']);operation.assert_not_called()
|