ghostwire/tests/test_whatsapp_suite.py

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()