emdee/toxygen_wrapper
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
emdee@macaw.me 2023-12-14 20:46:56 +00:00
parent b934928fe3
commit 012c7ea56e
8 changed files with 388 additions and 286 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.pyanal.sh Normal file
View File

@ -0,0 +1,8 @@
#!/bin/sh
ROLE=logging
PYTHONPATH=$PWD/wrapper /var/local/bin/python3.bash `which pyanalyze` \
wrapper wrapper_tests/tests_wrapper.py \
> .pyanal.out 2>&1

517
wrapper/tox.py
View File

File diff suppressed because it is too large Load Diff

52
wrapper/toxav.py
View File

@ -11,6 +11,7 @@ try:
except:
from libtox import LibToxAV
import toxav_enums as enum
class ToxError(RuntimeError): pass
def LOG_ERROR(a: str) -> None: print('EROR> '+a)
def LOG_WARN(a: str) -> None: print('WARN> '+a)
@ -91,7 +92,7 @@ class ToxAV:
# Call setup
def call(self, friend_number: int, audio_bit_rate: int, video_bit_rate: int) -> None:
def call(self, friend_number: int, audio_bit_rate: int, video_bit_rate: int) -> bool:
"""
Call a friend. This will start ringing the friend.
@ -111,22 +112,22 @@ class ToxAV:
toxav_err_call = toxav_err_call.value
if toxav_err_call == enum.TOXAV_ERR_CALL['OK']:
return bool(result)
elif toxav_err_call == enum.TOXAV_ERR_CALL['MALLOC']:
if toxav_err_call == enum.TOXAV_ERR_CALL['MALLOC']:
raise MemoryError('A resource allocation error occurred while trying to create the structures required for '
'the call.')
elif toxav_err_call == enum.TOXAV_ERR_CALL['SYNC']:
if toxav_err_call == enum.TOXAV_ERR_CALL['SYNC']:
raise RuntimeError('Synchronization error occurred.')
elif toxav_err_call == enum.TOXAV_ERR_CALL['FRIEND_NOT_FOUND']:
if toxav_err_call == enum.TOXAV_ERR_CALL['FRIEND_NOT_FOUND']:
raise ArgumentError('The friend number did not designate a valid friend.')
elif toxav_err_call == enum.TOXAV_ERR_CALL['FRIEND_NOT_CONNECTED']:
if toxav_err_call == enum.TOXAV_ERR_CALL['FRIEND_NOT_CONNECTED']:
raise ArgumentError('The friend was valid, but not currently connected.')
elif toxav_err_call == enum.TOXAV_ERR_CALL['FRIEND_ALREADY_IN_CALL']:
if toxav_err_call == enum.TOXAV_ERR_CALL['FRIEND_ALREADY_IN_CALL']:
raise ArgumentError('Attempted to call a friend while already in an audio or video call with them.')
elif toxav_err_call == enum.TOXAV_ERR_CALL['INVALID_BIT_RATE']:
if toxav_err_call == enum.TOXAV_ERR_CALL['INVALID_BIT_RATE']:
raise ArgumentError('Audio or video bit rate is invalid.')
raise ArgumentError('The function did not return OK')
def callback_call(self, callback: Callable, user_data) -> None:
def callback_call(self, callback: Union[Callable,None], user_data) -> None:
"""
Set the callback for the `call` event. Pass None to unset.
@ -148,7 +149,7 @@ class ToxAV:
self.call_cb = c_callback(callback)
self.libtoxav.toxav_callback_call(self._toxav_pointer, self.call_cb, user_data)
def answer(self, friend_number: int, audio_bit_rate: int, video_bit_rate: int) -> None:
def answer(self, friend_number: int, audio_bit_rate: int, video_bit_rate: int) -> bool:
"""
Accept an incoming call.
@ -186,7 +187,7 @@ class ToxAV:
# Call state graph
def callback_call_state(self, callback: Callable, user_data) -> None:
def callback_call_state(self, callback: Union[Callable,None], user_data) -> None:
"""
Set the callback for the `call_state` event. Pass None to unset.
@ -211,7 +212,7 @@ class ToxAV:
# Call control
def call_control(self, friend_number: int, control: int) -> None:
def call_control(self, friend_number: int, control: int) -> bool:
"""
Sends a call control command to a friend.
@ -221,8 +222,10 @@ class ToxAV:
"""
toxav_err_call_control = c_int()
LOG_DEBUG(f"call_control")
result = self.libtoxav.toxav_call_control(self._toxav_pointer, c_uint32(friend_number), c_int(control),
byref(toxav_err_call_control))
result = self.libtoxav.toxav_call_control(self._toxav_pointer,
c_uint32(friend_number),
c_int(control),
byref(toxav_err_call_control))
toxav_err_call_control = toxav_err_call_control.value
if toxav_err_call_control == enum.TOXAV_ERR_CALL_CONTROL['OK']:
return bool(result)
@ -242,7 +245,7 @@ class ToxAV:
# A/V sending
def audio_send_frame(self, friend_number: int, pcm, sample_count: int, channels: int, sampling_rate: int) -> None:
def audio_send_frame(self, friend_number: int, pcm, sample_count: int, channels: int, sampling_rate: int) -> bool:
"""
Send an audio frame to a friend.
@ -288,7 +291,7 @@ class ToxAV:
RuntimeError('Failed to push frame through rtp interface.')
raise ToxError('The function did not return OK.')
def video_send_frame(self, friend_number: int, width: int, height: int, y, u, v) -> None:
def video_send_frame(self, friend_number: int, width: int, height: int, y, u, v) -> bool:
"""
Send a video frame to a friend.
@ -316,26 +319,27 @@ class ToxAV:
toxav_err_send_frame = toxav_err_send_frame.value
if toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['OK']:
return bool(result)
elif toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['NULL']:
if toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['NULL']:
raise ArgumentError('One of Y, U, or V was NULL.')
elif toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['FRIEND_NOT_FOUND']:
if toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['FRIEND_NOT_FOUND']:
raise ArgumentError('The friend_number passed did not designate a valid friend.')
elif toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['FRIEND_NOT_IN_CALL']:
if toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['FRIEND_NOT_IN_CALL']:
raise RuntimeError('This client is currently not in a call with the friend.')
elif toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['SYNC']:
if toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['SYNC']:
raise RuntimeError('Synchronization error occurred.')
elif toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['INVALID']:
if toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['INVALID']:
raise ArgumentError('One of the frame parameters was invalid. E.g. the resolution may be too small or too '
'large, or the audio sampling rate may be unsupported.')
elif toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['PAYLOAD_TYPE_DISABLED']:
if toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['PAYLOAD_TYPE_DISABLED']:
raise RuntimeError('Either friend turned off audio or video receiving or we turned off sending for the said'
'payload.')
elif toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['RTP_FAILED']:
if toxav_err_send_frame == enum.TOXAV_ERR_SEND_FRAME['RTP_FAILED']:
RuntimeError('Failed to push frame through rtp interface.')
raise ToxError('The function did not return OK.')
# A/V receiving
def callback_audio_receive_frame(self, callback: Callable, user_data) -> None:
def callback_audio_receive_frame(self, callback: Union[Callable,None], user_data) -> None:
"""
Set the callback for the `audio_receive_frame` event. Pass None to unset.
@ -364,7 +368,7 @@ class ToxAV:
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)
def callback_video_receive_frame(self, callback: Callable, user_data) -> None:
def callback_video_receive_frame(self, callback: Union[Callable,None], user_data) -> None:
"""
Set the callback for the `video_receive_frame` event. Pass None to unset.

19
wrapper/toxencryptsave.py
View File

@ -27,7 +27,7 @@ class ToxEncryptSave:
result = func(c_char_p(bytes(data)))
return bool(result)
def pass_encrypt(self, data: bytes, password: str) -> bytes:
def pass_encrypt(self, data: bytes, password: Union[str,bytes]) -> bytes:
"""
Encrypts the given data with the given password.
@ -35,6 +35,7 @@ class ToxEncryptSave:
"""
out = create_string_buffer(len(data) + enum.TOX_PASS_ENCRYPTION_EXTRA_LENGTH)
tox_err_encryption = c_int()
assert password
if type(password) != bytes:
password = bytes(password, 'utf-8')
self.libtoxencryptsave.tox_pass_encrypt(c_char_p(data),
@ -45,7 +46,7 @@ class ToxEncryptSave:
byref(tox_err_encryption))
tox_err_encryption = tox_err_encryption.value
if tox_err_encryption == enum.TOX_ERR_ENCRYPTION['OK']:
return out[:]
return bytes(out[:])
if tox_err_encryption == enum.TOX_ERR_ENCRYPTION['NULL']:
raise ArgumentError('Some input data, or maybe the output pointer, was null.')
if tox_err_encryption == enum.TOX_ERR_ENCRYPTION['KEY_DERIVATION_FAILED']:
@ -55,7 +56,7 @@ class ToxEncryptSave:
raise RuntimeError('The encryption itself failed.')
raise ToxError('The function did not return OK.')
def pass_decrypt(self, data: bytes, password: str) -> bytes:
def pass_decrypt(self, data: bytes, password: Union[str,bytes]) -> bytes:
"""
Decrypts the given data with the given password.
@ -63,6 +64,7 @@ class ToxEncryptSave:
"""
out = create_string_buffer(len(data) - enum.TOX_PASS_ENCRYPTION_EXTRA_LENGTH)
tox_err_decryption = c_int()
assert password
if type(password) != bytes:
password = bytes(password, 'utf-8')
self.libtoxencryptsave.tox_pass_decrypt(c_char_p(bytes(data)),
@ -74,16 +76,17 @@ class ToxEncryptSave:
tox_err_decryption = tox_err_decryption.value
if tox_err_decryption == enum.TOX_ERR_DECRYPTION['OK']:
return bytes(out[:])
elif tox_err_decryption == enum.TOX_ERR_DECRYPTION['NULL']:
if tox_err_decryption == enum.TOX_ERR_DECRYPTION['NULL']:
raise ArgumentError('Some input data, or maybe the output pointer, was null.')
elif tox_err_decryption == enum.TOX_ERR_DECRYPTION['INVALID_LENGTH']:
if tox_err_decryption == enum.TOX_ERR_DECRYPTION['INVALID_LENGTH']:
raise ArgumentError('The input data was shorter than TOX_PASS_ENCRYPTION_EXTRA_LENGTH bytes')
elif tox_err_decryption == enum.TOX_ERR_DECRYPTION['BAD_FORMAT']:
if tox_err_decryption == enum.TOX_ERR_DECRYPTION['BAD_FORMAT']:
raise ArgumentError('The input data is missing the magic number (i.e. wasn\'t created by this module, or is'
' corrupted)')
elif tox_err_decryption == enum.TOX_ERR_DECRYPTION['KEY_DERIVATION_FAILED']:
if tox_err_decryption == enum.TOX_ERR_DECRYPTION['KEY_DERIVATION_FAILED']:
raise RuntimeError('The crypto lib was unable to derive a key from the given passphrase, which is usually a'
' lack of memory issue. The functions accepting keys do not produce this error.')
elif tox_err_decryption == enum.TOX_ERR_DECRYPTION['FAILED']:
if tox_err_decryption == enum.TOX_ERR_DECRYPTION['FAILED']:
raise RuntimeError('The encrypted byte array could not be decrypted. Either the data was corrupt or the '
'password/key was incorrect.')
raise ToxError('The function did not return OK.')

6
wrapper/toxygen_echo.py
View File

@ -29,7 +29,7 @@ def LOG_info(a): print('INFO_ '+a)
def LOG_debug(a): print('DBUG_ '+a)
def LOG_trace(a): pass # print('TRAC_ '+a)
import wrapper
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, \
@ -69,7 +69,7 @@ iDHT_TRY = 0
if not bHAVE_AV:
class AV(): pass
else:
class AV(wrapper.tox.ToxAV):
class AV(tox.ToxAV):
def __init__(self, core):
super(AV, self).__init__(core)
self.core = self.get_tox()
@ -147,7 +147,7 @@ class EchoBot():
message_data_size,
*largs) -> None:
key = ''.join(chr(x) for x in public_key[:TOX_PUBLIC_KEY_SIZE])
sPk = wrapper.tox.bin_to_string(key, TOX_PUBLIC_KEY_SIZE)
sPk = 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)

1
wrapper_tests/support_http.py
View File

@ -161,4 +161,3 @@ def download_url(url:str, settings:str = None) -> None:
return ''
return ''

8
wrapper_tests/support_testing.py
View File

@ -861,13 +861,11 @@ def bootstrap_tcp(lelts:list, lToxes:list, oArgs=None) -> None:
if not oRet:
LOG.warn(f'bootstrap_tcp failed to {host} : {oRet}')
elif hasattr(oTox, 'mycon_time') and oTox.mycon_time == 1:
LOG.debug(f'bootstrap_tcp to {host} not yet connected last=1')
LOG.debug(f'bootstrap_tcp to {host} not yet connected')
elif hasattr(oTox, 'mycon_status') and oTox.mycon_status is False:
LOG.debug(f'bootstrap_tcp to {host} not True' \
+f" last={int(oTox.mycon_time)}" )
LOG.debug(f'bootstrap_tcp to {host} not True')
elif oTox.self_get_connection_status() != enums.TOX_CONNECTION['NONE']:
LOG.info(f'bootstrap_tcp to {host} connected' \
+f" last={int(oTox.mycon_time)}" )
LOG.info(f'bootstrap_tcp to {host} connected')
break
else:
# LOG.debug(f'bootstrap_tcp to {host} but not connected'

63
wrapper_tests/tests_wrapper.py
View File

@ -540,7 +540,7 @@ class ToxSuite(unittest.TestCase):
return all([getattr(obj, attr) is not None for obj in objs])
def wait_otox_attrs(self, obj: list, attrs: list[str]) -> bool:
def wait_otox_attrs(self, obj, attrs: list[str]) -> bool:
assert all(attrs), f"wait_otox_attrs {attrs}"
i = 0
while i <= THRESHOLD:
@ -846,8 +846,8 @@ class ToxSuite(unittest.TestCase):
sP = otox.group_get_password(iGrp)
assert otox.group_get_privacy_state(iGrp) == privacy_state
assert otox.group_get_number_groups() > 0
LOG.info(f"group pK={sPk} iGrp={iGrp} n={otox.group_get_number_groups()}")
assert otox.group_get_number_groups() > 0, "numg={otox.group_get_number_groups()}"
LOG.info(f"group pK={sPk} iGrp={iGrp} numg={otox.group_get_number_groups()}")
return iGrp
def otox_verify_group(self, otox, iGrp) -> int:
@ -996,6 +996,16 @@ class ToxSuite(unittest.TestCase):
self.assertEqual(cm.output, ['INFO:foo:first message',
'ERROR:foo.bar:second message'])
def test_hash(self): # works
otox = self.bob
string = 'abcdef'
name = otox.hash(string)
assert name
string = b'abcdef'
name = otox.hash(string)
assert name
LOG.info(f"test_hash: {string} -> {name} ")
def test_tests_start(self) -> None: # works
"""
t:hash
@ -1014,6 +1024,7 @@ class ToxSuite(unittest.TestCase):
len(self.alice._address)
assert self.bob.self_get_address() == self.bob._address
assert self.alice.self_get_address() == self.alice._address
def test_bootstrap_local_netstat(self) -> None: # works
"""
@ -1471,11 +1482,11 @@ class ToxSuite(unittest.TestCase):
try:
if bUSE_NOREQUEST:
assert self.bob_add_alice_as_friend_norequest()
# assert self.alice_add_bob_as_friend_norequest()
assert self.alice_add_bob_as_friend_norequest()
else:
# no not connected error
assert self.bob_add_alice_as_friend()
# assert self.alice_add_bob_as_friend_norequest()
assert self.alice_add_bob_as_friend_norequest()
self.bob.callback_friend_status_message(bob_on_friend_status_message)
self.warn_if_no_cb(self.bob, sSlot)
@ -1628,7 +1639,7 @@ class ToxSuite(unittest.TestCase):
LOG.info("test_kill_remake maked alice")
if not self.wait_otox_attrs(self.bob, [sSlot]):
LOG_WARN(f' NO {sSlot}')
LOG_WARN(f'test_kill_remake NO {sSlot}')
except AssertionError as e:
LOG.error(f"test_kill_remake Failed test {e}")
raise
@ -1778,15 +1789,15 @@ class ToxSuite(unittest.TestCase):
sSlot = 'friend_read_action'
setattr(self.bob, sSlot, None)
def UNUSEDtheir_on_friend_action(iTox, fid:int, msg_type, action, *largs):
LOG_DEBUG(f"their_on_friend_action {fid} {msg_type} {action}")
LOG_DEBUG(f"their_on_friend_action {fid} {msg_type} {sSlot} {action}")
try:
assert msg_type == TOX_MESSAGE_TYPE['ACTION']
assert action == ACTION
except Exception as e:
LOG_ERROR(f"their_on_friend_action EXCEPTION {e}")
LOG_ERROR(f"their_on_friend_action EXCEPTION {sSlot} {e}")
else:
LOG_INFO(f"their_on_friend_action {action}")
setattr(self.bob, 'friend_read_action', True)
LOG_INFO(f"their_on_friend_action {sSlot} {action}")
setattr(self.bob, sSlot, True)
sSlot = 'friend_read_receipt'
setattr(self.alice, sSlot, None)
@ -1798,10 +1809,10 @@ class ToxSuite(unittest.TestCase):
assert fid == bob.baid or fid == alice.abid
assert msg_id >= 0
except Exception as e:
LOG_ERROR(f"their_on_read_reciept {e}")
LOG_ERROR(f"their_on_read_reciept {sSlot} {e}")
else:
LOG_INFO(f"their_on_read_reciept {fid}")
setattr(self.alice, 'friend_read_receipt', True)
LOG_INFO(f"their_on_read_reciept {sSlot} fid={fid}")
setattr(self.alice, sSlot, True)
try:
if bUSE_NOREQUEST:
@ -1885,9 +1896,6 @@ class ToxSuite(unittest.TestCase):
self.alice.self_set_typing(self.abid, False)
if not self.wait_otox_attrs(self.bob, [sSlot]):
LOG_WARN(f"bobs_on_friend_typing NO {sSlot}")
if not hasattr(self.bob, sSlot+'_cb') or \
not getattr(self.bob, sSlot+'_cb'):
LOG.warning(f"self.bob.{sSlot}_cb NOT EXIST")
except AssertionError as e:
LOG.error(f"Failed test {e}")
raise
@ -1929,6 +1937,8 @@ class ToxSuite(unittest.TestCase):
oFd.write(FILE)
FILE_SIZE = len(FILE)
OFFSET = 567
# was FILE_ID = FILE_NAME
FILE_ID = 32*'1' #
m = hashlib.md5()
m.update(FILE[OFFSET:])
@ -1967,7 +1977,7 @@ class ToxSuite(unittest.TestCase):
LOG_INFO(f"ALICE_ON_file_recv " + str(fid))
self.alice.completed = False
def alice_on_file_recv_chunk(iTox, fid:int, file_number:int, position:int, iNumBytes, *largs) -> None:
def alice_on_file_recv_chunk(iTox, fid:int, file_number:int, position:int, iNumBytes, *largs) -> bool:
LOG_DEBUG(f"ALICE_ON_file_recv_chunk {fid} {file_number}")
# FixMe - use file_number and iNumBytes to get data?
data = ''
@ -1980,7 +1990,7 @@ class ToxSuite(unittest.TestCase):
assert m.hexdigest() == FILE_DIGEST
self.alice.completed = True
self.alice.file_control(fid, file_number, TOX_FILE_CONTROL['CANCEL'])
return
return True
CONTEXT['FILE'] += data
CONTEXT['RECEIVED'] += len(data)
@ -1989,12 +1999,18 @@ class ToxSuite(unittest.TestCase):
# fid, file_number, 1) == FILE_SIZE - CONTEXT['RECEIVED']
except Exception as e:
LOG_ERROR(f"ALICE_ON_file_recv_chunk {e}")
return False
return True
# AliceTox.on_file_send_request = on_file_send_request
# AliceTox.on_file_control = on_file_control
# 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)
self.alice.callback_file_recv(alice_on_file_recv)
self.alice.callback_file_recv_control(alice_on_file_recv_control)
self.alice.callback_file_recv_chunk(alice_on_file_recv_chunk)
@ -2019,13 +2035,6 @@ class ToxSuite(unittest.TestCase):
self.bob.callback_file_recv_control(bob_on_file_recv_control2)
self.bob.callback_file_chunk_request(bob_on_file_chunk_request)
# was FILE_ID = FILE_NAME
FILE_ID = 32*'1' #
# required
assert self.wait_friend_get_connection_status(self.bob, self.baid, n=iN)
assert self.wait_friend_get_connection_status(self.alice, self.abid, n=iN)
i = 0
iKind = 0
while i < 2:
@ -2085,8 +2094,8 @@ class ToxSuite(unittest.TestCase):
LOG_INFO(f"test_file_transfer:: self.wait_objs_attr completed")
#? @unittest.skip('crashes')
def test_tox_savedata(self) -> None: # works sorta
@unittest.skip('crashes')
def test_tox_savedata(self) -> None: #
"""
t:get_savedata_size
t:get_savedata