This commit is contained in:
emdee 2023-12-11 06:38:55 +00:00
parent cd3b636393
commit ec39c353d6
5 changed files with 732 additions and 182 deletions

File diff suppressed because it is too large Load Diff

View File

@ -3,16 +3,17 @@
from ctypes import (CFUNCTYPE, POINTER, ArgumentError, byref, c_bool, c_char_p, from ctypes import (CFUNCTYPE, POINTER, ArgumentError, byref, c_bool, c_char_p,
c_int, c_int32, c_size_t, c_uint8, c_uint16, c_uint32, c_int, c_int32, c_size_t, c_uint8, c_uint16, c_uint32,
c_void_p, cast) c_void_p, cast)
from typing import Union, Callable
from wrapper.libtox import LibToxAV from wrapper.libtox import LibToxAV
from wrapper.toxav_enums import * from wrapper.toxav_enums import *
def LOG_ERROR(a): print('EROR> '+a) def LOG_ERROR(a: str) -> None: print('EROR> '+a)
def LOG_WARN(a): print('WARN> '+a) def LOG_WARN(a: str) -> None: print('WARN> '+a)
def LOG_INFO(a): print('INFO> '+a) def LOG_INFO(a: str) -> None: print('INFO> '+a)
def LOG_DEBUG(a): print('DBUG> '+a) def LOG_DEBUG(a: str) -> None: print('DBUG> '+a)
def LOG_TRACE(a): pass # print('DEBUGx: '+a) def LOG_TRACE(a: str) -> None: pass # print('DEBUGx: '+a)
class ToxAV: class ToxAV:
""" """
@ -49,7 +50,7 @@ class ToxAV:
self.video_receive_frame_cb = None self.video_receive_frame_cb = None
self.call_cb = None self.call_cb = None
def kill(self): def kill(self) -> None:
""" """
Releases all resources associated with the A/V session. Releases all resources associated with the A/V session.
@ -58,7 +59,7 @@ class ToxAV:
""" """
self.libtoxav.toxav_kill(self._toxav_pointer) self.libtoxav.toxav_kill(self._toxav_pointer)
def get_tox_pointer(self): def get_tox_pointer(self) -> None:
""" """
Returns the Tox instance the A/V object was created for. Returns the Tox instance the A/V object was created for.
@ -69,16 +70,16 @@ class ToxAV:
# A/V event loop # A/V event loop
def iteration_interval(self): def iteration_interval(self) -> int:
""" """
Returns the interval in milliseconds when the next toxav_iterate call should be. If no call is active at the Returns the interval in milliseconds when the next toxav_iterate call should be. If no call is active at the
moment, this function returns 200. moment, this function returns 200.
:return: interval in milliseconds :return: interval in milliseconds
""" """
return self.libtoxav.toxav_iteration_interval(self._toxav_pointer) return int(self.libtoxav.toxav_iteration_interval(self._toxav_pointer))
def iterate(self): def iterate(self) -> None:
""" """
Main loop for the session. This function needs to be called in intervals of toxav_iteration_interval() Main loop for the session. This function needs to be called in intervals of toxav_iteration_interval()
milliseconds. It is best called in the separate thread from tox_iterate. milliseconds. It is best called in the separate thread from tox_iterate.
@ -87,7 +88,7 @@ class ToxAV:
# Call setup # Call setup
def call(self, friend_number, audio_bit_rate, video_bit_rate): def call(self, friend_number: int, audio_bit_rate: int, video_bit_rate: int) -> None:
""" """
Call a friend. This will start ringing the friend. Call a friend. This will start ringing the friend.
@ -121,7 +122,7 @@ class ToxAV:
elif toxav_err_call == TOXAV_ERR_CALL['INVALID_BIT_RATE']: elif toxav_err_call == TOXAV_ERR_CALL['INVALID_BIT_RATE']:
raise ArgumentError('Audio or video bit rate is invalid.') raise ArgumentError('Audio or video bit rate is invalid.')
def callback_call(self, callback, user_data): def callback_call(self, callback: Callable, user_data) -> None:
""" """
Set the callback for the `call` event. Pass None to unset. Set the callback for the `call` event. Pass None to unset.
@ -143,7 +144,7 @@ class ToxAV:
self.call_cb = c_callback(callback) self.call_cb = c_callback(callback)
self.libtoxav.toxav_callback_call(self._toxav_pointer, self.call_cb, user_data) self.libtoxav.toxav_callback_call(self._toxav_pointer, self.call_cb, user_data)
def answer(self, friend_number, audio_bit_rate, video_bit_rate): def answer(self, friend_number: int, audio_bit_rate: int, video_bit_rate: int) -> None:
""" """
Accept an incoming call. Accept an incoming call.
@ -157,8 +158,11 @@ class ToxAV:
""" """
toxav_err_answer = c_int() toxav_err_answer = c_int()
LOG_DEBUG(f"toxav_answer") LOG_DEBUG(f"toxav_answer")
result = self.libtoxav.toxav_answer(self._toxav_pointer, c_uint32(friend_number), c_uint32(audio_bit_rate), result = self.libtoxav.toxav_answer(self._toxav_pointer,
c_uint32(video_bit_rate), byref(toxav_err_answer)) c_uint32(friend_number),
c_uint32(audio_bit_rate),
c_uint32(video_bit_rate),
byref(toxav_err_answer))
toxav_err_answer = toxav_err_answer.value toxav_err_answer = toxav_err_answer.value
if toxav_err_answer == TOXAV_ERR_ANSWER['OK']: if toxav_err_answer == TOXAV_ERR_ANSWER['OK']:
return bool(result) return bool(result)
@ -177,7 +181,7 @@ class ToxAV:
# Call state graph # Call state graph
def callback_call_state(self, callback, user_data): def callback_call_state(self, callback: Callable, user_data) -> None:
""" """
Set the callback for the `call_state` event. Pass None to unset. Set the callback for the `call_state` event. Pass None to unset.
@ -202,7 +206,7 @@ class ToxAV:
# Call control # Call control
def call_control(self, friend_number, control): def call_control(self, friend_number: int, control: int) -> None:
""" """
Sends a call control command to a friend. Sends a call control command to a friend.
@ -232,7 +236,7 @@ class ToxAV:
# A/V sending # A/V sending
def audio_send_frame(self, friend_number, pcm, sample_count, channels, sampling_rate): def audio_send_frame(self, friend_number: int, pcm, sample_count: int, channels: int, sampling_rate: int) -> None:
""" """
Send an audio frame to a friend. Send an audio frame to a friend.
@ -277,7 +281,7 @@ class ToxAV:
elif toxav_err_send_frame == TOXAV_ERR_SEND_FRAME['RTP_FAILED']: elif toxav_err_send_frame == TOXAV_ERR_SEND_FRAME['RTP_FAILED']:
RuntimeError('Failed to push frame through rtp interface.') RuntimeError('Failed to push frame through rtp interface.')
def video_send_frame(self, friend_number, width, height, y, u, v): def video_send_frame(self, friend_number: int, width: int, height: int, y, u, v) -> None:
""" """
Send a video frame to a friend. Send a video frame to a friend.
@ -294,9 +298,14 @@ class ToxAV:
""" """
toxav_err_send_frame = c_int() toxav_err_send_frame = c_int()
LOG_TRACE(f"toxav_video_send_frame") LOG_TRACE(f"toxav_video_send_frame")
result = self.libtoxav.toxav_video_send_frame(self._toxav_pointer, c_uint32(friend_number), c_uint16(width), result = self.libtoxav.toxav_video_send_frame(self._toxav_pointer,
c_uint16(height), c_char_p(y), c_char_p(u), c_char_p(v), c_uint32(friend_number),
byref(toxav_err_send_frame)) c_uint16(width),
c_uint16(height),
c_char_p(y),
c_char_p(u),
c_char_p(v),
byref(toxav_err_send_frame))
toxav_err_send_frame = toxav_err_send_frame.value toxav_err_send_frame = toxav_err_send_frame.value
if toxav_err_send_frame == TOXAV_ERR_SEND_FRAME['OK']: if toxav_err_send_frame == TOXAV_ERR_SEND_FRAME['OK']:
return bool(result) return bool(result)
@ -319,7 +328,7 @@ class ToxAV:
# A/V receiving # A/V receiving
def callback_audio_receive_frame(self, callback, user_data): def callback_audio_receive_frame(self, callback: Callable, user_data) -> None:
""" """
Set the callback for the `audio_receive_frame` event. Pass None to unset. Set the callback for the `audio_receive_frame` event. Pass None to unset.
@ -346,7 +355,7 @@ class ToxAV:
self.audio_receive_frame_cb = c_callback(callback) self.audio_receive_frame_cb = c_callback(callback)
self.libtoxav.toxav_callback_audio_receive_frame(self._toxav_pointer, self.audio_receive_frame_cb, user_data) self.libtoxav.toxav_callback_audio_receive_frame(self._toxav_pointer, self.audio_receive_frame_cb, user_data)
def callback_video_receive_frame(self, callback, user_data): def callback_video_receive_frame(self, callback: Callable, user_data) -> None:
""" """
Set the callback for the `video_receive_frame` event. Pass None to unset. Set the callback for the `video_receive_frame` event. Pass None to unset.

479
wrapper/toxygen_echo.py Normal file
View File

@ -0,0 +1,479 @@
#!/var/local/bin/python3.bash
# -*- mode: python; indent-tabs-mode: nil; py-indent-offset: 4; coding: utf-8 -*-
# A work in progress - chat works.
""" 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
"""
import sys
import os
import traceback
import threading
import random
from ctypes import *
import argparse
from time import sleep
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)
import wrapper
import wrapper.toxcore_enums_and_consts as enums
from wrapper.tox import Tox, UINT32_MAX, ToxError
from wrapper.toxcore_enums_and_consts import TOX_CONNECTION, TOX_USER_STATUS, \
TOX_MESSAGE_TYPE, TOX_PUBLIC_KEY_SIZE, TOX_FILE_CONTROL
import wrapper_tests.support_testing as ts
from wrapper_tests.support_testing import oMainArgparser
import time
def sleep(fSec):
if 'QtCore' in globals():
if fSec > .000001: QtCore.QThread.msleep(fSec)
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 = False
iDHT_TRIES = 100
iDHT_TRY = 0
#?SERVER = lLOCAL[-1]
if not bHAVE_AV:
class AV(): pass
else:
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("PyEchoBot")
LOG.info(f'ID: {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 = ts.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(lElts)
for lElt in lElts[: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
class App():
def __init__(self):
self.mode = 0
oAPP = App()
class EchobotTox(Tox):
def __init__(self, opts, app=None):
super(EchobotTox, self).__init__(opts, app=app)
self._address = self.self_get_address()
self.name = 'pyechobot'
self._opts = opts
self._app = app
class BaseThread(threading.Thread):
def __init__(self, name=None, target=None):
if name:
super().__init__(name=name, target=target)
else:
super().__init__(target=target)
self._stop_thread = False
self.name = name
def stop_thread(self, timeout=-1):
self._stop_thread = True
if timeout < 0:
timeout = ts.iTHREAD_TIMEOUT
i = 0
while i < ts.iTHREAD_JOINS:
self.join(timeout)
if not self.is_alive(): break
i = i + 1
else:
LOG.warning(f"{self.name} BLOCKED")
class ToxIterateThread(BaseThread):
def __init__(self, tox):
super().__init__(name='ToxIterateThread')
self._tox = tox
def run(self):
while not self._stop_thread:
self._tox.iterate()
sleep(self._tox.iteration_interval() / 1000)
def oArgparse(lArgv):
parser = ts.oMainArgparser()
parser.add_argument('--norequest',type=str, default='False',
choices=['True','False'],
help='Use _norequest')
parser.add_argument('profile', type=str, nargs='?', default=None,
help='Path to Tox profile')
oArgs = parser.parse_args(lArgv)
for key in ts.lBOOLEANS:
if key not in oArgs: continue
val = getattr(oArgs, key)
setattr(oArgs, key, bool(val))
if hasattr(oArgs, 'sleep'):
if oArgs.sleep == 'qt':
pass # broken or gevent.sleep(idle_period)
elif oArgs.sleep == 'gevent':
pass # broken or gevent.sleep(idle_period)
else:
oArgs.sleep = 'time'
return oArgs
def iMain(oArgs):
global sDATA_FILE
# oTOX_OPTIONS = ToxOptions()
global oTOX_OPTIONS
oMainArgparser
oTOX_OPTIONS = ts.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:
oTox = EchobotTox(opts, app=oAPP)
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"ERROR {e}")
iRet = 1
except Exception as e:
LOG.error(f"EXCEPTION {e}")
LOG.warn(' iMain(): ' \
+'\n' + traceback.format_exc())
iRet = 1
return iRet
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)
def main(lArgs=None):
global oTOX_OARGS
if lArgs is None: lArgs = []
oArgs = oArgparse(lArgs)
global bIS_LOCAL
bIS_LOCAL = oArgs.network in ['newlocal', 'localnew', 'local']
oTOX_OARGS = oArgs
setattr(oTOX_OARGS, 'bIS_LOCAL', bIS_LOCAL)
bIS_LOCAL = True
setattr(oTOX_OARGS, 'bIS_LOCAL', bIS_LOCAL)
# oTOX_OPTIONS = ToxOptions()
global oTOX_OPTIONS
oTOX_OPTIONS = ts.oToxygenToxOptions(oArgs)
if coloredlogs:
# https://pypi.org/project/coloredlogs/
coloredlogs.install(level=oArgs.loglevel,
logger=LOG,
# %(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:]))

View File

@ -37,7 +37,7 @@ except ImportError as e:
nmap = False nmap = False
import wrapper import wrapper
from wrapper.toxcore_enums_and_consts import TOX_CONNECTION, TOX_USER_STATUS import wrapper.toxcore_enums_and_consts as enums
from wrapper_tests.support_http import bAreWeConnected from wrapper_tests.support_http import bAreWeConnected
from wrapper_tests.support_onions import (is_valid_fingerprint, from wrapper_tests.support_onions import (is_valid_fingerprint,
@ -281,6 +281,53 @@ def get_audio():
'enabled': input_devices and output_devices} 'enabled': input_devices and output_devices}
return audio return audio
def oToxygenToxOptions(oArgs):
data = None
tox_options = wrapper.tox.Tox.options_new()
if oArgs.proxy_type:
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
else:
tox_options.contents.udp_enabled = oArgs.udp_enabled
if not os.path.exists('/proc/sys/net/ipv6'):
oArgs.ipv6_enabled = False
else:
tox_options.contents.ipv6_enabled = oArgs.ipv6_enabled
tox_options.contents.tcp_port = int(oArgs.tcp_port)
tox_options.contents.dht_announcements_enabled = oArgs.dht_announcements_enabled
tox_options.contents.hole_punching_enabled = oArgs.hole_punching_enabled
# overrides
tox_options.contents.local_discovery_enabled = False
tox_options.contents.experimental_thread_safety = False
# REQUIRED!!
if oArgs.ipv6_enabled and not os.path.exists('/proc/sys/net/ipv6'):
LOG.warning('Disabling IPV6 because /proc/sys/net/ipv6 does not exist' + repr(oArgs.ipv6_enabled))
tox_options.contents.ipv6_enabled = False
else:
tox_options.contents.ipv6_enabled = bool(oArgs.ipv6_enabled)
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
#? tox_options.contents.log_callback = LOG
if tox_options._options_pointer:
# LOG.debug("Adding logging to tox_options._options_pointer ")
vAddLoggerCallback(tox_options, on_log)
else:
LOG.warning("No tox_options._options_pointer " +repr(tox_options._options_pointer))
return tox_options
def oMainArgparser(_=None, iMode=0): def oMainArgparser(_=None, iMode=0):
# 'Mode: 0=chat 1=chat+audio 2=chat+audio+video default: 0' # 'Mode: 0=chat 1=chat+audio 2=chat+audio+video default: 0'
if not os.path.exists('/proc/sys/net/ipv6'): if not os.path.exists('/proc/sys/net/ipv6'):
@ -769,7 +816,7 @@ def bootstrap_udp(lelts, lToxes, oArgs=None):
continue continue
if not oRet: if not oRet:
LOG.warn(f'bootstrap_udp failed to {host} : {oRet}') LOG.warn(f'bootstrap_udp failed to {host} : {oRet}')
elif oTox.self_get_connection_status() != TOX_CONNECTION['NONE']: elif oTox.self_get_connection_status() != enums.TOX_CONNECTION['NONE']:
LOG.info(f'bootstrap_udp to {host} connected') LOG.info(f'bootstrap_udp to {host} connected')
break break
else: else:
@ -813,7 +860,7 @@ def bootstrap_tcp(lelts, lToxes, oArgs=None):
elif hasattr(oTox, 'mycon_status') and oTox.mycon_status is False: elif hasattr(oTox, 'mycon_status') and oTox.mycon_status is False:
LOG.info(f'bootstrap_tcp to {host} not True' \ LOG.info(f'bootstrap_tcp to {host} not True' \
+f" last={int(oTox.mycon_time)}" ) +f" last={int(oTox.mycon_time)}" )
elif oTox.self_get_connection_status() != TOX_CONNECTION['NONE']: elif oTox.self_get_connection_status() != enums.TOX_CONNECTION['NONE']:
LOG.info(f'bootstrap_tcp to {host} connected' ) LOG.info(f'bootstrap_tcp to {host} connected' )
break break
else: else:
@ -863,7 +910,7 @@ def iNmapInfo(sProt, sHost, sPort, key=None, environ=None, cmd='nmap'):
return 0 return 0
# ts.bootstrap_iNmapInfo(lElts, self._args, sProt) # bootstrap_iNmapInfo(lElts, self._args, sProt)
def bootstrap_iNmapInfo(lElts, oArgs, protocol="tcp4", bIS_LOCAL=False, iNODES=iNODES, cmd='nmap'): def bootstrap_iNmapInfo(lElts, oArgs, protocol="tcp4", bIS_LOCAL=False, iNODES=iNODES, cmd='nmap'):
if not bIS_LOCAL and not bAreWeConnected(): if not bIS_LOCAL and not bAreWeConnected():
LOG.warn(f"bootstrap_iNmapInfo not local and NOT CONNECTED") LOG.warn(f"bootstrap_iNmapInfo not local and NOT CONNECTED")

View File

@ -1588,7 +1588,7 @@ class ToxSuite(unittest.TestCase):
if hasattr(self, 'baid') and self.baid >= 0: if hasattr(self, 'baid') and self.baid >= 0:
self.bob.friend_delete(self.baid) self.bob.friend_delete(self.baid)
#? @unittest.skip('crashes') @unittest.skip('crashes')
def test_kill_remake(self): def test_kill_remake(self):
""" """
t:friend_get_kill_remake t:friend_get_kill_remake
@ -2078,7 +2078,7 @@ class ToxSuite(unittest.TestCase):
LOG_INFO(f"test_file_transfer:: self.wait_objs_attr completed") LOG_INFO(f"test_file_transfer:: self.wait_objs_attr completed")
@unittest.skip('crashes') #? @unittest.skip('crashes')
def test_tox_savedata(self): # works sorta def test_tox_savedata(self): # works sorta
""" """
t:get_savedata_size t:get_savedata_size
@ -2098,6 +2098,7 @@ class ToxSuite(unittest.TestCase):
LOG.info("test_tox_savedata alice.kill") LOG.info("test_tox_savedata alice.kill")
# crashes # crashes
self.alice.kill() self.alice.kill()
del self.alice
except: except:
pass pass