simple updates

toxygen/tests/README.md

@@ -0,0 +1,87 @@
+# toxygen_wrapper
+
+[ctypes](https://docs.python.org/3/library/ctypes.html)
+wrapping of [Tox](https://tox.chat/)
+[```libtoxcore```](https://github.com/TokTok/c-toxcore) into Python.
+Taken from the ```wrapper``` directory of the now abandoned
+<https://github.com/toxygen-project/toxygen> ```next_gen``` branch
+by Ingvar.
+
+The basics of NGC groups are supported, as well as AV and toxencryptsave.
+There is no coverage of conferences as they are not used in ```toxygen```
+and the list of still unwrapped calls as of Sept. 2022 can be found in
+```tox.c-toxcore.missing```. The code still needs double-checking
+that every call in ```tox.py``` has the right signature, but it runs
+```toxygen``` with no apparent issues.
+
+It has been tested with UDP and TCP proxy (Tor). It has ***not*** been
+tested on Windows, and there may be some minor breakage, which should be
+easy to fix. There is a good coverage integration testsuite in ```wrapper_tests```.
+Change to that directory and run ```tests_wrapper.py --help```; the test
+suite gives a good set of examples of usage.
+
+## Install
+
+Put the parent of the wrapper directory on your PYTHONPATH and
+touch a file called `__init__.py` in its parent directory.
+
+Then you need a ```libs``` directory beside the `wrapper` directory
+and you need to link your ```libtoxcore.so``` and ```libtoxav.so```
+and ```libtoxencryptsave.so``` into it. Link all 3 filenames
+to ```libtoxcore.so``` if you have only ```libtoxcore.so```
+(which is usually the case if you built ```c-toxcore``` with ```cmake```
+rather than ```autogen/configure```). If you want to be different,
+the environment variable TOXCORE_LIBS overrides the location of ```libs```.
+
+As is, the code in ```tox.py``` is very verbose. Edit the file to change
+```
+def LOG_ERROR(a): print('EROR> '+a)
+def LOG_WARN(a): print('WARN> '+a)
+def LOG_INFO(a): print('INFO> '+a)
+def LOG_DEBUG(a): print('DBUG> '+a)
+def LOG_TRACE(a): pass # print('TRAC> '+a)
+```
+to all ```pass #``` or use ```logging.logger``` to suite your tastes.
+```logging.logger``` can be dangerous in callbacks in ```Qt``` applications,
+so we use simple print statements as default. The same applies to
+```wrapper/tests_wrapper.py```.
+
+## Prerequisites
+
+No prerequisites in Python3.
+
+## Other wrappers
+
+There are a number of other wrappings into Python of Tox core.
+This one uses [ctypes](https://docs.python.org/3/library/ctypes.html)
+which has its merits - there is no need to recompile anything as with
+Cython - change the Python file and it's done. And you can follow things
+in a Python debugger, or with the utterly stupendous Python feature of
+```gdb``` (```gdb -ex r --args /usr/bin/python3.9 <pyfile>```).
+
+CTYPES code can be brittle, segfaulting if you've got things wrong,
+but if your wrapping is right, it is very efficient and easy to work on.
+The [faulthandler](https://docs.python.org/3/library/faulthandler.html)
+module can be helpful in debugging crashes
+(e.g. from segmentation faults produced by erroneous C library wrapping).
+
+Others include:
+
+* <https://github.com/TokTok/py-toxcore-c> Cython bindings.
+  Incomplete and not really actively supported. Maybe it will get
+  worked on in the future,  but TokTok seems to be working on
+  java, rust, scalla, go, etc. bindings instead.
+  No support for NGC groups or toxencryptsave.
+
+* <https://github.com/oxij/PyTox>
+  forked from https://github.com/aitjcize/PyTox
+  by Wei-Ning Huang <aitjcize@gmail.com>.
+  Hardcore C wrapping which is not easy to keep up to date.
+  No support for NGC or toxencryptsave. Abandonned.
+  This was the basis for the TokTok/py-toxcore-c code until recently.
+
+To our point of view, the ability of CTYPEs to follow code in the
+debugger is a crucial advantage.
+
+Work on this project is suspended until the
+[MultiDevice](https://git.plastiras.org/emdee/tox_profile/wiki/MultiDevice-Announcements-POC) problem is solved. Fork me!

toxygen/tests/conference_tests.py

@@ -0,0 +1,151 @@
+if False:
+    @unittest.skip # to yet
+    def test_conference(self):
+        """
+        t:group_new
+        t:conference_delete
+        t:conference_get_chatlist_size
+        t:conference_get_chatlist
+        t:conference_send_message
+        """
+        bob_addr = self.bob.self_get_address()
+        alice_addr = self.alice.self_get_address()
+
+        self.abid = self.alice.friend_by_public_key(bob_addr)
+        self.baid = self.bob.friend_by_public_key(alice_addr)
+
+        assert self.bob_just_add_alice_as_friend()
+
+        #: Test group add
+        privacy_state = enums.TOX_GROUP_PRIVACY_STATE['PUBLIC']
+        group_name = 'test_group'
+        nick = 'test_nick'
+        status = None # dunno
+        self.group_id = self.bob.group_new(privacy_state, group_name, nick, status)
+        # :return group number on success, UINT32_MAX on failure.
+        assert self.group_id >= 0
+
+        self.loop(50)
+
+        BID = self.abid
+
+        def alices_on_conference_invite(self, fid, type_, data):
+            assert fid == BID
+            assert type_ == 0
+            gn = self.conference_join(fid, data)
+            assert type_ == self.conference_get_type(gn)
+            self.gi = True
+
+        def alices_on_conference_peer_list_changed(self, gid):
+            logging.debug("alices_on_conference_peer_list_changed")
+            assert gid == self.group_id
+            self.gn = True
+
+        try:
+            AliceTox.on_conference_invite = alices_on_conference_invite
+            AliceTox.on_conference_peer_list_changed = alices_on_conference_peer_list_changed
+
+            self.alice.gi = False
+            self.alice.gn = False
+
+            self.wait_ensure_exec(self.bob.conference_invite, (self.aid, self.group_id))
+
+            assert self.wait_callback_trues(self.alice, ['gi', 'gn'])
+        except AssertionError as e:
+            raise
+        finally:
+            AliceTox.on_conference_invite = Tox.on_conference_invite
+            AliceTox.on_conference_peer_list_change = Tox.on_conference_peer_list_changed
+
+        #: Test group number of peers
+        self.loop(50)
+        assert self.bob.conference_peer_count(self.group_id) == 2
+
+        #: Test group peername
+        self.alice.self_set_name('Alice')
+        self.bob.self_set_name('Bob')
+
+        def alices_on_conference_peer_list_changed(self, gid):
+            logging.debug("alices_on_conference_peer_list_changed")
+            self.gn = True
+        try:
+            AliceTox.on_conference_peer_list_changed = alices_on_conference_peer_list_changed
+            self.alice.gn = False
+
+            assert self.wait_callback_true(self.alice, 'gn')
+        except AssertionError as e:
+            raise
+        finally:
+            AliceTox.on_conference_peer_list_changed = Tox.on_conference_peer_list_changed
+
+        peernames = [self.bob.conference_peer_get_name(self.group_id, i) for i in
+                     range(self.bob.conference_peer_count(self.group_id))]
+        assert 'Alice' in peernames
+        assert 'Bob' in peernames
+
+        #: Test title change
+        self.bob.conference_set_title(self.group_id, 'My special title')
+        assert self.bob.conference_get_title(self.group_id) == 'My special title'
+
+        #: Test group message
+        AID = self.aid
+        BID = self.bid
+        MSG = 'Group message test'
+
+        def alices_on_conference_message(self, gid, fgid, msg_type, message):
+            logging.debug("alices_on_conference_message" +repr(message))
+            if fgid == AID:
+                assert gid == self.group_id
+                assert str(message, 'UTF-8') == MSG
+                self.alice.gm = True
+
+        try:
+            AliceTox.on_conference_message = alices_on_conference_message
+            self.alice.gm = False
+
+            self.wait_ensure_exec(self.bob.conference_send_message, (
+                self.group_id, TOX_MESSAGE_TYPE['NORMAL'], MSG))
+            assert self.wait_callback_true(self.alice, 'gm')
+        except AssertionError as e:
+            raise
+        finally:
+            AliceTox.on_conference_message = Tox.on_conference_message
+
+        #: Test group action
+        AID = self.aid
+        BID = self.bid
+        MSG = 'Group action test'
+
+        def on_conference_action(self, gid, fgid, msg_type, action):
+            if fgid == AID:
+                assert gid == self.group_id
+                assert msg_type == TOX_MESSAGE_TYPE['ACTION']
+                assert str(action, 'UTF-8') == MSG
+                self.ga = True
+
+        try:
+            AliceTox.on_conference_message = on_conference_action
+            self.alice.ga = False
+
+            self.wait_ensure_exec(self.bob.conference_send_message,
+                             (self.group_id, TOX_MESSAGE_TYPE['ACTION'], MSG))
+
+            assert self.wait_callback_true(self.alice, 'ga')
+
+            #: Test chatlist
+            assert len(self.bob.conference_get_chatlist()) == self.bob.conference_get_chatlist_size(), \
+              print(len(self.bob.conference_get_chatlist()), '!=', self.bob.conference_get_chatlist_size())
+            assert len(self.alice.conference_get_chatlist()) == self.bob.conference_get_chatlist_size(), \
+              print(len(self.alice.conference_get_chatlist()), '!=', self.bob.conference_get_chatlist_size())
+            assert self.bob.conference_get_chatlist_size() == 1, \
+              self.bob.conference_get_chatlist_size()
+            self.bob.conference_delete(self.group_id)
+            assert self.bob.conference_get_chatlist_size() == 0, \
+              self.bob.conference_get_chatlist_size()
+
+        except AssertionError as e:
+            raise
+        finally:
+            AliceTox.on_conference_message = Tox.on_conference_message
+
+

toxygen/tests/logging_toxygen_echo.py

@@ -0,0 +1,439 @@
+#!/var/local/bin/python3.bash
+#
+""" echo.py features
+ - accept friend request
+ - echo back friend message
+ - accept and answer friend call request
+ - send back friend audio/video data
+ - send back files friend sent
+"""
+
+from __future__ import print_function
+
+import sys
+import os
+import traceback
+import random
+from ctypes import *
+import argparse
+
+import time
+from os.path import exists
+
+# LOG=util.log
+global LOG
+import logging
+# log = lambda x: LOG.info(x)
+LOG = logging.getLogger('app')
+def LOG_error(a): print('EROR_ '+a)
+def LOG_warn(a):  print('WARN_ '+a)
+def LOG_info(a):  print('INFO_ '+a)
+def LOG_debug(a): print('DBUG_ '+a)
+def LOG_trace(a): pass # print('TRAC_ '+a)
+
+from middleware.tox_factory import tox_factory
+import wrapper
+import wrapper.toxcore_enums_and_consts as enums
+from wrapper.toxcore_enums_and_consts import TOX_CONNECTION, TOX_USER_STATUS, \
+    TOX_MESSAGE_TYPE, TOX_PUBLIC_KEY_SIZE, TOX_FILE_CONTROL
+import user_data
+from wrapper.libtox import LibToxCore
+import wrapper_tests.support_testing as ts
+from wrapper_tests.support_testing import oMainArgparser
+from wrapper_tests.support_testing import logging_toxygen_echo
+
+def sleep(fSec):
+    if 'QtCore' in globals():
+        if fSec > .000001: globals['QtCore'].QThread.msleep(fSec)
+        globals['QtCore'].QCoreApplication.processEvents()
+    else:
+        time.sleep(fSec)
+
+try:
+    import coloredlogs
+    if 'COLOREDLOGS_LEVEL_STYLES' not in os.environ:
+        os.environ['COLOREDLOGS_LEVEL_STYLES'] = 'spam=22;debug=28;verbose=34;notice=220;warning=202;success=118,bold;error=124;critical=background=red'
+except ImportError as e:
+    # logging.log(logging.DEBUG, f"coloredlogs not available:  {e}")
+    coloredlogs = None
+
+import wrapper_tests.support_testing as ts
+if 'USER' in os.environ:
+    sDATA_FILE = '/tmp/logging_toxygen_' +os.environ['USER'] +'.tox'
+elif 'USERNAME' in os.environ:
+    sDATA_FILE = '/tmp/logging_toxygen_' +os.environ['USERNAME'] +'.tox'
+else:
+    sDATA_FILE = '/tmp/logging_toxygen_' +'data' +'.tox'
+
+bHAVE_AV = True
+iDHT_TRIES = 100
+iDHT_TRY = 0
+
+#?SERVER = lLOCAL[-1]
+
+class AV(wrapper.tox.ToxAV):
+    def __init__(self, core):
+        super(AV, self).__init__(core)
+        self.core = self.get_tox()
+
+    def on_call(self, fid, audio_enabled, video_enabled):
+        LOG.info("Incoming %s call from %d:%s ..." % (
+            "video" if video_enabled else "audio", fid,
+            self.core.friend_get_name(fid)))
+        bret = self.answer(fid, 48, 64)
+        LOG.info(f"Answered, in call... {bret!s}")
+
+    def on_call_state(self, fid, state):
+        LOG.info('call state:fn=%d, state=%d' % (fid, state))
+
+    def on_audio_bit_rate(self, fid, audio_bit_rate):
+        LOG.info('audio bit rate status: fn=%d, abr=%d' %
+              (fid, audio_bit_rate))
+
+    def on_video_bit_rate(self, fid, video_bit_rate):
+        LOG.info('video bit rate status: fn=%d, vbr=%d' %
+              (fid, video_bit_rate))
+
+    def on_audio_receive_frame(self, fid, pcm, sample_count,
+                               channels, sampling_rate):
+        # LOG.info('audio frame: %d, %d, %d, %d' %
+        #      (fid, sample_count, channels, sampling_rate))
+        # LOG.info('pcm len:%d, %s' % (len(pcm), str(type(pcm))))
+        sys.stdout.write('.')
+        sys.stdout.flush()
+        bret = self.audio_send_frame(fid, pcm, sample_count,
+                                     channels, sampling_rate)
+        if bret is False:
+            LOG.error('on_audio_receive_frame error.')
+
+    def on_video_receive_frame(self, fid, width, height, frame, u, v):
+        LOG.info('video frame: %d, %d, %d, ' % (fid, width, height))
+        sys.stdout.write('*')
+        sys.stdout.flush()
+        bret = self.video_send_frame(fid, width, height, frame, u, v)
+        if bret is False:
+            LOG.error('on_video_receive_frame error.')
+
+    def witerate(self):
+        self.iterate()
+
+
+def save_to_file(tox, fname):
+    data = tox.get_savedata()
+    with open(fname, 'wb') as f:
+        f.write(data)
+
+def load_from_file(fname):
+    assert os.path.exists(fname)
+    return open(fname, 'rb').read()
+
+class EchoBot():
+    def __init__(self, oTox):
+        self._tox = oTox
+        self._tox.self_set_name("EchoBot")
+        LOG.info('ID: %s' % self._tox.self_get_address())
+
+        self.files = {}
+        self.av = None
+        self.on_connection_status = None
+
+    def start(self):
+        self.connect()
+        if bHAVE_AV:
+            # RuntimeError: Attempted to create a second session for the same Tox instance.
+
+            self.av = True # AV(self._tox_pointer)
+        def bobs_on_friend_request(iTox,
+                                     public_key,
+                                     message_data,
+                                     message_data_size,
+                                     *largs):
+            key = ''.join(chr(x) for x in public_key[:TOX_PUBLIC_KEY_SIZE])
+            sPk = wrapper.tox.bin_to_string(key, TOX_PUBLIC_KEY_SIZE)
+            sMd = str(message_data, 'UTF-8')
+            LOG.debug('on_friend_request ' +sPk +' ' +sMd)
+            self.on_friend_request(sPk, sMd)
+        LOG.info('setting bobs_on_friend_request')
+        self._tox.callback_friend_request(bobs_on_friend_request)
+
+        def bobs_on_friend_message(iTox,
+                                   iFriendNum,
+                                   iMessageType,
+                                   message_data,
+                                   message_data_size,
+                                   *largs):
+            sMd = str(message_data, 'UTF-8')
+            LOG_debug(f"on_friend_message  {iFriendNum}" +' ' +sMd)
+            self.on_friend_message(iFriendNum, iMessageType, sMd)
+        LOG.info('setting bobs_on_friend_message')
+        self._tox.callback_friend_message(bobs_on_friend_message)
+
+        def bobs_on_file_chunk_request(iTox, fid, filenumber, position, length, *largs):
+            if length == 0:
+                return
+
+            data = self.files[(fid, filenumber)]['f'][position:(position + length)]
+            self._tox.file_send_chunk(fid, filenumber, position, data)
+        self._tox.callback_file_chunk_request(bobs_on_file_chunk_request)
+
+        def bobs_on_file_recv(iTox, fid, filenumber, kind, size, filename, *largs):
+            LOG_info(f"on_file_recv {fid!s} {filenumber!s} {kind!s} {size!s} {filename}")
+            if size == 0:
+                return
+            self.files[(fid, filenumber)] = {
+                'f': bytes(),
+                'filename': filename,
+                'size': size
+            }
+            self._tox.file_control(fid, filenumber, TOX_FILE_CONTROL['RESUME'])
+
+
+    def connect(self):
+        if not self.on_connection_status:
+            def on_connection_status(iTox, iCon, *largs):
+                LOG_info('ON_CONNECTION_STATUS - CONNECTED ' + repr(iCon))
+            self._tox.callback_self_connection_status(on_connection_status)
+            LOG.info('setting on_connection_status callback ')
+            self.on_connection_status = on_connection_status
+        if self._oargs.network in ['newlocal', 'local']:
+            LOG.info('connecting on the new network ')
+            sNet = 'newlocal'
+        elif self._oargs.network == 'new':
+            LOG.info('connecting on the new network ')
+            sNet = 'new'
+        else: # main old
+            LOG.info('connecting on the old network ')
+            sNet = 'old'
+        sFile = self._oargs.nodes_json
+        lNodes = generate_nodes_from_file(sFile)
+        lElts = lNodes
+        random.shuffle(lElts)
+        for lElt in lElts[:10]:
+            status = self._tox.self_get_connection_status()
+            try:
+                if self._tox.bootstrap(*lElt):
+                    LOG.info('connected to ' + lElt[0]+' '+repr(status))
+                else:
+                    LOG.warn('failed connecting to ' + lElt[0])
+            except Exception as e:
+                LOG.warn('error connecting to ' + lElt[0])
+
+        if self._oargs.proxy_type > 0:
+            random.shuffle(ts.lRELAYS)
+            for lElt in ts.lRELAYS[:10]:
+                status = self._tox.self_get_connection_status()
+                try:
+                    if self._tox.add_tcp_relay(*lElt):
+                        LOG.info('relayed to ' + lElt[0] +' '+repr(status))
+                    else:
+                        LOG.warn('failed relay to ' + lElt[0])
+                except Exception as e:
+                    LOG.warn('error relay to ' + lElt[0])
+
+    def loop(self):
+        if not self.av:
+            self.start()
+        checked = False
+        save_to_file(self._tox, sDATA_FILE)
+
+        LOG.info('Starting loop.')
+        while True:
+
+            status = self._tox.self_get_connection_status()
+            if not checked and status:
+                LOG.info('Connected to DHT.')
+                checked = True
+            if not checked and not status:
+                global iDHT_TRY
+                iDHT_TRY += 10
+                self.connect()
+                self.iterate(100)
+                if iDHT_TRY >= iDHT_TRIES:
+                    raise RuntimeError("Failed to connect to the DHT.")
+                LOG.warn(f"NOT Connected to DHT. {iDHT_TRY}")
+                checked = True
+            if checked and not status:
+                LOG.info('Disconnected from DHT.')
+                self.connect()
+                checked = False
+
+            if bHAVE_AV:
+                True # self.av.witerate()
+            self.iterate(100)
+
+        LOG.info('Ending loop.')
+
+    def iterate(self, n=100):
+        interval = self._tox.iteration_interval()
+        for i in range(n):
+            self._tox.iterate()
+            sleep(interval / 1000.0)
+            self._tox.iterate()
+
+    def on_friend_request(self, pk, message):
+        LOG.debug('Friend request from %s: %s' % (pk, message))
+        self._tox.friend_add_norequest(pk)
+        LOG.info('on_friend_request Accepted.')
+        save_to_file(self._tox, sDATA_FILE)
+
+    def on_friend_message(self, friendId, type, message):
+        name = self._tox.friend_get_name(friendId)
+        LOG.debug('%s: %s' % (name, message))
+        yMessage = bytes(message, 'UTF-8')
+        self._tox.friend_send_message(friendId, TOX_MESSAGE_TYPE['NORMAL'], yMessage)
+        LOG.info('EchoBot sent: %s' % message)
+
+    def on_file_recv_chunk(self, fid, filenumber, position, data):
+        filename = self.files[(fid, filenumber)]['filename']
+        size = self.files[(fid, filenumber)]['size']
+        LOG.debug(f"on_file_recv_chunk {fid!s} {filenumber!s} {filename} {position/float(size)*100!s}")
+
+        if data is None:
+            msg = "I got '{}', sending it back right away!".format(filename)
+            self._tox.friend_send_message(fid, TOX_MESSAGE_TYPE['NORMAL'], msg)
+
+            self.files[(fid, 0)] = self.files[(fid, filenumber)]
+
+            length = self.files[(fid, filenumber)]['size']
+            self.file_send(fid, 0, length, filename, filename)
+
+            del self.files[(fid, filenumber)]
+            return
+
+        self.files[(fid, filenumber)]['f'] += data
+
+def iMain(oArgs):
+    global sDATA_FILE
+    # oTOX_OPTIONS = ToxOptions()
+    global oTOX_OPTIONS
+    oTOX_OPTIONS = oToxygenToxOptions(oArgs)
+    opts = oTOX_OPTIONS
+    if coloredlogs:
+        coloredlogs.install(
+                            level=oArgs.loglevel,
+                            logger=LOG,
+                            # %(asctime)s,%(msecs)03d %(hostname)s [%(process)d]
+                            fmt='%(name)s %(levelname)s %(message)s'
+                        )
+    else:
+        if 'logfile' in oArgs:
+            logging.basicConfig(filename=oArgs.logfile,
+                                level=oArgs.loglevel,
+                                format='%(levelname)-8s %(message)s')
+        else:
+            logging.basicConfig(level=oArgs.loglevel,
+                                format='%(levelname)-8s %(message)s')
+
+    iRet = 0
+    if hasattr(oArgs,'profile') and oArgs.profile and os.path.isfile(oArgs.profile):
+        sDATA_FILE = oArgs.profile
+        LOG.info(f"loading from  {sDATA_FILE}")
+        opts.savedata_data = load_from_file(sDATA_FILE)
+        opts.savedata_length = len(opts.savedata_data)
+        opts.savedata_type = enums.TOX_SAVEDATA_TYPE['TOX_SAVE']
+    else:
+        opts.savedata_data = None
+
+    try:
+        if False:
+            oTox = tox_factory(data=opts.savedata_data,
+                        settings=opts, args=oArgs, app=None)
+        else:
+            oTox = wrapper.tox.Tox(opts)
+        t = EchoBot(oTox)
+        t._oargs = oArgs
+        t.start()
+        t.loop()
+        save_to_file(t._tox, sDATA_FILE)
+    except KeyboardInterrupt:
+        save_to_file(t._tox, sDATA_FILE)
+    except RuntimeError as e:
+        LOG.error(f"exiting with  {e}")
+        iRet = 1
+    except Exception as e:
+        LOG.error(f"exiting with  {e}")
+        LOG.warn(' iMain(): ' \
+                     +'\n' + traceback.format_exc())
+        iRet = 1
+    return iRet
+
+def oToxygenToxOptions(oArgs, data=None):
+    tox_options = wrapper.tox.Tox.options_new()
+
+    tox_options.contents.local_discovery_enabled = False
+    tox_options.contents.dht_announcements_enabled = False
+    tox_options.contents.hole_punching_enabled = False
+    tox_options.contents.experimental_thread_safety = False
+    tox_options.contents.ipv6_enabled = False
+    tox_options.contents.tcp_port = 3390
+
+    if oArgs.proxy_type > 0:
+        tox_options.contents.proxy_type = int(oArgs.proxy_type)
+        tox_options.contents.proxy_host = bytes(oArgs.proxy_host, 'UTF-8')
+        tox_options.contents.proxy_port = int(oArgs.proxy_port)
+        tox_options.contents.udp_enabled = False
+        LOG.debug('setting oArgs.proxy_host = ' +oArgs.proxy_host)
+    else:
+        tox_options.contents.udp_enabled = True
+
+    if data:  # load existing profile
+        tox_options.contents.savedata_type = enums.TOX_SAVEDATA_TYPE['TOX_SAVE']
+        tox_options.contents.savedata_data = c_char_p(data)
+        tox_options.contents.savedata_length = len(data)
+    else:  # create new profile
+        tox_options.contents.savedata_type = enums.TOX_SAVEDATA_TYPE['NONE']
+        tox_options.contents.savedata_data = None
+        tox_options.contents.savedata_length = 0
+
+    if tox_options._options_pointer:
+        ts.vAddLoggerCallback(tox_options, ts.on_log)
+    else:
+        logging.warn("No tox_options._options_pointer " +repr(tox_options._options_pointer))
+
+    return tox_options
+
+def oArgparse(lArgv):
+    parser = ts.oMainArgparser()
+    parser.add_argument('profile', type=str, nargs='?',
+                        default=sDATA_FILE,
+                        help='Path to Tox profile to save')
+    oArgs = parser.parse_args(lArgv)
+    if hasattr(oArgs, 'sleep') and oArgs.sleep == 'qt':
+        pass # broken
+    else:
+        oArgs.sleep = 'time'
+    for key in ts.lBOOLEANS:
+        if key not in oArgs: continue
+        val = getattr(oArgs, key)
+        if val in ['False', 'false', 0]:
+            setattr(oArgs, key, False)
+        else:
+            setattr(oArgs, key, True)
+    if not os.path.exists('/proc/sys/net/ipv6') and oArgs.ipv6_enabled:
+        LOG.warn('setting oArgs.ipv6_enabled = False')
+        oArgs.ipv6_enabled = False
+    return oArgs
+
+def main(largs=None):
+    if largs is None: largs = []
+    oArgs = oArgparse(largs)
+    global     oTOX_OARGS
+    oTOX_OARGS = oArgs
+    print(oArgs)
+
+    if coloredlogs:
+        logger = logging.getLogger()
+        # https://pypi.org/project/coloredlogs/
+        coloredlogs.install(level=oArgs.loglevel,
+                        logger=logger,
+                        # %(asctime)s,%(msecs)03d %(hostname)s [%(process)d]
+                        fmt='%(name)s %(levelname)s %(message)s'
+                        )
+    else:
+        logging.basicConfig(level=oArgs.loglevel) #  logging.INFO
+
+    return iMain(oArgs)
+
+
+if __name__ == '__main__':
+    sys.exit(main(sys.argv[1:]))

toxygen/tests/socks.py

@@ -0,0 +1,391 @@
+"""SocksiPy - Python SOCKS module.
+Version 1.00
+
+Copyright 2006 Dan-Haim. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+1. Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer.
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+3. Neither the name of Dan Haim nor the names of his contributors may be used
+   to endorse or promote products derived from this software without specific
+   prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY DAN HAIM "AS IS" AND ANY EXPRESS OR IMPLIED
+WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
+EVENT SHALL DAN HAIM OR HIS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA
+OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
+OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMANGE.
+
+
+This module provides a standard socket-like interface for Python
+for tunneling connections through SOCKS proxies.
+
+"""
+
+"""
+
+Minor modifications made by Christopher Gilbert (http://motomastyle.com/)
+for use in PyLoris (http://pyloris.sourceforge.net/)
+
+Minor modifications made by Mario Vilas (http://breakingcode.wordpress.com/)
+mainly to merge bug fixes found in Sourceforge
+
+Minor modifications made by Eugene Dementiev (http://www.dementiev.eu/)
+
+"""
+
+import socket
+import struct
+import sys
+
+PROXY_TYPE_SOCKS4 = 1
+PROXY_TYPE_SOCKS5 = 2
+PROXY_TYPE_HTTP = 3
+
+_defaultproxy = None
+_orgsocket = socket.socket
+
+class ProxyError(Exception): pass
+class GeneralProxyError(ProxyError): pass
+class Socks5AuthError(ProxyError): pass
+class Socks5Error(ProxyError): pass
+class Socks4Error(ProxyError): pass
+class HTTPError(ProxyError): pass
+
+_generalerrors = ("success",
+    "invalid data",
+    "not connected",
+    "not available",
+    "bad proxy type",
+    "bad input")
+
+_socks5errors = ("succeeded",
+    "general SOCKS server failure",
+    "connection not allowed by ruleset",
+    "Network unreachable",
+    "Host unreachable",
+    "Connection refused",
+    "TTL expired",
+    "Command not supported",
+    "Address type not supported",
+    "Unknown error")
+
+_socks5autherrors = ("succeeded",
+    "authentication is required",
+    "all offered authentication methods were rejected",
+    "unknown username or invalid password",
+    "unknown error")
+
+_socks4errors = ("request granted",
+    "request rejected or failed",
+    "request rejected because SOCKS server cannot connect to identd on the client",
+    "request rejected because the client program and identd report different user-ids",
+    "unknown error")
+
+def setdefaultproxy(proxytype=None, addr=None, port=None, rdns=True, username=None, password=None):
+    """setdefaultproxy(proxytype, addr[, port[, rdns[, username[, password]]]])
+    Sets a default proxy which all further socksocket objects will use,
+    unless explicitly changed.
+    """
+    global _defaultproxy
+    _defaultproxy = (proxytype, addr, port, rdns, username, password)
+
+def wrapmodule(module):
+    """wrapmodule(module)
+    Attempts to replace a module's socket library with a SOCKS socket. Must set
+    a default proxy using setdefaultproxy(...) first.
+    This will only work on modules that import socket directly into the namespace;
+    most of the Python Standard Library falls into this category.
+    """
+    if _defaultproxy != None:
+        module.socket.socket = socksocket
+    else:
+        raise GeneralProxyError((4, "no proxy specified"))
+
+class socksocket(socket.socket):
+    """socksocket([family[, type[, proto]]]) -> socket object
+    Open a SOCKS enabled socket. The parameters are the same as
+    those of the standard socket init. In order for SOCKS to work,
+    you must specify family=AF_INET, type=SOCK_STREAM and proto=0.
+    """
+
+    def __init__(self, family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0, _sock=None):
+        _orgsocket.__init__(self, family, type, proto, _sock)
+        if _defaultproxy != None:
+            self.__proxy = _defaultproxy
+        else:
+            self.__proxy = (None, None, None, None, None, None)
+        self.__proxysockname = None
+        self.__proxypeername = None
+
+    def __recvall(self, count):
+        """__recvall(count) -> data
+        Receive EXACTLY the number of bytes requested from the socket.
+        Blocks until the required number of bytes have been received.
+        """
+        data = self.recv(count)
+        while len(data) < count:
+            d = self.recv(count-len(data))
+            if not d: raise GeneralProxyError((0, "connection closed unexpectedly"))
+            data = data + d
+        return data
+
+    def setproxy(self, proxytype=None, addr=None, port=None, rdns=True, username=None, password=None):
+        """setproxy(proxytype, addr[, port[, rdns[, username[, password]]]])
+        Sets the proxy to be used.
+        proxytype -    The type of the proxy to be used. Three types
+                are supported: PROXY_TYPE_SOCKS4 (including socks4a),
+                PROXY_TYPE_SOCKS5 and PROXY_TYPE_HTTP
+        addr -        The address of the server (IP or DNS).
+        port -        The port of the server. Defaults to 1080 for SOCKS
+                servers and 8080 for HTTP proxy servers.
+        rdns -        Should DNS queries be preformed on the remote side
+                (rather than the local side). The default is True.
+                Note: This has no effect with SOCKS4 servers.
+        username -    Username to authenticate with to the server.
+                The default is no authentication.
+        password -    Password to authenticate with to the server.
+                Only relevant when username is also provided.
+        """
+        self.__proxy = (proxytype, addr, port, rdns, username, password)
+
+    def __negotiatesocks5(self, destaddr, destport):
+        """__negotiatesocks5(self,destaddr,destport)
+        Negotiates a connection through a SOCKS5 server.
+        """
+        # First we'll send the authentication packages we support.
+        if (self.__proxy[4]!=None) and (self.__proxy[5]!=None):
+            # The username/password details were supplied to the
+            # setproxy method so we support the USERNAME/PASSWORD
+            # authentication (in addition to the standard none).
+            self.sendall(struct.pack('BBBB', 0x05, 0x02, 0x00, 0x02))
+        else:
+            # No username/password were entered, therefore we
+            # only support connections with no authentication.
+            self.sendall(struct.pack('BBB', 0x05, 0x01, 0x00))
+        # We'll receive the server's response to determine which
+        # method was selected
+        chosenauth = self.__recvall(2)
+        if chosenauth[0:1] != chr(0x05).encode():
+            self.close()
+            raise GeneralProxyError((1, _generalerrors[1]))
+        # Check the chosen authentication method
+        if chosenauth[1:2] == chr(0x00).encode():
+            # No authentication is required
+            pass
+        elif chosenauth[1:2] == chr(0x02).encode():
+            # Okay, we need to perform a basic username/password
+            # authentication.
+            self.sendall(chr(0x01).encode() + chr(len(self.__proxy[4])) + self.__proxy[4] + chr(len(self.__proxy[5])) + self.__proxy[5])
+            authstat = self.__recvall(2)
+            if authstat[0:1] != chr(0x01).encode():
+                # Bad response
+                self.close()
+                raise GeneralProxyError((1, _generalerrors[1]))
+            if authstat[1:2] != chr(0x00).encode():
+                # Authentication failed
+                self.close()
+                raise Socks5AuthError((3, _socks5autherrors[3]))
+            # Authentication succeeded
+        else:
+            # Reaching here is always bad
+            self.close()
+            if chosenauth[1] == chr(0xFF).encode():
+                raise Socks5AuthError((2, _socks5autherrors[2]))
+            else:
+                raise GeneralProxyError((1, _generalerrors[1]))
+        # Now we can request the actual connection
+        req = struct.pack('BBB', 0x05, 0x01, 0x00)
+        # If the given destination address is an IP address, we'll
+        # use the IPv4 address request even if remote resolving was specified.
+        try:
+            ipaddr = socket.inet_aton(destaddr)
+            req = req + chr(0x01).encode() + ipaddr
+        except socket.error:
+            # Well it's not an IP number,  so it's probably a DNS name.
+            if self.__proxy[3]:
+                # Resolve remotely
+                ipaddr = None
+                if type(destaddr) != type(b''): # python3
+                    destaddr_bytes = destaddr.encode(encoding='idna')
+                else:
+                    destaddr_bytes = destaddr
+                req = req + chr(0x03).encode() + chr(len(destaddr_bytes)).encode() + destaddr_bytes
+            else:
+                # Resolve locally
+                ipaddr = socket.inet_aton(socket.gethostbyname(destaddr))
+                req = req + chr(0x01).encode() + ipaddr
+        req = req + struct.pack(">H", destport)
+        self.sendall(req)
+        # Get the response
+        resp = self.__recvall(4)
+        if resp[0:1] != chr(0x05).encode():
+            self.close()
+            raise GeneralProxyError((1, _generalerrors[1]))
+        elif resp[1:2] != chr(0x00).encode():
+            # Connection failed
+            self.close()
+            if ord(resp[1:2])<=8:
+                raise Socks5Error((ord(resp[1:2]), _socks5errors[ord(resp[1:2])]))
+            else:
+                raise Socks5Error((9, _socks5errors[9]))
+        # Get the bound address/port
+        elif resp[3:4] == chr(0x01).encode():
+            boundaddr = self.__recvall(4)
+        elif resp[3:4] == chr(0x03).encode():
+            resp = resp + self.recv(1)
+            boundaddr = self.__recvall(ord(resp[4:5]))
+        else:
+            self.close()
+            raise GeneralProxyError((1,_generalerrors[1]))
+        boundport = struct.unpack(">H", self.__recvall(2))[0]
+        self.__proxysockname = (boundaddr, boundport)
+        if ipaddr != None:
+            self.__proxypeername = (socket.inet_ntoa(ipaddr), destport)
+        else:
+            self.__proxypeername = (destaddr, destport)
+
+    def getproxysockname(self):
+        """getsockname() -> address info
+        Returns the bound IP address and port number at the proxy.
+        """
+        return self.__proxysockname
+
+    def getproxypeername(self):
+        """getproxypeername() -> address info
+        Returns the IP and port number of the proxy.
+        """
+        return _orgsocket.getpeername(self)
+
+    def getpeername(self):
+        """getpeername() -> address info
+        Returns the IP address and port number of the destination
+        machine (note: getproxypeername returns the proxy)
+        """
+        return self.__proxypeername
+
+    def __negotiatesocks4(self,destaddr,destport):
+        """__negotiatesocks4(self,destaddr,destport)
+        Negotiates a connection through a SOCKS4 server.
+        """
+        # Check if the destination address provided is an IP address
+        rmtrslv = False
+        try:
+            ipaddr = socket.inet_aton(destaddr)
+        except socket.error:
+            # It's a DNS name. Check where it should be resolved.
+            if self.__proxy[3]:
+                ipaddr = struct.pack("BBBB", 0x00, 0x00, 0x00, 0x01)
+                rmtrslv = True
+            else:
+                ipaddr = socket.inet_aton(socket.gethostbyname(destaddr))
+        # Construct the request packet
+        req = struct.pack(">BBH", 0x04, 0x01, destport) + ipaddr
+        # The username parameter is considered userid for SOCKS4
+        if self.__proxy[4] != None:
+            req = req + self.__proxy[4]
+        req = req + chr(0x00).encode()
+        # DNS name if remote resolving is required
+        # NOTE: This is actually an extension to the SOCKS4 protocol
+        # called SOCKS4A and may not be supported in all cases.
+        if rmtrslv:
+            req = req + destaddr + chr(0x00).encode()
+        self.sendall(req)
+        # Get the response from the server
+        resp = self.__recvall(8)
+        if resp[0:1] != chr(0x00).encode():
+            # Bad data
+            self.close()
+            raise GeneralProxyError((1,_generalerrors[1]))
+        if resp[1:2] != chr(0x5A).encode():
+            # Server returned an error
+            self.close()
+            if ord(resp[1:2]) in (91, 92, 93):
+                self.close()
+                raise Socks4Error((ord(resp[1:2]), _socks4errors[ord(resp[1:2]) - 90]))
+            else:
+                raise Socks4Error((94, _socks4errors[4]))
+        # Get the bound address/port
+        self.__proxysockname = (socket.inet_ntoa(resp[4:]), struct.unpack(">H", resp[2:4])[0])
+        if rmtrslv != None:
+            self.__proxypeername = (socket.inet_ntoa(ipaddr), destport)
+        else:
+            self.__proxypeername = (destaddr, destport)
+
+    def __negotiatehttp(self, destaddr, destport):
+        """__negotiatehttp(self,destaddr,destport)
+        Negotiates a connection through an HTTP server.
+        """
+        # If we need to resolve locally, we do this now
+        if not self.__proxy[3]:
+            addr = socket.gethostbyname(destaddr)
+        else:
+            addr = destaddr
+        self.sendall(("CONNECT " + addr + ":" + str(destport) + " HTTP/1.1\r\n" + "Host: " + destaddr + "\r\n\r\n").encode())
+        # We read the response until we get the string "\r\n\r\n"
+        resp = self.recv(1)
+        while resp.find("\r\n\r\n".encode()) == -1:
+            recv = self.recv(1)
+            if not recv:
+                raise GeneralProxyError((1, _generalerrors[1]))
+            resp = resp + recv
+        # We just need the first line to check if the connection
+        # was successful
+        statusline = resp.splitlines()[0].split(" ".encode(), 2)
+        if statusline[0] not in ("HTTP/1.0".encode(), "HTTP/1.1".encode()):
+            self.close()
+            raise GeneralProxyError((1, _generalerrors[1]))
+        try:
+            statuscode = int(statusline[1])
+        except ValueError:
+            self.close()
+            raise GeneralProxyError((1, _generalerrors[1]))
+        if statuscode != 200:
+            self.close()
+            raise HTTPError((statuscode, statusline[2]))
+        self.__proxysockname = ("0.0.0.0", 0)
+        self.__proxypeername = (addr, destport)
+
+    def connect(self, destpair):
+        """connect(self, despair)
+        Connects to the specified destination through a proxy.
+        destpar - A tuple of the IP/DNS address and the port number.
+        (identical to socket's connect).
+        To select the proxy server use setproxy().
+        """
+        # Do a minimal input check first
+        if (not type(destpair) in (list,tuple)) or (len(destpair) < 2) or (type(destpair[0]) != type('')) or (type(destpair[1]) != int):
+            raise GeneralProxyError((5, _generalerrors[5]))
+        if self.__proxy[0] == PROXY_TYPE_SOCKS5:
+            if self.__proxy[2] != None:
+                portnum = int(self.__proxy[2])
+            else:
+                portnum = 1080
+            _orgsocket.connect(self, (self.__proxy[1], portnum))
+            self.__negotiatesocks5(destpair[0], destpair[1])
+        elif self.__proxy[0] == PROXY_TYPE_SOCKS4:
+            if self.__proxy[2] != None:
+                portnum = self.__proxy[2]
+            else:
+                portnum = 1080
+            _orgsocket.connect(self,(self.__proxy[1], portnum))
+            self.__negotiatesocks4(destpair[0], destpair[1])
+        elif self.__proxy[0] == PROXY_TYPE_HTTP:
+            if self.__proxy[2] != None:
+                portnum = self.__proxy[2]
+            else:
+                portnum = 8080
+            _orgsocket.connect(self,(self.__proxy[1], portnum))
+            self.__negotiatehttp(destpair[0], destpair[1])
+        elif self.__proxy[0] == None:
+            _orgsocket.connect(self, (destpair[0], destpair[1]))
+        else:
+            raise GeneralProxyError((4, _generalerrors[4]))

toxygen/tests/support_testing.py

@@ -0,0 +1,914 @@
+# -*- mode: python; indent-tabs-mode: nil; py-indent-offset: 4; coding: utf-8 -*-
+
+import argparse
+import contextlib
+import inspect
+import json
+import logging
+import os
+import re
+import select
+import shutil
+import socket
+import sys
+import time
+import traceback
+import unittest
+from ctypes import *
+from random import Random
+import functools
+
+random = Random()
+
+try:
+    import coloredlogs
+    if 'COLOREDLOGS_LEVEL_STYLES' not in os.environ:
+        os.environ['COLOREDLOGS_LEVEL_STYLES'] = 'spam=22;debug=28;verbose=34;notice=220;warning=202;success=118,bold;error=124;critical=background=red'
+    # https://pypi.org/project/coloredlogs/
+except ImportError as e:
+    coloredlogs = False
+try:
+    import stem
+except ImportError as e:
+    stem = False
+try:
+    import nmap
+except ImportError as e:
+    nmap = False
+
+import wrapper
+from wrapper.toxcore_enums_and_consts import TOX_CONNECTION, TOX_USER_STATUS
+
+from wrapper_tests.support_http import bAreWeConnected
+from wrapper_tests.support_onions import (is_valid_fingerprint,
+                                          lIntroductionPoints,
+                                          oGetStemController,
+                                          sMapaddressResolv, sTorResolve)
+
+try:
+    from user_data.settings import get_user_config_path
+except ImportError:
+    get_user_config_path = None
+
+# LOG=util.log
+global LOG
+LOG = logging.getLogger()
+
+def LOG_ERROR(l): print('ERRORc: '+l)
+def LOG_WARN(l):  print('WARNc: ' +l)
+def LOG_INFO(l):  print('INFOc: ' +l)
+def LOG_DEBUG(l): print('DEBUGc: '+l)
+def LOG_TRACE(l): pass # print('TRACE+ '+l)
+
+try:
+    from trepan.api import debug
+    from trepan.interfaces import server as Mserver
+except:
+#    print('trepan3 TCP server NOT available.')
+    pass
+else:
+#    print('trepan3 TCP server available.')
+    def trepan_handler(num=None, f=None):
+        connection_opts={'IO': 'TCP', 'PORT': 6666}
+        intf = Mserver.ServerInterface(connection_opts=connection_opts)
+        dbg_opts = { 'interface': intf }
+        print(f'Starting TCP server listening on port 6666.')
+        debug(dbg_opts=dbg_opts)
+        return
+
+# self._audio_thread.isAlive
+iTHREAD_TIMEOUT = 1
+iTHREAD_SLEEP = 1
+iTHREAD_JOINS = 8
+iNODES = 6
+
+lToxSamplerates = [8000, 12000, 16000, 24000, 48000]
+lToxSampleratesK = [8, 12, 16, 24, 48]
+lBOOLEANS = [
+        'local_discovery_enabled',
+        'udp_enabled',
+        'ipv6_enabled',
+        'trace_enabled',
+        'compact_mode',
+        'allow_inline',
+        'notifications',
+        'sound_notifications',
+        'calls_sound',
+        'hole_punching_enabled',
+        'dht_announcements_enabled',
+        'save_history',
+        'download_nodes_list'
+        'core_logging',
+        ]
+
+sDIR = os.environ.get('TMPDIR', '/tmp')
+sTOX_VERSION = "1000002018"
+bHAVE_NMAP = shutil.which('nmap')
+bHAVE_JQ = shutil.which('jq')
+bHAVE_BASH = shutil.which('bash')
+bHAVE_TORR = shutil.which('tor-resolve')
+
+lDEAD_BS = [
+    # Failed to resolve "tox3.plastiras.org"
+    "tox3.plastiras.org",
+    'tox.kolka.tech',
+    # IPs that do not reverse resolve
+    '49.12.229.145',
+    "46.101.197.175",
+    '114.35.245.150',
+    '172.93.52.70',
+    '195.123.208.139',
+    '205.185.115.131',
+    # IPs that do not rreverse resolve
+    'yggnode.cf', '188.225.9.167',
+    '85-143-221-42.simplecloud.ru', '85.143.221.42',
+    # IPs that do not ping
+    '104.244.74.69', 'tox.plastiras.org',
+    '195.123.208.139',
+    'gt.sot-te.ch', '32.226.5.82',
+    # suspicious IPs
+    'tox.abilinski.com', '172.103.164.250', '172.103.164.250.tpia.cipherkey.com',
+    ]
+
+
+def assert_main_thread():
+    from PyQt5 import QtCore, QtWidgets
+    from qtpy.QtWidgets import QApplication
+
+    # this "instance" method is very useful!
+    app_thread = QtWidgets.QApplication.instance().thread()
+    curr_thread = QtCore.QThread.currentThread()
+    if app_thread != curr_thread:
+        raise RuntimeError('attempt to call MainWindow.append_message from non-app thread')
+
+@contextlib.contextmanager
+def ignoreStdout():
+    devnull = os.open(os.devnull, os.O_WRONLY)
+    old_stdout = os.dup(1)
+    sys.stdout.flush()
+    os.dup2(devnull, 1)
+    os.close(devnull)
+    try:
+        yield
+    finally:
+        os.dup2(old_stdout, 1)
+        os.close(old_stdout)
+
+@contextlib.contextmanager
+def ignoreStderr():
+    devnull = os.open(os.devnull, os.O_WRONLY)
+    old_stderr = os.dup(2)
+    sys.stderr.flush()
+    os.dup2(devnull, 2)
+    os.close(devnull)
+    try:
+        yield
+    finally:
+        os.dup2(old_stderr, 2)
+        os.close(old_stderr)
+
+def clean_booleans(oArgs):
+    for key in lBOOLEANS:
+        if not hasattr(oArgs, key): continue
+        val = getattr(oArgs, key)
+        if type(val) == bool: continue
+        if val in ['False', 'false', '0']:
+            setattr(oArgs, key, False)
+        else:
+            setattr(oArgs, key, True)
+
+def on_log(iTox, level, filename, line, func, message, *data):
+    # LOG.debug(repr((level, filename, line, func, message,)))
+    tox_log_cb(level, filename, line, func, message)
+
+def tox_log_cb(level, filename, line, func, message, *args):
+    """
+    * @param level The severity of the log message.
+    * @param filename The source file from which the message originated.
+    * @param line The source line from which the message originated.
+    * @param func The function from which the message originated.
+    * @param message The log message.
+    * @param user_data The user data pointer passed to tox_new in options.
+    """
+    if type(func) == bytes:
+        func = str(func, 'utf-8')
+        message = str(message, 'UTF-8')
+        filename = str(filename, 'UTF-8')
+
+        if filename == 'network.c':
+            if line == 660: return
+                # root WARNING 3network.c#944:b'send_packet'attempted to send message with network family 10 (probably IPv6) on IPv4 socket
+            if line == 944: return
+            i = message.find('07 = GET_NODES')
+            if i > 0:
+                return
+        if filename == 'TCP_common.c': return
+
+        i = message.find(' | ')
+        if i > 0:
+            message = message[:i]
+        # message = filename +'#' +str(line) +':'+func +' '+message
+
+        name = 'core'
+        # old level is meaningless
+        level = 10 # LOG.level
+
+        # LOG._log(LOG.level, f"{level}: {message}", list())
+
+        i = message.find('(0: OK)')
+        if i > 0:
+            level = 10 # LOG.debug
+        else:
+            i = message.find('(1: ')
+            if i > 0:
+                level = 30 # LOG.warn
+            else:
+                level = 20 # LOG.info
+
+        o = LOG.makeRecord(filename, level, func, line, message, list(), None)
+        # LOG.handle(o)
+        LOG_TRACE(f"{level}: {func}{line} {message}")
+        return
+
+    elif level == 1:
+        LOG.critical(f"{level}: {message}")
+    elif level == 2:
+        LOG.error(f"{level}: {message}")
+    elif level == 3:
+        LOG.warn(f"{level}: {message}")
+    elif level == 4:
+        LOG.info(f"{level}: {message}")
+    elif level == 5:
+        LOG.debug(f"{level}: {message}")
+    else:
+        LOG_TRACE(f"{level}: {message}")
+
+def vAddLoggerCallback(tox_options, callback=None):
+    if callback is None:
+        wrapper.tox.Tox.libtoxcore.tox_options_set_log_callback(
+            tox_options._options_pointer,
+            POINTER(None)())
+        tox_options.self_logger_cb = None
+        return
+
+    c_callback = CFUNCTYPE(None, c_void_p, c_int, c_char_p, c_int, c_char_p, c_char_p, c_void_p)
+    tox_options.self_logger_cb = c_callback(callback)
+    wrapper.tox.Tox.libtoxcore.tox_options_set_log_callback(
+        tox_options._options_pointer,
+        tox_options.self_logger_cb)
+
+def get_video_indexes():
+    # Linux
+    return [str(l[5:]) for l in os.listdir('/dev/') if l.startswith('video')]
+
+def get_audio():
+    with ignoreStderr():
+        import pyaudio
+    oPyA = pyaudio.PyAudio()
+
+    input_devices = output_devices = 0
+    for i in range(oPyA.get_device_count()):
+        device = oPyA.get_device_info_by_index(i)
+        if device["maxInputChannels"]:
+            input_devices += 1
+        if device["maxOutputChannels"]:
+            output_devices += 1
+    # {'index': 21, 'structVersion': 2, 'name': 'default', 'hostApi': 0, 'maxInputChannels': 64, 'maxOutputChannels': 64, 'defaultLowInputLatency': 0.008707482993197279, 'defaultLowOutputLatency': 0.008707482993197279, 'defaultHighInputLatency': 0.034829931972789115, 'defaultHighOutputLatency': 0.034829931972789115, 'defaultSampleRate': 44100.0}
+    audio = {'input': oPyA.get_default_input_device_info()['index'] if input_devices else -1,
+             'output': oPyA.get_default_output_device_info()['index'] if output_devices else -1,
+             'enabled': input_devices and output_devices}
+    return audio
+
+def oMainArgparser(_=None, iMode=0):
+    # 'Mode: 0=chat 1=chat+audio 2=chat+audio+video default: 0'
+    if not os.path.exists('/proc/sys/net/ipv6'):
+        bIpV6 = 'False'
+    else:
+        bIpV6 = 'True'
+    lIpV6Choices=[bIpV6, 'False']
+
+    sNodesJson = os.path.join(os.environ['HOME'], '.config', 'tox', 'DHTnodes.json')
+    if not os.path.exists(sNodesJson): sNodesJson = ''
+
+    logfile = os.path.join(os.environ.get('TMPDIR', '/tmp'), 'toxygen.log')
+    if not os.path.exists(sNodesJson): logfile = ''
+
+    parser = argparse.ArgumentParser(add_help=True)
+    parser.add_argument('--proxy_host', '--proxy-host', type=str,
+                        # oddball - we want to use '' as a setting
+                        default='0.0.0.0',
+                        help='proxy host')
+    parser.add_argument('--proxy_port', '--proxy-port', default=0, type=int,
+                        help='proxy port')
+    parser.add_argument('--proxy_type', '--proxy-type', default=0, type=int,
+                        choices=[0,1,2],
+                        help='proxy type 1=http, 2=socks')
+    parser.add_argument('--tcp_port', '--tcp-port', default=0, type=int,
+                        help='tcp port')
+    parser.add_argument('--udp_enabled', type=str, default='True',
+                        choices=['True', 'False'],
+                        help='En/Disable udp')
+    parser.add_argument('--ipv6_enabled', type=str, default=bIpV6,
+                        choices=lIpV6Choices,
+                        help=f"En/Disable ipv6 - default  {bIpV6}")
+    parser.add_argument('--trace_enabled',type=str,
+                        default='True' if os.environ.get('DEBUG') else 'False',
+                        choices=['True','False'],
+                        help='Debugging from toxcore logger_trace or env DEBUG=1')
+    parser.add_argument('--download_nodes_list', type=str, default='False',
+                        choices=['True', 'False'],
+                        help='Download nodes list')
+    parser.add_argument('--nodes_json', type=str,
+                        default=sNodesJson)
+    parser.add_argument('--network', type=str,
+                        choices=['main', 'local'],
+                        default='main')
+    parser.add_argument('--download_nodes_url', type=str,
+                        default='https://nodes.tox.chat/json')
+    parser.add_argument('--logfile', default=logfile,
+                        help='Filename for logging - start with + for stdout too')
+    parser.add_argument('--loglevel', default=logging.INFO, type=int,
+                        # choices=[logging.info,logging.trace,logging.debug,logging.error]
+                        help='Threshold for logging (lower is more) default: 20')
+    parser.add_argument('--mode', type=int, default=iMode,
+                        choices=[0,1,2],
+                        help='Mode: 0=chat 1=chat+audio 2=chat+audio+video default: 0')
+    parser.add_argument('--hole_punching_enabled',type=str,
+                        default='False', choices=['True','False'],
+                        help='En/Enable hole punching')
+    parser.add_argument('--dht_announcements_enabled',type=str,
+                        default='True', choices=['True','False'],
+                        help='En/Disable DHT announcements')
+    return parser
+
+def vSetupLogging(oArgs):
+    global LOG
+    logging._defaultFormatter = logging.Formatter(datefmt='%m-%d %H:%M:%S')
+    logging._defaultFormatter.default_time_format = '%m-%d %H:%M:%S'
+    logging._defaultFormatter.default_msec_format = ''
+
+    add = None
+    kwargs = dict(level=oArgs.loglevel,
+                  format='%(levelname)-8s %(message)s')
+    if oArgs.logfile:
+        add = oArgs.logfile.startswith('+')
+        sub = oArgs.logfile.startswith('-')
+        if add or sub:
+            oArgs.logfile = oArgs.logfile[1:]
+        kwargs['filename'] = oArgs.logfile
+
+    if coloredlogs:
+        # https://pypi.org/project/coloredlogs/
+        aKw = dict(level=oArgs.loglevel,
+                   logger=LOG,
+                   stream=sys.stdout,
+                   fmt='%(name)s %(levelname)s %(message)s'
+                   )
+        coloredlogs.install(**aKw)
+        if oArgs.logfile:
+            oHandler = logging.FileHandler(oArgs.logfile)
+            LOG.addHandler(oHandler)
+    else:
+        logging.basicConfig(**kwargs)
+        if add:
+            oHandler = logging.StreamHandler(sys.stdout)
+            LOG.addHandler(oHandler)
+
+    LOG.info(f"Setting loglevel to {oArgs.loglevel!s}")
+
+
+def setup_logging(oArgs):
+    global LOG
+    if coloredlogs:
+        aKw = dict(level=oArgs.loglevel,
+                   logger=LOG,
+                   fmt='%(name)s %(levelname)s %(message)s')
+        if oArgs.logfile:
+            oFd = open(oArgs.logfile, 'wt')
+            setattr(oArgs, 'log_oFd', oFd)
+            aKw['stream'] = oFd
+        coloredlogs.install(**aKw)
+        if oArgs.logfile:
+            oHandler = logging.StreamHandler(stream=sys.stdout)
+            LOG.addHandler(oHandler)
+    else:
+        aKw = dict(level=oArgs.loglevel,
+                   format='%(name)s %(levelname)-4s %(message)s')
+        if oArgs.logfile:
+            aKw['filename'] = oArgs.logfile
+        logging.basicConfig(**aKw)
+
+    logging._defaultFormatter = logging.Formatter(datefmt='%m-%d %H:%M:%S')
+    logging._defaultFormatter.default_time_format = '%m-%d %H:%M:%S'
+    logging._defaultFormatter.default_msec_format = ''
+
+    LOG.setLevel(oArgs.loglevel)
+#    LOG.trace = lambda l: LOG.log(0, repr(l))
+    LOG.info(f"Setting loglevel to {oArgs.loglevel!s}")
+
+def signal_handler(num, f):
+    from trepan.api import debug
+    from trepan.interfaces import server as Mserver
+    connection_opts={'IO': 'TCP', 'PORT': 6666}
+    intf = Mserver.ServerInterface(connection_opts=connection_opts)
+    dbg_opts = {'interface': intf}
+    LOG.info('Starting TCP server listening on port 6666.')
+    debug(dbg_opts=dbg_opts)
+    return
+
+def merge_args_into_settings(args, settings):
+    if args:
+        if not hasattr(args, 'audio'):
+            LOG.warn('No audio ' +repr(args))
+        settings['audio'] = getattr(args, 'audio')
+        if not hasattr(args, 'video'):
+            LOG.warn('No video ' +repr(args))
+        settings['video'] = getattr(args, 'video')
+        for key in settings.keys():
+            # proxy_type proxy_port proxy_host
+            not_key = 'not_' +key
+            if hasattr(args, key):
+                val = getattr(args, key)
+                if type(val) == bytes:
+                    # proxy_host - ascii?
+                    # filenames - ascii?
+                    val = str(val, 'UTF-8')
+                settings[key] = val
+            elif hasattr(args, not_key):
+                val = not getattr(args, not_key)
+                settings[key] = val
+    clean_settings(settings)
+    return
+
+def clean_settings(self):
+    # failsafe to ensure C tox is bytes and Py settings is str
+
+    # overrides
+    self['mirror_mode'] = False
+    # REQUIRED!!
+    if not os.path.exists('/proc/sys/net/ipv6'):
+        LOG.warn('Disabling IPV6 because /proc/sys/net/ipv6 does not exist')
+        self['ipv6_enabled'] = False
+
+    if 'proxy_type' in self and self['proxy_type'] == 0:
+        self['proxy_host'] = ''
+        self['proxy_port'] = 0
+
+    if 'proxy_type' in self and self['proxy_type'] != 0 and \
+        'proxy_host' in self and self['proxy_host'] != '' and \
+        'proxy_port' in self and self['proxy_port'] != 0:
+        if 'udp_enabled' in self and self['udp_enabled']:
+            # We don't currently support UDP over proxy.
+            LOG.info("UDP enabled and proxy set: disabling UDP")
+        self['udp_enabled'] = False
+        if 'local_discovery_enabled' in self and self['local_discovery_enabled']:
+            LOG.info("local_discovery_enabled enabled and proxy set: disabling local_discovery_enabled")
+        self['local_discovery_enabled'] = False
+        if 'dht_announcements_enabled' in self and self['dht_announcements_enabled']:
+            LOG.info("dht_announcements_enabled enabled and proxy set: disabling dht_announcements_enabled")
+        self['dht_announcements_enabled'] = False
+
+    if 'auto_accept_path' in self and \
+       type(self['auto_accept_path']) == bytes:
+        self['auto_accept_path'] = str(self['auto_accept_path'], 'UTF-8')
+
+    LOG.debug("Cleaned settings")
+
+def lSdSamplerates(iDev):
+    try:
+        import sounddevice as sd
+    except ImportError:
+        return []
+    samplerates = (32000, 44100, 48000, 96000, )
+    device = iDev
+    supported_samplerates = []
+    for fs in samplerates:
+        try:
+            sd.check_output_settings(device=device, samplerate=fs)
+        except Exception as e:
+            # LOG.debug(f"Sample rate not supported {fs}" +' '+str(e))
+            pass
+        else:
+            supported_samplerates.append(fs)
+    return supported_samplerates
+
+def _get_nodes_path(oArgs=None):
+    if oArgs and oArgs.nodes_json and os.path.isfile(oArgs.nodes_json):
+        LOG.debug("_get_nodes_path: " +oArgs.nodes_json)
+        default = oArgs.nodes_json
+    elif get_user_config_path:
+        default = os.path.join(get_user_config_path(), 'toxygen_nodes.json')
+    else:
+        # Windwoes
+        default = os.path.join(os.getenv('HOME'), '.config', 'tox', 'toxygen_nodes.json')
+    LOG.debug("_get_nodes_path: " +default)
+    return default
+
+DEFAULT_NODES_COUNT = 8
+
+global aNODES
+aNODES = {}
+
+
+# @functools.lru_cache(maxsize=12) TypeError: unhashable type: 'Namespace'
+def generate_nodes(oArgs=None,
+                   nodes_count=DEFAULT_NODES_COUNT,
+                   ipv='ipv4',
+                   udp_not_tcp=True):
+    global aNODES
+    sKey = ipv
+    sKey += ',0' if udp_not_tcp else ',1'
+    if sKey in aNODES and aNODES[sKey]:
+        return aNODES[sKey]
+    sFile = _get_nodes_path(oArgs=oArgs)
+    assert os.path.exists(sFile), sFile
+    lNodes = generate_nodes_from_file(sFile,
+                                      nodes_count=nodes_count,
+                                      ipv=ipv, udp_not_tcp=udp_not_tcp)
+    assert lNodes
+    aNODES[sKey] = lNodes
+    return aNODES[sKey]
+
+aNODES_CACHE = {}
+def generate_nodes_from_file(sFile,
+                             nodes_count=DEFAULT_NODES_COUNT,
+                             ipv='ipv4',
+                             udp_not_tcp=True,
+                             ):
+    """https://github.com/TokTok/c-toxcore/issues/469
+I had a conversation with @irungentoo on IRC about whether we really need to call tox_bootstrap() when having UDP disabled and why. The answer is yes, because in addition to TCP relays (tox_add_tcp_relay()), toxcore also needs to know addresses of UDP onion nodes in order to work correctly. The DHT, however, is not used when UDP is disabled. tox_bootstrap() function resolves the address passed to it as argument and calls onion_add_bs_node_path() and DHT_bootstrap() functions. Although calling DHT_bootstrap() is not really necessary as DHT is not used, we still need to resolve the address of the DHT node in order to populate the onion routes with onion_add_bs_node_path() call.
+"""
+    global aNODES_CACHE
+
+    key = ipv
+    key += ',0' if udp_not_tcp else ',1'
+    if key in aNODES_CACHE:
+        sorted_nodes =  aNODES_CACHE[key]
+    else:
+        if not os.path.exists(sFile):
+            LOG.error("generate_nodes_from_file file not found " +sFile)
+            return []
+        try:
+            with open(sFile, 'rt') as fl:
+                json_nodes = json.loads(fl.read())['nodes']
+        except Exception as e:
+            LOG.error(f"generate_nodes_from_file error {sFile}\n{e}")
+            return []
+        else:
+            LOG.debug("generate_nodes_from_file " +sFile)
+
+        if udp_not_tcp:
+            nodes = [(node[ipv], node['port'], node['public_key'],) for
+                     node in json_nodes if node[ipv] != 'NONE' \
+                          and node["status_udp"] in [True, "true"]
+                     ]
+        else:
+            nodes = []
+            elts = [(node[ipv], node['tcp_ports'], node['public_key'],) \
+                    for node in json_nodes if node[ipv] != 'NONE' \
+                        and node["status_tcp"] in [True, "true"]
+                    ]
+            for (ipv, ports, public_key,) in elts:
+                for port in ports:
+                    nodes += [(ipv, port, public_key)]
+        if not nodes:
+            LOG.warn(f'empty generate_nodes from {sFile} {json_nodes!r}')
+            return []
+        sorted_nodes = nodes
+        aNODES_CACHE[key] = sorted_nodes
+
+    random.shuffle(sorted_nodes)
+    if nodes_count is not None and len(sorted_nodes) > nodes_count:
+        sorted_nodes = sorted_nodes[-nodes_count:]
+    LOG.debug(f"generate_nodes_from_file {sFile} len={len(sorted_nodes)}")
+    return sorted_nodes
+
+def tox_bootstrapd_port():
+    port = 33446
+    sFile = '/etc/tox-bootstrapd.conf'
+    if os.path.exists(sFile):
+        with open(sFile, 'rt') as oFd:
+            for line in  oFd.readlines():
+                if line.startswith('port = '):
+                    port = int(line[7:])
+    return port
+
+def bootstrap_local(elts, lToxes, oArgs=None):
+    if os.path.exists('/run/tox-bootstrapd/tox-bootstrapd.pid'):
+        LOG.debug('/run/tox-bootstrapd/tox-bootstrapd.pid')
+        iRet = True
+    else:
+        iRet = os.system("netstat -nle4|grep -q :33")
+        if iRet > 0:
+            LOG.warn(f'bootstraping local No local DHT running')
+    LOG.info(f'bootstraping local')
+    return bootstrap_udp(elts, lToxes, oArgs)
+
+def lDNSClean(l):
+    global lDEAD_BS
+    # list(set(l).difference(set(lDEAD_BS)))
+    return [elt for elt in l if elt not in lDEAD_BS]
+
+def lExitExcluder(oArgs, iPort=9051):
+    """
+    https://raw.githubusercontent.com/nusenu/noContactInfo_Exit_Excluder/main/exclude_noContactInfo_Exits.py
+    """
+    if not stem:
+        LOG.warn('please install the stem Python package')
+        return ''
+    LOG.debug('lExcludeExitNodes')
+
+    try:
+        controller = oGetStemController(log_level=10)
+        # generator
+        relays = controller.get_server_descriptors()
+    except Exception as e:
+        LOG.error(f'Failed to get relay descriptors {e}')
+        return None
+
+    if controller.is_set('ExcludeExitNodes'):
+        LOG.info('ExcludeExitNodes is in use already.')
+        return None
+
+    exit_excludelist=[]
+    LOG.debug("Excluded exit relays:")
+    for relay in relays:
+        if relay.exit_policy.is_exiting_allowed() and not relay.contact:
+            if is_valid_fingerprint(relay.fingerprint):
+                exit_excludelist.append(relay.fingerprint)
+                LOG.debug("https://metrics.torproject.org/rs.html#details/%s" % relay.fingerprint)
+            else:
+                LOG.warn('Invalid Fingerprint: %s' % relay.fingerprint)
+
+    try:
+        controller.set_conf('ExcludeExitNodes', exit_excludelist)
+        LOG.info('Excluded a total of %s exit relays without ContactInfo from the exit position.' % len(exit_excludelist))
+    except Exception as e:
+        LOG.exception('ExcludeExitNodes ' +str(e))
+    return exit_excludelist
+
+aHOSTS = {}
+@functools.lru_cache(maxsize=20)
+def sDNSLookup(host):
+    global aHOSTS
+    ipv = 0
+    if host in lDEAD_BS:
+#        LOG.warn(f"address skipped because in lDEAD_BS {host}")
+        return ''
+    if host in aHOSTS:
+        return aHOSTS[host]
+
+    try:
+        s = host.replace('.','')
+        int(s)
+        ipv = 4
+    except:
+        try:
+            s = host.replace(':','')
+            int(s)
+            ipv = 6
+        except: pass
+
+    if ipv > 0:
+#        LOG.debug(f"v={ipv} IP address {host}")
+        return host
+
+    LOG.debug(f"sDNSLookup {host}")
+    ip = ''
+    if host.endswith('.tox') or host.endswith('.onion'):
+        if False and stem:
+            ip = sMapaddressResolv(host)
+            if ip: return ip
+
+        ip = sTorResolve(host)
+        if ip: return ip
+
+        if not bHAVE_TORR:
+            LOG.warn(f"onion address skipped because no tor-resolve {host}")
+            return ''
+        try:
+            sout = f"/tmp/TR{os.getpid()}.log"
+            i = os.system(f"tor-resolve -4 {host} > {sout}")
+            if not i:
+                LOG.warn(f"onion address skipped because tor-resolve on {host}")
+                return ''
+            ip = open(sout, 'rt').read()
+            if ip.endswith('failed.'):
+                LOG.warn(f"onion address skipped because tor-resolve failed on {host}")
+                return ''
+            LOG.debug(f"onion address tor-resolve {ip} on {host}")
+            return ip
+        except:
+            pass
+    else:
+        try:
+            ip = socket.gethostbyname(host)
+            LOG.debug(f"host={host} gethostbyname IP address {ip}")
+            if ip:
+                aHOSTS[host] = ip
+                return ip
+            # drop through
+        except:
+            # drop through
+            pass
+
+    if ip == '':
+        try:
+            sout = f"/tmp/TR{os.getpid()}.log"
+            i = os.system(f"dig {host} +timeout=15|grep ^{host}|sed -e 's/.* //'> {sout}")
+            if not i:
+                LOG.warn(f"address skipped because dig failed on {host}")
+                return ''
+            ip = open(sout, 'rt').read().strip()
+            LOG.debug(f"address dig {ip} on {host}")
+            aHOSTS[host] = ip
+            return ip
+        except:
+            ip = host
+    LOG.debug(f'sDNSLookup {host} -> {ip}')
+    if ip and ip != host:
+        aHOSTS[host] = ip
+    return ip
+
+def bootstrap_udp(lelts, lToxes, oArgs=None):
+    lelts = lDNSClean(lelts)
+    socket.setdefaulttimeout(15.0)
+    for oTox in lToxes:
+        random.shuffle(lelts)
+        if hasattr(oTox, 'oArgs'):
+            oArgs = oTox.oArgs
+            if hasattr(oArgs, 'contents') and oArgs.contents.proxy_type != 0:
+                lelts = lelts[:1]
+
+#        LOG.debug(f'bootstrap_udp DHT bootstraping {oTox.name} {len(lelts)}')
+        for largs in lelts:
+            assert len(largs) == 3
+            host, port, key = largs
+            assert host; assert port; assert key
+            if host in lDEAD_BS: continue
+            ip = sDNSLookup(host)
+            if not ip:
+                LOG.warn(f'bootstrap_udp to host={host} port={port} did not resolve ip={ip}')
+                continue
+
+            if type(port) == str:
+                port = int(port)
+            try:
+                assert len(key) == 64, key
+                # NOT ip
+                oRet = oTox.bootstrap(host,
+                                     port,
+                                     key)
+            except Exception as e:
+                if oArgs is None or (
+                    hasattr(oArgs, 'contents') and oArgs.contents.proxy_type == 0):
+                    pass
+                    # LOG.error(f'bootstrap_udp failed to host={host} port={port} {e}')
+                continue
+            if not oRet:
+                LOG.warn(f'bootstrap_udp failed to {host} :  {oRet}')
+            elif oTox.self_get_connection_status() != TOX_CONNECTION['NONE']:
+                LOG.info(f'bootstrap_udp to {host} connected')
+                break
+            else:
+#                LOG.debug(f'bootstrap_udp to {host} not connected')
+                pass
+
+def bootstrap_tcp(lelts, lToxes, oArgs=None):
+    lelts = lDNSClean(lelts)
+    for oTox in lToxes:
+        if hasattr(oTox, 'oArgs'): oArgs = oTox.oArgs
+        random.shuffle(lelts)
+#        LOG.debug(f'bootstrap_tcp bootstapping {oTox.name} {len(lelts)}')
+        for (host, port, key,) in lelts:
+            assert host; assert port;assert key
+            if host in lDEAD_BS: continue
+            ip = sDNSLookup(host)
+            if not ip:
+                LOG.warn(f'bootstrap_tcp to {host} did not resolve ip={ip}')
+#                continue
+                ip = host
+            if host.endswith('.onion') and stem:
+                l = lIntroductionPoints(host)
+                if not l:
+                    LOG.warn(f'bootstrap_tcp to {host} has no introduction points')
+                    continue
+            if type(port) == str:
+                port = int(port)
+            try:
+                assert len(key) == 64, key
+                oRet = oTox.add_tcp_relay(ip,
+                                         port,
+                                         key)
+            except Exception as e:
+                LOG.error(f'bootstrap_tcp to {host} : ' +str(e))
+                continue
+            if not oRet:
+                LOG.warn(f'bootstrap_tcp failed to {host} : {oRet}')
+            elif oTox.mycon_time == 1:
+                LOG.info(f'bootstrap_tcp to {host} not yet connected last=1')
+            elif oTox.mycon_status is False:
+                LOG.info(f'bootstrap_tcp to {host} not True' \
+                         +f" last={int(oTox.mycon_time)}" )
+            elif oTox.self_get_connection_status() != TOX_CONNECTION['NONE']:
+                LOG.info(f'bootstrap_tcp to {host} connected' \
+                         +f" last={int(oTox.mycon_time)}" )
+                break
+            else:
+                LOG.debug(f'bootstrap_tcp to {host} but not connected' \
+                         +f" last={int(oTox.mycon_time)}" )
+                pass
+
+def iNmapInfoNmap(sProt, sHost, sPort, key=None, environ=None, cmd=''):
+    if sHost in ['-', 'NONE']: return 0
+    if not nmap: return 0
+    nmps = nmap.PortScanner
+    if sProt in ['socks', 'socks5', 'tcp4']:
+        prot = 'tcp'
+        cmd = f" -Pn -n -sT -p T:{sPort}"
+    else:
+        prot = 'udp'
+        cmd = f" -Pn -n -sU -p U:{sPort}"
+    LOG.debug(f"iNmapInfoNmap cmd={cmd}")
+    sys.stdout.flush()
+    o = nmps().scan(hosts=sHost, arguments=cmd)
+    aScan = o['scan']
+    ip = list(aScan.keys())[0]
+    state = aScan[ip][prot][sPort]['state']
+    LOG.info(f"iNmapInfoNmap: to {sHost} {state}")
+    return 0
+
+def iNmapInfo(sProt, sHost, sPort, key=None, environ=None, cmd='nmap'):
+    if sHost in ['-', 'NONE']: return 0
+    sFile = os.path.join("/tmp", f"{sHost}.{os.getpid()}.nmap")
+    if sProt in ['socks', 'socks5', 'tcp4']:
+        cmd += f" -Pn -n -sT -p T:{sPort} {sHost} | grep /tcp "
+    else:
+        cmd += f" -Pn -n -sU -p U:{sPort} {sHost} | grep /udp "
+    LOG.debug(f"iNmapInfo cmd={cmd}")
+    sys.stdout.flush()
+    iRet = os.system('sudo ' +cmd +f" >{sFile} 2>&1 ")
+    LOG.debug(f"iNmapInfo cmd={cmd} iRet={iRet}")
+    if iRet != 0:
+        return iRet
+    assert os.path.exists(sFile), sFile
+    with open(sFile, 'rt') as oFd:
+        l = oFd.readlines()
+    assert len(l)
+    l = [line for line in l if line and not line.startswith('WARNING:')]
+    s = '\n'.join([s.strip() for s in l])
+    LOG.info(f"iNmapInfo: to {sHost}\n{s}")
+    return 0
+
+def bootstrap_iNmapInfo(lElts, oArgs, protocol="tcp4", bIS_LOCAL=False, iNODES=iNODES, cmd='nmap'):
+    if not bIS_LOCAL and not bAreWeConnected():
+        LOG.warn(f"bootstrap_iNmapInfo not local and NOT CONNECTED")
+        return True
+    if os.environ['USER'] != 'root':
+        LOG.warn(f"bootstrap_iNmapInfo not ROOT")
+        return True
+
+    lRetval = []
+    for elts in lElts[:iNODES]:
+        host, port, key = elts
+        ip = sDNSLookup(host)
+        if not ip:
+            LOG.info('bootstrap_iNmapInfo to {host} did not resolve ip={ip}')
+            continue
+        if type(port) == str:
+            port = int(port)
+        iRet = -1
+        try:
+            if not nmap:
+                iRet = iNmapInfo(protocol, ip, port, key, cmd=cmd)
+            else:
+                iRet = iNmapInfoNmap(protocol, ip, port, key)
+            if iRet != 0:
+                LOG.warn('iNmapInfo to ' +repr(host) +' retval=' +str(iRet))
+                lRetval += [False]
+            else:
+                LOG.debug('iNmapInfo to ' +repr(host) +' retval=' +str(iRet))
+                lRetval += [True]
+        except Exception as e:
+            LOG.exception('iNmapInfo to {host} : ' +str(e)
+                          )
+            lRetval += [False]
+    return any(lRetval)
+
+def caseFactory(cases):
+    """We want the tests run in order."""
+    if len(cases) > 1:
+        ordered_cases = sorted(cases, key=lambda f: inspect.findsource(f)[1])
+    else:
+        ordered_cases = cases
+    return ordered_cases
+
+def suiteFactory(*testcases):
+    """We want the tests run in order."""
+    linen    = lambda f: getattr(tc, f).__code__.co_firstlineno
+    lncmp = lambda a, b: linen(a) - linen(b)
+
+    test_suite = unittest.TestSuite()
+    for tc in testcases:
+        test_suite.addTest(unittest.makeSuite(tc, sortUsing=lncmp))
+    return test_suite

toxygen/tests/test_gdb.py

@@ -0,0 +1,936 @@
+# Verify that gdb can pretty-print the various PyObject* types
+#
+# The code for testing gdb was adapted from similar work in Unladen Swallow's
+# Lib/test/test_jit_gdb.py
+
+import locale
+import os
+import re
+import subprocess
+import sys
+import sysconfig
+import textwrap
+import unittest
+
+# Is this Python configured to support threads?
+try:
+    import _thread
+except ImportError:
+    _thread = None
+
+from test import support
+from test.support import run_unittest, findfile, python_is_optimized
+
+def get_gdb_version():
+    try:
+        proc = subprocess.Popen(["gdb", "-nx", "--version"],
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE,
+                                universal_newlines=True)
+        with proc:
+            version = proc.communicate()[0]
+    except OSError:
+        # This is what "no gdb" looks like.  There may, however, be other
+        # errors that manifest this way too.
+        raise unittest.SkipTest("Couldn't find gdb on the path")
+
+    # Regex to parse:
+    # 'GNU gdb (GDB; SUSE Linux Enterprise 12) 7.7\n' -> 7.7
+    # 'GNU gdb (GDB) Fedora 7.9.1-17.fc22\n' -> 7.9
+    # 'GNU gdb 6.1.1 [FreeBSD]\n' -> 6.1
+    # 'GNU gdb (GDB) Fedora (7.5.1-37.fc18)\n' -> 7.5
+    match = re.search(r"^GNU gdb.*?\b(\d+)\.(\d+)", version)
+    if match is None:
+        raise Exception("unable to parse GDB version: %r" % version)
+    return (version, int(match.group(1)), int(match.group(2)))
+
+gdb_version, gdb_major_version, gdb_minor_version = get_gdb_version()
+if gdb_major_version < 7:
+    raise unittest.SkipTest("gdb versions before 7.0 didn't support python "
+                            "embedding. Saw %s.%s:\n%s"
+                            % (gdb_major_version, gdb_minor_version,
+                               gdb_version))
+
+if not sysconfig.is_python_build():
+    raise unittest.SkipTest("test_gdb only works on source builds at the moment.")
+
+# Location of custom hooks file in a repository checkout.
+checkout_hook_path = os.path.join(os.path.dirname(sys.executable),
+                                  'python-gdb.py')
+
+PYTHONHASHSEED = '123'
+
+def run_gdb(*args, **env_vars):
+    """Runs gdb in --batch mode with the additional arguments given by *args.
+
+    Returns its (stdout, stderr) decoded from utf-8 using the replace handler.
+    """
+    if env_vars:
+        env = os.environ.copy()
+        env.update(env_vars)
+    else:
+        env = None
+    # -nx: Do not execute commands from any .gdbinit initialization files
+    #      (issue #22188)
+    base_cmd = ('gdb', '--batch', '-nx')
+    if (gdb_major_version, gdb_minor_version) >= (7, 4):
+        base_cmd += ('-iex', 'add-auto-load-safe-path ' + checkout_hook_path)
+    proc = subprocess.Popen(base_cmd + args,
+                            # Redirect stdin to prevent GDB from messing with
+                            # the terminal settings
+                            stdin=subprocess.PIPE,
+                            stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE,
+                            env=env)
+    with proc:
+        out, err = proc.communicate()
+    return out.decode('utf-8', 'replace'), err.decode('utf-8', 'replace')
+
+# Verify that "gdb" was built with the embedded python support enabled:
+gdbpy_version, _ = run_gdb("--eval-command=python import sys; print(sys.version_info)")
+if not gdbpy_version:
+    raise unittest.SkipTest("gdb not built with embedded python support")
+
+# Verify that "gdb" can load our custom hooks, as OS security settings may
+# disallow this without a customized .gdbinit.
+_, gdbpy_errors = run_gdb('--args', sys.executable)
+if "auto-loading has been declined" in gdbpy_errors:
+    msg = "gdb security settings prevent use of custom hooks: "
+    raise unittest.SkipTest(msg + gdbpy_errors.rstrip())
+
+def gdb_has_frame_select():
+    # Does this build of gdb have gdb.Frame.select ?
+    stdout, _ = run_gdb("--eval-command=python print(dir(gdb.Frame))")
+    m = re.match(r'.*\[(.*)\].*', stdout)
+    if not m:
+        raise unittest.SkipTest("Unable to parse output from gdb.Frame.select test")
+    gdb_frame_dir = m.group(1).split(', ')
+    return "'select'" in gdb_frame_dir
+
+HAS_PYUP_PYDOWN = gdb_has_frame_select()
+
+BREAKPOINT_FN='builtin_id'
+
+@unittest.skipIf(support.PGO, "not useful for PGO")
+class DebuggerTests(unittest.TestCase):
+
+    """Test that the debugger can debug Python."""
+
+    def get_stack_trace(self, source=None, script=None,
+                        breakpoint=BREAKPOINT_FN,
+                        cmds_after_breakpoint=None,
+                        import_site=False):
+        '''
+        Run 'python -c SOURCE' under gdb with a breakpoint.
+
+        Support injecting commands after the breakpoint is reached
+
+        Returns the stdout from gdb
+
+        cmds_after_breakpoint: if provided, a list of strings: gdb commands
+        '''
+        # We use "set breakpoint pending yes" to avoid blocking with a:
+        #   Function "foo" not defined.
+        #   Make breakpoint pending on future shared library load? (y or [n])
+        # error, which typically happens python is dynamically linked (the
+        # breakpoints of interest are to be found in the shared library)
+        # When this happens, we still get:
+        #   Function "textiowrapper_write" not defined.
+        # emitted to stderr each time, alas.
+
+        # Initially I had "--eval-command=continue" here, but removed it to
+        # avoid repeated print breakpoints when traversing hierarchical data
+        # structures
+
+        # Generate a list of commands in gdb's language:
+        commands = ['set breakpoint pending yes',
+                    'break %s' % breakpoint,
+
+                    # The tests assume that the first frame of printed
+                    #  backtrace will not contain program counter,
+                    #  that is however not guaranteed by gdb
+                    #  therefore we need to use 'set print address off' to
+                    #  make sure the counter is not there. For example:
+                    # #0 in PyObject_Print ...
+                    #  is assumed, but sometimes this can be e.g.
+                    # #0 0x00003fffb7dd1798 in PyObject_Print ...
+                    'set print address off',
+
+                    'run']
+
+        # GDB as of 7.4 onwards can distinguish between the
+        # value of a variable at entry vs current value:
+        #   http://sourceware.org/gdb/onlinedocs/gdb/Variables.html
+        # which leads to the selftests failing with errors like this:
+        #   AssertionError: 'v@entry=()' != '()'
+        # Disable this:
+        if (gdb_major_version, gdb_minor_version) >= (7, 4):
+            commands += ['set print entry-values no']
+
+        if cmds_after_breakpoint:
+            commands += cmds_after_breakpoint
+        else:
+            commands += ['backtrace']
+
+        # print commands
+
+        # Use "commands" to generate the arguments with which to invoke "gdb":
+        args = ['--eval-command=%s' % cmd for cmd in commands]
+        args += ["--args",
+                 sys.executable]
+        args.extend(subprocess._args_from_interpreter_flags())
+
+        if not import_site:
+            # -S suppresses the default 'import site'
+            args += ["-S"]
+
+        if source:
+            args += ["-c", source]
+        elif script:
+            args += [script]
+
+        # print args
+        # print (' '.join(args))
+
+        # Use "args" to invoke gdb, capturing stdout, stderr:
+        out, err = run_gdb(*args, PYTHONHASHSEED=PYTHONHASHSEED)
+
+        errlines = err.splitlines()
+        unexpected_errlines = []
+
+        # Ignore some benign messages on stderr.
+        ignore_patterns = (
+            'Function "%s" not defined.' % breakpoint,
+            'Do you need "set solib-search-path" or '
+            '"set sysroot"?',
+            # BFD: /usr/lib/debug/(...): unable to initialize decompress
+            # status for section .debug_aranges
+            'BFD: ',
+            # ignore all warnings
+            'warning: ',
+            )
+        for line in errlines:
+            if not line:
+                continue
+            if not line.startswith(ignore_patterns):
+                unexpected_errlines.append(line)
+
+        # Ensure no unexpected error messages:
+        self.assertEqual(unexpected_errlines, [])
+        return out
+
+    def get_gdb_repr(self, source,
+                     cmds_after_breakpoint=None,
+                     import_site=False):
+        # Given an input python source representation of data,
+        # run "python -c'id(DATA)'" under gdb with a breakpoint on
+        # builtin_id and scrape out gdb's representation of the "op"
+        # parameter, and verify that the gdb displays the same string
+        #
+        # Verify that the gdb displays the expected string
+        #
+        # For a nested structure, the first time we hit the breakpoint will
+        # give us the top-level structure
+
+        # NOTE: avoid decoding too much of the traceback as some
+        # undecodable characters may lurk there in optimized mode
+        # (issue #19743).
+        cmds_after_breakpoint = cmds_after_breakpoint or ["backtrace 1"]
+        gdb_output = self.get_stack_trace(source, breakpoint=BREAKPOINT_FN,
+                                          cmds_after_breakpoint=cmds_after_breakpoint,
+                                          import_site=import_site)
+        # gdb can insert additional '\n' and space characters in various places
+        # in its output, depending on the width of the terminal it's connected
+        # to (using its "wrap_here" function)
+        m = re.match(r'.*#0\s+builtin_id\s+\(self\=.*,\s+v=\s*(.*?)\)\s+at\s+\S*Python/bltinmodule.c.*',
+                     gdb_output, re.DOTALL)
+        if not m:
+            self.fail('Unexpected gdb output: %r\n%s' % (gdb_output, gdb_output))
+        return m.group(1), gdb_output
+
+    def assertEndsWith(self, actual, exp_end):
+        '''Ensure that the given "actual" string ends with "exp_end"'''
+        self.assertTrue(actual.endswith(exp_end),
+                        msg='%r did not end with %r' % (actual, exp_end))
+
+    def assertMultilineMatches(self, actual, pattern):
+        m = re.match(pattern, actual, re.DOTALL)
+        if not m:
+            self.fail(msg='%r did not match %r' % (actual, pattern))
+
+    def get_sample_script(self):
+        return findfile('gdb_sample.py')
+
+class PrettyPrintTests(DebuggerTests):
+    def test_getting_backtrace(self):
+        gdb_output = self.get_stack_trace('id(42)')
+        self.assertTrue(BREAKPOINT_FN in gdb_output)
+
+    def assertGdbRepr(self, val, exp_repr=None):
+        # Ensure that gdb's rendering of the value in a debugged process
+        # matches repr(value) in this process:
+        gdb_repr, gdb_output = self.get_gdb_repr('id(' + ascii(val) + ')')
+        if not exp_repr:
+            exp_repr = repr(val)
+        self.assertEqual(gdb_repr, exp_repr,
+                         ('%r did not equal expected %r; full output was:\n%s'
+                          % (gdb_repr, exp_repr, gdb_output)))
+
+    def test_int(self):
+        'Verify the pretty-printing of various int values'
+        self.assertGdbRepr(42)
+        self.assertGdbRepr(0)
+        self.assertGdbRepr(-7)
+        self.assertGdbRepr(1000000000000)
+        self.assertGdbRepr(-1000000000000000)
+
+    def test_singletons(self):
+        'Verify the pretty-printing of True, False and None'
+        self.assertGdbRepr(True)
+        self.assertGdbRepr(False)
+        self.assertGdbRepr(None)
+
+    def test_dicts(self):
+        'Verify the pretty-printing of dictionaries'
+        self.assertGdbRepr({})
+        self.assertGdbRepr({'foo': 'bar'}, "{'foo': 'bar'}")
+        # Python preserves insertion order since 3.6
+        self.assertGdbRepr({'foo': 'bar', 'douglas': 42}, "{'foo': 'bar', 'douglas': 42}")
+
+    def test_lists(self):
+        'Verify the pretty-printing of lists'
+        self.assertGdbRepr([])
+        self.assertGdbRepr(list(range(5)))
+
+    def test_bytes(self):
+        'Verify the pretty-printing of bytes'
+        self.assertGdbRepr(b'')
+        self.assertGdbRepr(b'And now for something hopefully the same')
+        self.assertGdbRepr(b'string with embedded NUL here \0 and then some more text')
+        self.assertGdbRepr(b'this is a tab:\t'
+                           b' this is a slash-N:\n'
+                           b' this is a slash-R:\r'
+                           )
+
+        self.assertGdbRepr(b'this is byte 255:\xff and byte 128:\x80')
+
+        self.assertGdbRepr(bytes([b for b in range(255)]))
+
+    def test_strings(self):
+        'Verify the pretty-printing of unicode strings'
+        encoding = locale.getpreferredencoding()
+        def check_repr(text):
+            try:
+                text.encode(encoding)
+                printable = True
+            except UnicodeEncodeError:
+                self.assertGdbRepr(text, ascii(text))
+            else:
+                self.assertGdbRepr(text)
+
+        self.assertGdbRepr('')
+        self.assertGdbRepr('And now for something hopefully the same')
+        self.assertGdbRepr('string with embedded NUL here \0 and then some more text')
+
+        # Test printing a single character:
+        #    U+2620 SKULL AND CROSSBONES
+        check_repr('\u2620')
+
+        # Test printing a Japanese unicode string
+        # (I believe this reads "mojibake", using 3 characters from the CJK
+        # Unified Ideographs area, followed by U+3051 HIRAGANA LETTER KE)
+        check_repr('\u6587\u5b57\u5316\u3051')
+
+        # Test a character outside the BMP:
+        #    U+1D121 MUSICAL SYMBOL C CLEF
+        # This is:
+        # UTF-8: 0xF0 0x9D 0x84 0xA1
+        # UTF-16: 0xD834 0xDD21
+        check_repr(chr(0x1D121))
+
+    def test_tuples(self):
+        'Verify the pretty-printing of tuples'
+        self.assertGdbRepr(tuple(), '()')
+        self.assertGdbRepr((1,), '(1,)')
+        self.assertGdbRepr(('foo', 'bar', 'baz'))
+
+    def test_sets(self):
+        'Verify the pretty-printing of sets'
+        if (gdb_major_version, gdb_minor_version) < (7, 3):
+            self.skipTest("pretty-printing of sets needs gdb 7.3 or later")
+        self.assertGdbRepr(set(), "set()")
+        self.assertGdbRepr(set(['a']), "{'a'}")
+        # PYTHONHASHSEED is need to get the exact frozenset item order
+        if not sys.flags.ignore_environment:
+            self.assertGdbRepr(set(['a', 'b']), "{'a', 'b'}")
+            self.assertGdbRepr(set([4, 5, 6]), "{4, 5, 6}")
+
+        # Ensure that we handle sets containing the "dummy" key value,
+        # which happens on deletion:
+        gdb_repr, gdb_output = self.get_gdb_repr('''s = set(['a','b'])
+s.remove('a')
+id(s)''')
+        self.assertEqual(gdb_repr, "{'b'}")
+
+    def test_frozensets(self):
+        'Verify the pretty-printing of frozensets'
+        if (gdb_major_version, gdb_minor_version) < (7, 3):
+            self.skipTest("pretty-printing of frozensets needs gdb 7.3 or later")
+        self.assertGdbRepr(frozenset(), "frozenset()")
+        self.assertGdbRepr(frozenset(['a']), "frozenset({'a'})")
+        # PYTHONHASHSEED is need to get the exact frozenset item order
+        if not sys.flags.ignore_environment:
+            self.assertGdbRepr(frozenset(['a', 'b']), "frozenset({'a', 'b'})")
+            self.assertGdbRepr(frozenset([4, 5, 6]), "frozenset({4, 5, 6})")
+
+    def test_exceptions(self):
+        # Test a RuntimeError
+        gdb_repr, gdb_output = self.get_gdb_repr('''
+try:
+    raise RuntimeError("I am an error")
+except RuntimeError as e:
+    id(e)
+''')
+        self.assertEqual(gdb_repr,
+                         "RuntimeError('I am an error',)")
+
+
+        # Test division by zero:
+        gdb_repr, gdb_output = self.get_gdb_repr('''
+try:
+    a = 1 / 0
+except ZeroDivisionError as e:
+    id(e)
+''')
+        self.assertEqual(gdb_repr,
+                         "ZeroDivisionError('division by zero',)")
+
+    def test_modern_class(self):
+        'Verify the pretty-printing of new-style class instances'
+        gdb_repr, gdb_output = self.get_gdb_repr('''
+class Foo:
+    pass
+foo = Foo()
+foo.an_int = 42
+id(foo)''')
+        m = re.match(r'<Foo\(an_int=42\) at remote 0x-?[0-9a-f]+>', gdb_repr)
+        self.assertTrue(m,
+                        msg='Unexpected new-style class rendering %r' % gdb_repr)
+
+    def test_subclassing_list(self):
+        'Verify the pretty-printing of an instance of a list subclass'
+        gdb_repr, gdb_output = self.get_gdb_repr('''
+class Foo(list):
+    pass
+foo = Foo()
+foo += [1, 2, 3]
+foo.an_int = 42
+id(foo)''')
+        m = re.match(r'<Foo\(an_int=42\) at remote 0x-?[0-9a-f]+>', gdb_repr)
+
+        self.assertTrue(m,
+                        msg='Unexpected new-style class rendering %r' % gdb_repr)
+
+    def test_subclassing_tuple(self):
+        'Verify the pretty-printing of an instance of a tuple subclass'
+        # This should exercise the negative tp_dictoffset code in the
+        # new-style class support
+        gdb_repr, gdb_output = self.get_gdb_repr('''
+class Foo(tuple):
+    pass
+foo = Foo((1, 2, 3))
+foo.an_int = 42
+id(foo)''')
+        m = re.match(r'<Foo\(an_int=42\) at remote 0x-?[0-9a-f]+>', gdb_repr)
+
+        self.assertTrue(m,
+                        msg='Unexpected new-style class rendering %r' % gdb_repr)
+
+    def assertSane(self, source, corruption, exprepr=None):
+        '''Run Python under gdb, corrupting variables in the inferior process
+        immediately before taking a backtrace.
+
+        Verify that the variable's representation is the expected failsafe
+        representation'''
+        if corruption:
+            cmds_after_breakpoint=[corruption, 'backtrace']
+        else:
+            cmds_after_breakpoint=['backtrace']
+
+        gdb_repr, gdb_output = \
+            self.get_gdb_repr(source,
+                              cmds_after_breakpoint=cmds_after_breakpoint)
+        if exprepr:
+            if gdb_repr == exprepr:
+                # gdb managed to print the value in spite of the corruption;
+                # this is good (see http://bugs.python.org/issue8330)
+                return
+
+        # Match anything for the type name; 0xDEADBEEF could point to
+        # something arbitrary (see  http://bugs.python.org/issue8330)
+        pattern = '<.* at remote 0x-?[0-9a-f]+>'
+
+        m = re.match(pattern, gdb_repr)
+        if not m:
+            self.fail('Unexpected gdb representation: %r\n%s' % \
+                          (gdb_repr, gdb_output))
+
+    def test_NULL_ptr(self):
+        'Ensure that a NULL PyObject* is handled gracefully'
+        gdb_repr, gdb_output = (
+            self.get_gdb_repr('id(42)',
+                              cmds_after_breakpoint=['set variable v=0',
+                                                     'backtrace'])
+            )
+
+        self.assertEqual(gdb_repr, '0x0')
+
+    def test_NULL_ob_type(self):
+        'Ensure that a PyObject* with NULL ob_type is handled gracefully'
+        self.assertSane('id(42)',
+                        'set v->ob_type=0')
+
+    def test_corrupt_ob_type(self):
+        'Ensure that a PyObject* with a corrupt ob_type is handled gracefully'
+        self.assertSane('id(42)',
+                        'set v->ob_type=0xDEADBEEF',
+                        exprepr='42')
+
+    def test_corrupt_tp_flags(self):
+        'Ensure that a PyObject* with a type with corrupt tp_flags is handled'
+        self.assertSane('id(42)',
+                        'set v->ob_type->tp_flags=0x0',
+                        exprepr='42')
+
+    def test_corrupt_tp_name(self):
+        'Ensure that a PyObject* with a type with corrupt tp_name is handled'
+        self.assertSane('id(42)',
+                        'set v->ob_type->tp_name=0xDEADBEEF',
+                        exprepr='42')
+
+    def test_builtins_help(self):
+        'Ensure that the new-style class _Helper in site.py can be handled'
+
+        if sys.flags.no_site:
+            self.skipTest("need site module, but -S option was used")
+
+        # (this was the issue causing tracebacks in
+        #  http://bugs.python.org/issue8032#msg100537 )
+        gdb_repr, gdb_output = self.get_gdb_repr('id(__builtins__.help)', import_site=True)
+
+        m = re.match(r'<_Helper at remote 0x-?[0-9a-f]+>', gdb_repr)
+        self.assertTrue(m,
+                        msg='Unexpected rendering %r' % gdb_repr)
+
+    def test_selfreferential_list(self):
+        '''Ensure that a reference loop involving a list doesn't lead proxyval
+        into an infinite loop:'''
+        gdb_repr, gdb_output = \
+            self.get_gdb_repr("a = [3, 4, 5] ; a.append(a) ; id(a)")
+        self.assertEqual(gdb_repr, '[3, 4, 5, [...]]')
+
+        gdb_repr, gdb_output = \
+            self.get_gdb_repr("a = [3, 4, 5] ; b = [a] ; a.append(b) ; id(a)")
+        self.assertEqual(gdb_repr, '[3, 4, 5, [[...]]]')
+
+    def test_selfreferential_dict(self):
+        '''Ensure that a reference loop involving a dict doesn't lead proxyval
+        into an infinite loop:'''
+        gdb_repr, gdb_output = \
+            self.get_gdb_repr("a = {} ; b = {'bar':a} ; a['foo'] = b ; id(a)")
+
+        self.assertEqual(gdb_repr, "{'foo': {'bar': {...}}}")
+
+    def test_selfreferential_old_style_instance(self):
+        gdb_repr, gdb_output = \
+            self.get_gdb_repr('''
+class Foo:
+    pass
+foo = Foo()
+foo.an_attr = foo
+id(foo)''')
+        self.assertTrue(re.match(r'<Foo\(an_attr=<\.\.\.>\) at remote 0x-?[0-9a-f]+>',
+                                 gdb_repr),
+                        'Unexpected gdb representation: %r\n%s' % \
+                            (gdb_repr, gdb_output))
+
+    def test_selfreferential_new_style_instance(self):
+        gdb_repr, gdb_output = \
+            self.get_gdb_repr('''
+class Foo(object):
+    pass
+foo = Foo()
+foo.an_attr = foo
+id(foo)''')
+        self.assertTrue(re.match(r'<Foo\(an_attr=<\.\.\.>\) at remote 0x-?[0-9a-f]+>',
+                                 gdb_repr),
+                        'Unexpected gdb representation: %r\n%s' % \
+                            (gdb_repr, gdb_output))
+
+        gdb_repr, gdb_output = \
+            self.get_gdb_repr('''
+class Foo(object):
+    pass
+a = Foo()
+b = Foo()
+a.an_attr = b
+b.an_attr = a
+id(a)''')
+        self.assertTrue(re.match(r'<Foo\(an_attr=<Foo\(an_attr=<\.\.\.>\) at remote 0x-?[0-9a-f]+>\) at remote 0x-?[0-9a-f]+>',
+                                 gdb_repr),
+                        'Unexpected gdb representation: %r\n%s' % \
+                            (gdb_repr, gdb_output))
+
+    def test_truncation(self):
+        'Verify that very long output is truncated'
+        gdb_repr, gdb_output = self.get_gdb_repr('id(list(range(1000)))')
+        self.assertEqual(gdb_repr,
+                         "[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, "
+                         "14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, "
+                         "27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, "
+                         "40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, "
+                         "53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, "
+                         "66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, "
+                         "79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, "
+                         "92, 93, 94, 95, 96, 97, 98, 99, 100, 101, 102, 103, "
+                         "104, 105, 106, 107, 108, 109, 110, 111, 112, 113, "
+                         "114, 115, 116, 117, 118, 119, 120, 121, 122, 123, "
+                         "124, 125, 126, 127, 128, 129, 130, 131, 132, 133, "
+                         "134, 135, 136, 137, 138, 139, 140, 141, 142, 143, "
+                         "144, 145, 146, 147, 148, 149, 150, 151, 152, 153, "
+                         "154, 155, 156, 157, 158, 159, 160, 161, 162, 163, "
+                         "164, 165, 166, 167, 168, 169, 170, 171, 172, 173, "
+                         "174, 175, 176, 177, 178, 179, 180, 181, 182, 183, "
+                         "184, 185, 186, 187, 188, 189, 190, 191, 192, 193, "
+                         "194, 195, 196, 197, 198, 199, 200, 201, 202, 203, "
+                         "204, 205, 206, 207, 208, 209, 210, 211, 212, 213, "
+                         "214, 215, 216, 217, 218, 219, 220, 221, 222, 223, "
+                         "224, 225, 226...(truncated)")
+        self.assertEqual(len(gdb_repr),
+                         1024 + len('...(truncated)'))
+
+    def test_builtin_method(self):
+        gdb_repr, gdb_output = self.get_gdb_repr('import sys; id(sys.stdout.readlines)')
+        self.assertTrue(re.match(r'<built-in method readlines of _io.TextIOWrapper object at remote 0x-?[0-9a-f]+>',
+                                 gdb_repr),
+                        'Unexpected gdb representation: %r\n%s' % \
+                            (gdb_repr, gdb_output))
+
+    def test_frames(self):
+        gdb_output = self.get_stack_trace('''
+def foo(a, b, c):
+    pass
+
+foo(3, 4, 5)
+id(foo.__code__)''',
+                                          breakpoint='builtin_id',
+                                          cmds_after_breakpoint=['print (PyFrameObject*)(((PyCodeObject*)v)->co_zombieframe)']
+                                          )
+        self.assertTrue(re.match(r'.*\s+\$1 =\s+Frame 0x-?[0-9a-f]+, for file <string>, line 3, in foo \(\)\s+.*',
+                                 gdb_output,
+                                 re.DOTALL),
+                        'Unexpected gdb representation: %r\n%s' % (gdb_output, gdb_output))
+
+@unittest.skipIf(python_is_optimized(),
+                 "Python was compiled with optimizations")
+class PyListTests(DebuggerTests):
+    def assertListing(self, expected, actual):
+        self.assertEndsWith(actual, expected)
+
+    def test_basic_command(self):
+        'Verify that the "py-list" command works'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-list'])
+
+        self.assertListing('   5    \n'
+                           '   6    def bar(a, b, c):\n'
+                           '   7        baz(a, b, c)\n'
+                           '   8    \n'
+                           '   9    def baz(*args):\n'
+                           ' >10        id(42)\n'
+                           '  11    \n'
+                           '  12    foo(1, 2, 3)\n',
+                           bt)
+
+    def test_one_abs_arg(self):
+        'Verify the "py-list" command with one absolute argument'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-list 9'])
+
+        self.assertListing('   9    def baz(*args):\n'
+                           ' >10        id(42)\n'
+                           '  11    \n'
+                           '  12    foo(1, 2, 3)\n',
+                           bt)
+
+    def test_two_abs_args(self):
+        'Verify the "py-list" command with two absolute arguments'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-list 1,3'])
+
+        self.assertListing('   1    # Sample script for use by test_gdb.py\n'
+                           '   2    \n'
+                           '   3    def foo(a, b, c):\n',
+                           bt)
+
+class StackNavigationTests(DebuggerTests):
+    @unittest.skipUnless(HAS_PYUP_PYDOWN, "test requires py-up/py-down commands")
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_pyup_command(self):
+        'Verify that the "py-up" command works'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up', 'py-up'])
+        self.assertMultilineMatches(bt,
+                                    r'''^.*
+#[0-9]+ Frame 0x-?[0-9a-f]+, for file .*gdb_sample.py, line 7, in bar \(a=1, b=2, c=3\)
+    baz\(a, b, c\)
+$''')
+
+    @unittest.skipUnless(HAS_PYUP_PYDOWN, "test requires py-up/py-down commands")
+    def test_down_at_bottom(self):
+        'Verify handling of "py-down" at the bottom of the stack'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-down'])
+        self.assertEndsWith(bt,
+                            'Unable to find a newer python frame\n')
+
+    @unittest.skipUnless(HAS_PYUP_PYDOWN, "test requires py-up/py-down commands")
+    def test_up_at_top(self):
+        'Verify handling of "py-up" at the top of the stack'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up'] * 5)
+        self.assertEndsWith(bt,
+                            'Unable to find an older python frame\n')
+
+    @unittest.skipUnless(HAS_PYUP_PYDOWN, "test requires py-up/py-down commands")
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_up_then_down(self):
+        'Verify "py-up" followed by "py-down"'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up', 'py-up', 'py-down'])
+        self.assertMultilineMatches(bt,
+                                    r'''^.*
+#[0-9]+ Frame 0x-?[0-9a-f]+, for file .*gdb_sample.py, line 7, in bar \(a=1, b=2, c=3\)
+    baz\(a, b, c\)
+#[0-9]+ Frame 0x-?[0-9a-f]+, for file .*gdb_sample.py, line 10, in baz \(args=\(1, 2, 3\)\)
+    id\(42\)
+$''')
+
+class PyBtTests(DebuggerTests):
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_bt(self):
+        'Verify that the "py-bt" command works'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-bt'])
+        self.assertMultilineMatches(bt,
+                                    r'''^.*
+Traceback \(most recent call first\):
+  <built-in method id of module object .*>
+  File ".*gdb_sample.py", line 10, in baz
+    id\(42\)
+  File ".*gdb_sample.py", line 7, in bar
+    baz\(a, b, c\)
+  File ".*gdb_sample.py", line 4, in foo
+    bar\(a, b, c\)
+  File ".*gdb_sample.py", line 12, in <module>
+    foo\(1, 2, 3\)
+''')
+
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_bt_full(self):
+        'Verify that the "py-bt-full" command works'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-bt-full'])
+        self.assertMultilineMatches(bt,
+                                    r'''^.*
+#[0-9]+ Frame 0x-?[0-9a-f]+, for file .*gdb_sample.py, line 7, in bar \(a=1, b=2, c=3\)
+    baz\(a, b, c\)
+#[0-9]+ Frame 0x-?[0-9a-f]+, for file .*gdb_sample.py, line 4, in foo \(a=1, b=2, c=3\)
+    bar\(a, b, c\)
+#[0-9]+ Frame 0x-?[0-9a-f]+, for file .*gdb_sample.py, line 12, in <module> \(\)
+    foo\(1, 2, 3\)
+''')
+
+    @unittest.skipUnless(_thread,
+                         "Python was compiled without thread support")
+    def test_threads(self):
+        'Verify that "py-bt" indicates threads that are waiting for the GIL'
+        cmd = '''
+from threading import Thread
+
+class TestThread(Thread):
+    # These threads would run forever, but we'll interrupt things with the
+    # debugger
+    def run(self):
+        i = 0
+        while 1:
+             i += 1
+
+t = {}
+for i in range(4):
+   t[i] = TestThread()
+   t[i].start()
+
+# Trigger a breakpoint on the main thread
+id(42)
+
+'''
+        # Verify with "py-bt":
+        gdb_output = self.get_stack_trace(cmd,
+                                          cmds_after_breakpoint=['thread apply all py-bt'])
+        self.assertIn('Waiting for the GIL', gdb_output)
+
+        # Verify with "py-bt-full":
+        gdb_output = self.get_stack_trace(cmd,
+                                          cmds_after_breakpoint=['thread apply all py-bt-full'])
+        self.assertIn('Waiting for the GIL', gdb_output)
+
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    # Some older versions of gdb will fail with
+    #  "Cannot find new threads: generic error"
+    # unless we add LD_PRELOAD=PATH-TO-libpthread.so.1 as a workaround
+    @unittest.skipUnless(_thread,
+                         "Python was compiled without thread support")
+    def test_gc(self):
+        'Verify that "py-bt" indicates if a thread is garbage-collecting'
+        cmd = ('from gc import collect\n'
+               'id(42)\n'
+               'def foo():\n'
+               '    collect()\n'
+               'def bar():\n'
+               '    foo()\n'
+               'bar()\n')
+        # Verify with "py-bt":
+        gdb_output = self.get_stack_trace(cmd,
+                                          cmds_after_breakpoint=['break update_refs', 'continue', 'py-bt'],
+                                          )
+        self.assertIn('Garbage-collecting', gdb_output)
+
+        # Verify with "py-bt-full":
+        gdb_output = self.get_stack_trace(cmd,
+                                          cmds_after_breakpoint=['break update_refs', 'continue', 'py-bt-full'],
+                                          )
+        self.assertIn('Garbage-collecting', gdb_output)
+
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    # Some older versions of gdb will fail with
+    #  "Cannot find new threads: generic error"
+    # unless we add LD_PRELOAD=PATH-TO-libpthread.so.1 as a workaround
+    @unittest.skipUnless(_thread,
+                         "Python was compiled without thread support")
+    def test_pycfunction(self):
+        'Verify that "py-bt" displays invocations of PyCFunction instances'
+        # Tested function must not be defined with METH_NOARGS or METH_O,
+        # otherwise call_function() doesn't call PyCFunction_Call()
+        cmd = ('from time import gmtime\n'
+               'def foo():\n'
+               '    gmtime(1)\n'
+               'def bar():\n'
+               '    foo()\n'
+               'bar()\n')
+        # Verify with "py-bt":
+        gdb_output = self.get_stack_trace(cmd,
+                                          breakpoint='time_gmtime',
+                                          cmds_after_breakpoint=['bt', 'py-bt'],
+                                          )
+        self.assertIn('<built-in method gmtime', gdb_output)
+
+        # Verify with "py-bt-full":
+        gdb_output = self.get_stack_trace(cmd,
+                                          breakpoint='time_gmtime',
+                                          cmds_after_breakpoint=['py-bt-full'],
+                                          )
+        self.assertIn('#2 <built-in method gmtime', gdb_output)
+
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_wrapper_call(self):
+        cmd = textwrap.dedent('''
+            class MyList(list):
+                def __init__(self):
+                    super().__init__()   # wrapper_call()
+
+            id("first break point")
+            l = MyList()
+        ''')
+        # Verify with "py-bt":
+        gdb_output = self.get_stack_trace(cmd,
+                                          cmds_after_breakpoint=['break wrapper_call', 'continue', 'py-bt'])
+        self.assertRegex(gdb_output,
+                         r"<method-wrapper u?'__init__' of MyList object at ")
+
+
+class PyPrintTests(DebuggerTests):
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_basic_command(self):
+        'Verify that the "py-print" command works'
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up', 'py-print args'])
+        self.assertMultilineMatches(bt,
+                                    r".*\nlocal 'args' = \(1, 2, 3\)\n.*")
+
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    @unittest.skipUnless(HAS_PYUP_PYDOWN, "test requires py-up/py-down commands")
+    def test_print_after_up(self):
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up', 'py-up', 'py-print c', 'py-print b', 'py-print a'])
+        self.assertMultilineMatches(bt,
+                                    r".*\nlocal 'c' = 3\nlocal 'b' = 2\nlocal 'a' = 1\n.*")
+
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_printing_global(self):
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up', 'py-print __name__'])
+        self.assertMultilineMatches(bt,
+                                    r".*\nglobal '__name__' = '__main__'\n.*")
+
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_printing_builtin(self):
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up', 'py-print len'])
+        self.assertMultilineMatches(bt,
+                                    r".*\nbuiltin 'len' = <built-in method len of module object at remote 0x-?[0-9a-f]+>\n.*")
+
+class PyLocalsTests(DebuggerTests):
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_basic_command(self):
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up', 'py-locals'])
+        self.assertMultilineMatches(bt,
+                                    r".*\nargs = \(1, 2, 3\)\n.*")
+
+    @unittest.skipUnless(HAS_PYUP_PYDOWN, "test requires py-up/py-down commands")
+    @unittest.skipIf(python_is_optimized(),
+                     "Python was compiled with optimizations")
+    def test_locals_after_up(self):
+        bt = self.get_stack_trace(script=self.get_sample_script(),
+                                  cmds_after_breakpoint=['py-up', 'py-up', 'py-locals'])
+        self.assertMultilineMatches(bt,
+                                    r".*\na = 1\nb = 2\nc = 3\n.*")
+
+def test_main():
+    if support.verbose:
+        print("GDB version %s.%s:" % (gdb_major_version, gdb_minor_version))
+        for line in gdb_version.splitlines():
+            print(" " * 4 + line)
+    run_unittest(PrettyPrintTests,
+                 PyListTests,
+                 StackNavigationTests,
+                 PyBtTests,
+                 PyPrintTests,
+                 PyLocalsTests
+                 )
+
+if __name__ == "__main__":
+    test_main()

toxygen/tests/test_gdb.urls

@@ -0,0 +1 @@
+https://github.com/akheron/cpython/raw/master/Lib/test/test_gdb.py

toxygen/tests/toxygen_tests.py

@@ -0,0 +1,17 @@
+# -*- mode: python; indent-tabs-mode: nil; py-indent-offset: 4; coding: utf-8 -*-
+
+from notifications import sound
+from notifications.sound import SOUND_NOTIFICATION
+from time import sleep
+
+if True:
+    def test_sound_notification(self):
+        """
+        Plays sound notification
+        :param  type of notification
+        """
+        sound.sound_notification( SOUND_NOTIFICATION['MESSAGE'] )
+        sleep(10)
+        sound.sound_notification( SOUND_NOTIFICATION['FILE_TRANSFER'] )
+        sleep(10)
+        sound.sound_notification( None )