updates
Some checks failed
Gitea Actions Demo / Explore-Gitea-Actions (push) Has been cancelled

This commit is contained in:
emdee@macaw.me 2023-12-16 23:23:00 +00:00
parent 6deea64979
commit 4987618fba
9 changed files with 201 additions and 158 deletions

1
.gitignore vendored
View File

@ -170,3 +170,4 @@ cython_debug/
*.dst
*~
.rsync.sh

View File

@ -4,16 +4,19 @@ prepare::
bash .pylint.sh
check::
PYTHONPATH=${PWD} pyanalyze \
PYTHONPATH=$${PWD} pyanalyze \
tox_wrapper/tox.py tox_wrapper/tests/tests_wrapper.py \
> .pyanal.out 2>&1
install::
$(PYTHON) setup.py install
test;:
help::
$(PYTHON) tox_wrapper/tests/tests_wrapper.py --help
test::
$(PYTHON) tox_wrapper/tests/tests_wrapper.py
clean::
rm -f *~ */*~ */*/*~
rm -f .[a-z]* *~ */*~ */*/*~
rm -rf *.egg-info

View File

@ -280,7 +280,7 @@ def bAreWeConnected() -> bool:
i += 1
return i > 0
def sMapaddressResolv(target:str, iPort:int = 9051, log_leve:int = 10) -> str:
def sMapaddressResolv(target:str, iPort:int = 9051, log_level:int = 10) -> str:
if not stem:
LOG.warn('please install the stem Python package')
return ''

View File

@ -37,11 +37,11 @@ try:
except ImportError as e:
nmap = False
import wrapper
import wrapper.toxcore_enums_and_consts as enums
import tox_wrapper
import tox_wrapper.toxcore_enums_and_consts as enums
from wrapper_tests.support_http import bAreWeConnected
from wrapper_tests.support_onions import (is_valid_fingerprint,
from tox_wrapper.tests.support_http import bAreWeConnected
from tox_wrapper.tests.support_onions import (is_valid_fingerprint,
lIntroductionPoints,
oGetStemController,
sMapaddressResolv, sTorResolve)
@ -55,11 +55,25 @@ except ImportError:
global LOG
LOG = logging.getLogger()
def LOG_ERROR(l:str) -> None: print('ERRORc: '+l)
def LOG_WARN(l:str) -> None: print('WARNc: ' +l)
def LOG_INFO(l:str) -> None: print('INFOc: ' +l)
def LOG_DEBUG(l:str) -> None: print('DEBUGc: '+l)
# callbacks can be called in any thread so were being careful
def LOG_ERROR(l): print('EROR< '+l)
def LOG_WARN(l): print('WARN< '+l)
def LOG_INFO(l):
bIsVerbose = not hasattr(__builtins__, 'app') or app.oArgs.loglevel <= 20-1
if bIsVerbose: print('INFO< '+l)
def LOG_DEBUG(l):
bIsVerbose = not hasattr(__builtins__, 'app') or app.oArgs.loglevel <= 10-1
if bIsVerbose: print('DBUG< '+l)
def LOG_TRACE(l):
bIsVerbose = not hasattr(__builtins__, 'app') or app.oArgs.loglevel < 10-1
pass # print('TRACE+ '+l)
def LOG_ERROR(l:str) -> None: print('ERROR: '+l)
def LOG_WARN(l:str) -> None: print('WARN: ' +l)
def LOG_INFO(l:str) -> None: print('INFO: ' +l)
def LOG_DEBUG(l:str) -> None: print('DEBUG: '+l)
def LOG_TRACE(l:str) -> None: pass # print('TRACE+ '+l)
def LOG_LOG(l:str) -> None: print('CORE: ' +l)
try:
from trepan.api import debug
@ -99,7 +113,6 @@ lBOOLEANS = [
'dht_announcements_enabled',
'save_history',
'download_nodes_list'
'core_logging',
]
sDIR = os.environ.get('TMPDIR', '/tmp')
@ -198,12 +211,9 @@ def tox_log_cb(level, filename, line, func, message, *args) -> None:
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 line in [944, 660]: return
# root WARNING 3network.c#944:b'send_packet'attempted to send message with network family 10 (probably IPv6) on IPv4 socket
if message.find('07 = GET_NODES') > 0: return
if filename == 'TCP_common.c': return
i = message.find(' | ')
@ -215,8 +225,6 @@ def tox_log_cb(level, filename, line, func, message, *args) -> None:
# 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
@ -247,86 +255,20 @@ def tox_log_cb(level, filename, line, func, message, *args) -> None:
def vAddLoggerCallback(tox_options, callback=None) -> None:
if callback is None:
wrapper.tox.Tox.libtoxcore.tox_options_set_log_callback(
tox_wrapper.tox.Tox.libtoxcore.tox_options_set_log_callback(
tox_options._options_pointer,
POINTER(None)())
tox_options.self_logger_cb = None
LOG.debug("toxcore logging disabled")
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_wrapper.tox.Tox.libtoxcore.tox_options_set_log_callback(
tox_options._options_pointer,
tox_options.self_logger_cb)
LOG.debug("toxcore logging enabled")
def get_video_indexes() -> list:
# 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 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):
# 'Mode: 0=chat 1=chat+audio 2=chat+audio+video default: 0'
@ -394,6 +336,95 @@ def oMainArgparser(_=None, iMode=0):
# help='En/Disable saving history')
return parser
def get_video_indexes() -> list:
# 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 oToxygenToxOptions(oArgs, logger_cb=None):
data = None
tox_options = tox_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 and logger_cb:
LOG.debug("Adding logging to tox_options._options_pointer ")
vAddLoggerCallback(tox_options, logger_cb)
else:
LOG.warning("No tox_options._options_pointer " +repr(tox_options._options_pointer))
return tox_options
def toxygen_log_cb(iTox, level, file, line, func, message, *args):
"""
* @param level The severity of the log message.
* @param file 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.
"""
try:
file = str(file, 'UTF-8')
# root WARNING 3network.c#944:b'send_packet'attempted to send message with network family 10 (probably IPv6) on IPv4 socket
if file == 'network.c' and line in [944, 660]: return
func = str(func, 'UTF-8')
message = str(message, 'UTF-8')
message = f"{file}#{line}:{func} {message}"
LOG_LOG(message)
except Exception as e:
LOG_ERROR(f"toxygen_log_cb EXCEPTION {e}")
def vSetupLogging(oArgs) -> None:
global LOG
logging._defaultFormatter = logging.Formatter(datefmt='%m-%d %H:%M:%S')

View File

@ -68,11 +68,11 @@ except ImportError as e:
logging.log(logging.DEBUG, f"color_runner not available: {e}")
color_runner = None
import wrapper
import wrapper.toxcore_enums_and_consts as enums
from wrapper.tox import Tox, UINT32_MAX, ToxError
import tox_wrapper
import tox_wrapper.toxcore_enums_and_consts as enums
from tox_wrapper.tox import Tox, UINT32_MAX, ToxError
from wrapper.toxcore_enums_and_consts import (TOX_ADDRESS_SIZE, TOX_CONNECTION,
from tox_wrapper.toxcore_enums_and_consts import (TOX_ADDRESS_SIZE, TOX_CONNECTION,
TOX_FILE_CONTROL,
TOX_MESSAGE_TYPE,
TOX_SECRET_KEY_SIZE,
@ -81,7 +81,7 @@ from wrapper.toxcore_enums_and_consts import (TOX_ADDRESS_SIZE, TOX_CONNECTION,
try:
import support_testing as ts
except ImportError:
import wrapper_tests.support_testing as ts
import tox_wrapper.tests.support_testing as ts
try:
from tests.toxygen_tests import test_sound_notification
@ -232,6 +232,7 @@ def prepare(self):
else:
if self.bob.self_get_connection_status() != status:
LOG_WARN(f"bobs_on_self_connection_status DISAGREE {status}")
self.bob.dht_connected = status
def alices_on_self_connection_status(iTox, connection_state: int, *args) -> None:
#FixMe connection_num
@ -251,7 +252,7 @@ def prepare(self):
LOG_ERROR(f"alices_on_self_connection_status error={e}")
self.alice.dht_connected = status
opts = oToxygenToxOptions(oTOX_OARGS)
opts = oTestsToxOptions(oTOX_OARGS)
global bUSE_NOREQUEST
bUSE_NOREQUEST = oTOX_OARGS.norequest == 'True'
@ -262,6 +263,13 @@ def prepare(self):
alice.mycon_time = 1
alice.callback_self_connection_status(alices_on_self_connection_status)
# only bob logs trace_enabled
if True or oTOX_OARGS.trace_enabled:
LOG.info(f"toxcore trace_enabled")
ts.vAddLoggerCallback(opts, callback=ts.tox_log_cb)
else:
LOG.debug(f"toxcore trace_enabled")
bob = BobTox(opts, app=oAPP)
bob.oArgs = opts
bob.dht_connected = -1
@ -305,7 +313,7 @@ class ToxSuite(unittest.TestCase):
for elt in self.alice.self_get_friend_list():
self.alice.friend_delete(elt)
LOG.debug(f"tearDown threads={threading.active_count()}")
# LOG.debug(f"tearDown threads={threading.active_count()}")
if hasattr(self, 'bob'):
bob.callback_self_connection_status(None)
if hasattr(self.bob, 'main_loop'):
@ -999,7 +1007,7 @@ class ToxSuite(unittest.TestCase):
def test_hash(self): # works
otox = self.bob
string = 'abcdef'
name = otox.hash(string)
name = otox.hash(bytes(string, 'utf-8'))
assert name
string = b'abcdef'
name = otox.hash(string)
@ -1622,7 +1630,7 @@ class ToxSuite(unittest.TestCase):
LOG_ERROR(f"bobs_on_friend_connection_status ERROR {e}")
setattr(self.bob, sSlot, True)
opts = oToxygenToxOptions(oTOX_OARGS)
opts = oTestsToxOptions(oTOX_OARGS)
setattr(self.bob, sSlot, True)
try:
if bUSE_NOREQUEST:
@ -2007,9 +2015,12 @@ class ToxSuite(unittest.TestCase):
# AliceTox.on_file_data = on_file_data
try:
# required
assert self.wait_friend_get_connection_status(self.bob, self.baid, n=2*iN)
assert self.wait_friend_get_connection_status(self.alice, self.abid, n=2*iN)
# required?
if not self.wait_friend_get_connection_status(self.bob, self.baid, n=2*iN):
LOG_WARN(f"bobs wait_friend_get_connection_status {2*iN}")
if not self.wait_friend_get_connection_status(self.alice, self.abid, n=2*iN):
LOG_WARN(f"alices' wait_friend_get_connection_status {2*iN}")
self.alice.callback_file_recv(alice_on_file_recv)
self.alice.callback_file_recv_control(alice_on_file_recv_control)
@ -2118,7 +2129,7 @@ class ToxSuite(unittest.TestCase):
pass
oArgs = oTOX_OARGS
opts = oToxygenToxOptions(oArgs)
opts = oTestsToxOptions(oArgs)
opts.savedata_data = data
opts.savedata_length = len(data)
@ -2171,9 +2182,9 @@ def iMain(oArgs, failfast=True) -> int:
runner.run(cases)
# collect_types.dump_stats('tests_wrapper.out')
def oToxygenToxOptions(oArgs):
def oTestsToxOptions(oArgs):
data = None
tox_options = wrapper.tox.Tox.options_new()
tox_options = tox_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')
@ -2209,13 +2220,6 @@ def oToxygenToxOptions(oArgs):
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 ")
ts.vAddLoggerCallback(tox_options, ts.on_log)
else:
LOG.warning("No tox_options._options_pointer " +repr(tox_options._options_pointer))
return tox_options
def oArgparse(lArgv):
@ -2254,7 +2258,7 @@ def main(lArgs=None) -> int:
setattr(oTOX_OARGS, 'bIS_LOCAL', bIS_LOCAL)
# oTOX_OPTIONS = ToxOptions()
global oTOX_OPTIONS
oTOX_OPTIONS = oToxygenToxOptions(oArgs)
oTOX_OPTIONS = oTestsToxOptions(oArgs)
if coloredlogs:
# https://pypi.org/project/coloredlogs/
coloredlogs.install(level=oArgs.loglevel,

View File

@ -11,16 +11,18 @@ from datetime import datetime
from typing import Union, Callable
try:
from wrapper.libtox import LibToxCore
from wrapper.toxav import ToxAV
from wrapper.toxcore_enums_and_consts import *
import wrapper.toxcore_enums_and_consts as enum
from tox_wrapper.libtox import LibToxCore
from tox_wrapper.toxav import ToxAV
from tox_wrapper.toxcore_enums_and_consts import *
import tox_wrapper.toxcore_enums_and_consts as enum
except:
from libtox import LibToxCore
from toxav import ToxAV
from toxcore_enums_and_consts import *
import toxcore_enums_and_consts as enum
_c_uint32 = POINTER(c_uint32)
# callbacks can be called in any thread so were being careful
# tox.py can be called by callbacks
def LOG_ERROR(a) -> None:
@ -90,11 +92,14 @@ class GroupChatSelfPeerInfo(Structure):
]
def string_to_bin_charp(tox_id):
if tox_id is None: return None
assert type(tox_id) == str, f"{type(tox_id)} != str"
return c_char_p(bytes.fromhex(tox_id)) if tox_id is not None else None
return c_char_p(bytes.fromhex(tox_id))
def bin_to_string(raw_id, length) -> str:
assert isinstance(raw_id, bytes) or isinstance(raw_id, Array), \
f"{type(raw_id)} != bytes"
res = ''.join('{:02x}'.format(ord(raw_id[i])) for i in range(length))
return res.upper()
@ -428,6 +433,7 @@ class Tox:
isinstance(user_data, Array), type(user_data)
try:
LOG_TRACE(f"tox_iterate")
# void pointer
Tox.libtoxcore.tox_iterate(self._tox_pointer, c_char_p(user_data))
except Exception as e:
# Fatal Python error: Segmentation fault
@ -453,7 +459,7 @@ class Tox:
address = create_string_buffer(TOX_ADDRESS_SIZE)
else:
isinstance(address, Array), type(address)
LOG_DEBUG(f"tox.self_get_address")
LOG_INFO(f"tox.self_get_address")
Tox.libtoxcore.tox_self_get_address(self._tox_pointer, address)
return bin_to_string(address, TOX_ADDRESS_SIZE)
@ -522,7 +528,7 @@ class Tox:
name = bytes(name, 'utf-8')
LOG_DEBUG(f"tox.self_set_name")
result = Tox.libtoxcore.tox_self_set_name(self._tox_pointer,
c_char_p(name),
name, # sic
c_size_t(len(name)),
byref(tox_err_set_info))
tox_err_set_info = tox_err_set_info.value
@ -582,7 +588,7 @@ class Tox:
status_message = bytes(status_message, 'utf-8')
LOG_DEBUG(f"tox.self_set_status_message")
result = Tox.libtoxcore.tox_self_set_status_message(self._tox_pointer,
c_char_p(status_message),
status_message, # sic
c_size_t(len(status_message)),
byref(tox_err_set_info))
tox_err_set_info = tox_err_set_info.value
@ -678,7 +684,7 @@ class Tox:
message = bytes(message, 'utf-8')
result = Tox.libtoxcore.tox_friend_add(self._tox_pointer,
string_to_bin_charp(address),
c_char_p(message),
message, # sic
c_size_t(len(message)),
byref(tox_err_friend_add))
tox_err_friend_add = tox_err_friend_add.value
@ -1249,7 +1255,8 @@ class Tox:
result = Tox.libtoxcore.tox_friend_send_message(self._tox_pointer,
c_uint32(friend_number),
c_int(message_type),
c_char_p(message), c_size_t(len(message)),
message, # sicNO
c_size_t(len(message)),
byref(tox_err_friend_send_message))
tox_err_friend_send_message = tox_err_friend_send_message.value
if tox_err_friend_send_message == TOX_ERR_FRIEND_SEND_MESSAGE['OK']:
@ -1348,7 +1355,7 @@ class Tox:
# File transmission: common between sending and receiving
@staticmethod
def hash(data: bytes, hash=None) -> str:
def hash(data: Union[str,bytes], hash=None) -> str:
"""Generates a cryptographic hash of the given data.
This function may be used by clients for any purpose, but is
@ -1357,7 +1364,7 @@ class Tox:
If hash is NULL or data is NULL while length is not 0 the function returns false, otherwise it returns true.
This function is a wrapper to internal message-digest functions.
This function is a tox_wrapper to internal message-digest functions.
:param hash: A valid memory location the hash data. It must be at least TOX_HASH_LENGTH bytes in size.
:param data: Data to be hashed or NULL.
@ -1368,7 +1375,8 @@ class Tox:
hash = create_string_buffer(TOX_HASH_LENGTH)
else:
assert isinstance(hash, Array), f"{type(hash)}"
assert type(data) == bytes, f"{type(data)} != bytes"
if type(data) == str:
data = bytes(data, 'utf-8') # f"{type(data)} != bytes"
Tox.libtoxcore.tox_hash(hash, c_char_p(data), c_size_t(len(data)))
return bin_to_string(hash, TOX_HASH_LENGTH)
@ -1503,7 +1511,7 @@ class Tox:
# File transmission: sending
def file_send(self, friend_number: int, kind: int, file_size: int, file_id, filename: str) -> int:
def file_send(self, friend_number: int, kind: int, file_size: int, file_id: int, filename: Union[bytes,str]) -> int:
"""Send a file transmission request.
Maximum filename length is TOX_MAX_FILENAME_LENGTH bytes. The
@ -1555,16 +1563,14 @@ class Tox:
"""
LOG_DEBUG(f"tox.file_send")
tox_err_file_send = c_int()
if type(file_id) == bytes:
public_key = str(file_id, 'utf-8')
if type(filename) == bytes:
filename = str(filename, 'utf-8')
if type(filename) == str:
filename = bytes(filename, 'utf-8')
result = self.libtoxcore.tox_file_send(self._tox_pointer,
c_uint32(friend_number),
c_uint32(kind),
c_uint64(file_size),
string_to_bin_charp(file_id),
c_char_p(filename),
filename, # sic
c_size_t(len(filename)),
byref(tox_err_file_send))
err_file = tox_err_file_send.value
@ -1584,7 +1590,7 @@ class Tox:
'friend per direction (sending and receiving).')
raise ToxError('The function did not return OK')
def file_send_chunk(self, friend_number: int, file_number: int, position, data: str) -> int:
def file_send_chunk(self, friend_number: int, file_number: int, position, data: Union[Array,bytes]) -> int:
"""
Send a chunk of file data to a friend.
@ -1603,12 +1609,12 @@ class Tox:
LOG_DEBUG(f"tox.file_send_chunk")
tox_err_file_send_chunk = c_int()
isinstance(data, Array), type(data)
isinstance(data, Array) or isinstance(data, bytes), "file_send_chunk type(data)"
result = self.libtoxcore.tox_file_send_chunk(self._tox_pointer,
c_uint32(friend_number), c_uint32(file_number),
c_uint64(position),
c_char_p(data),
data, # sic
c_size_t(len(data)),
byref(tox_err_file_send_chunk))
tox_err_file_send_chunk = tox_err_file_send_chunk.value
@ -1754,7 +1760,7 @@ class Tox:
# Low-level custom packet sending and receiving
def friend_send_lossy_packet(self, friend_number: int, data: bytes) -> bool:
def friend_send_lossy_packet(self, friend_number: int, data: Union[Array,bytes]) -> bool:
"""
Send a custom lossy packet to a friend.
The first byte of data must be in the range 200-254. Maximum length of a
@ -1771,11 +1777,11 @@ class Tox:
:return: True on success.
"""
LOG_DEBUG(f"friend_send_lossy_packet")
isinstance(data, Array), type(data)
isinstance(data, bytes) or isinstance(data, Array), f"{type(data)}"
tox_err_friend_custom_packet = c_int()
result = self.libtoxcore.tox_friend_send_lossy_packet(self._tox_pointer,
c_uint32(friend_number),
c_char_p(data),
data, # sic
c_size_t(len(data)),
byref(tox_err_friend_custom_packet))
tox_err_friend_custom_packet = tox_err_friend_custom_packet.value
@ -1798,7 +1804,7 @@ class Tox:
raise ToxError('Packet queue is full.')
raise ToxError('The function did not return OK')
def friend_send_lossless_packet(self, friend_number: int, data: bytes) -> int:
def friend_send_lossless_packet(self, friend_number: int, data: Union[Array,bytes]) -> int:
"""
Send a custom lossless packet to a friend.
The first byte of data must be in the range 160-191. Maximum length of a
@ -1813,10 +1819,10 @@ class Tox:
"""
LOG_DEBUG(f"friend_send_lossless_packet")
tox_err_friend_custom_packet = c_int()
isinstance(data, Array), type(data)
isinstance(data, bytes) or isinstance(data, Array), f"{type(data)}"
result = self.libtoxcore.tox_friend_send_lossless_packet(self._tox_pointer,
c_uint32(friend_number),
c_char_p(data),
data, # sic
c_size_t(len(data)),
byref(tox_err_friend_custom_packet))
tox_err_friend_custom_packet = tox_err_friend_custom_packet.value
@ -1966,7 +1972,7 @@ class Tox:
result = Tox.libtoxcore.tox_group_new(self._tox_pointer,
privacy_state,
c_char_p(group_name),
group_name, # sic
c_size_t(len(group_name)),
nick,
c_size_t(len(nick)),
@ -2008,10 +2014,10 @@ class Tox:
if password and type(password) == str:
nick = bytes(password, 'utf-8')
#?no cpassword = c_char_p(password) # it's const uint8_t *password
cpassword = password
cpassword = password # sic
result = Tox.libtoxcore.tox_group_join(self._tox_pointer,
string_to_bin_charp(chat_id),
c_char_p(nick),
nick, # sic
c_size_t(len(nick)),
cpassword,
c_size_t(len(password)) if password else 0,

View File

@ -6,8 +6,8 @@ from ctypes import (CFUNCTYPE, POINTER, ArgumentError, byref, c_bool, c_char_p,
from typing import Union, Callable
try:
from wrapper.libtox import LibToxAV
import wrapper.toxav_enums as enum
from tox_wrapper.libtox import LibToxAV
import tox_wrapper.toxav_enums as enum
except:
from libtox import LibToxAV
import toxav_enums as enum

View File

@ -1,18 +1,16 @@
# -*- mode: python; indent-tabs-mode: nil; py-indent-offset: 4; coding: utf-8 -*-
try:
from wrapper import libtox
#? from wrapper.toxencryptsave_enums_and_consts import *
import wrapper.toxencryptsave_enums_and_consts as enum
from tox_wrapper import libtox
import tox_wrapper.toxencryptsave_enums_and_consts as enum
except:
import libtox
#? from toxencryptsave_enums_and_consts import *
import toxencryptsave_enums_and_consts as enum
from typing import Union, Callable
from ctypes import (ArgumentError, byref, c_bool, c_char_p, c_int, c_size_t,
create_string_buffer, Array)
def ToxError(ArgumentError): pass
class ToxEncryptSave:

View File

@ -29,14 +29,14 @@ def LOG_info(a): print('INFO_ '+a)
def LOG_debug(a): print('DBUG_ '+a)
def LOG_trace(a): pass # print('TRAC_ '+a)
from wrapper import tox
import wrapper.toxcore_enums_and_consts as enums
from wrapper.tox import Tox, UINT32_MAX
from wrapper.toxcore_enums_and_consts import TOX_CONNECTION, TOX_USER_STATUS, \
from tox_wrapper import tox
import tox_wrapper.toxcore_enums_and_consts as enums
from tox_wrapper.tox import Tox, UINT32_MAX
from tox_wrapper.toxcore_enums_and_consts import TOX_CONNECTION, TOX_USER_STATUS, \
TOX_MESSAGE_TYPE, TOX_PUBLIC_KEY_SIZE, TOX_FILE_CONTROL, TOX_FILE_KIND
import wrapper_tests.support_testing as ts
from wrapper_tests.support_testing import oMainArgparser
import tox_wrapper.tests.support_testing as ts
from tox_wrapper.tests.support_testing import oMainArgparser
def sleep(fSec):
if 'QtCore' in globals():