Files
facefusion/tests/test_rtc.py
T
Henry Ruhs 76c413a2c1 Refactor/ffmpeg less stream (#1092)
* remove ffmpeg from stream to use opus and vpx, add bunch of todos

* fix testing

* improve download checkout

* fix datachannel download, fix super dirty test clients - setup logic does not belong there

* fix testing
2026-05-11 16:36:23 +02:00

93 lines
2.9 KiB
Python

from typing import List
import pytest
from facefusion import rtc
from facefusion.libraries import datachannel as datachannel_module
from facefusion.types import RtcPeer
@pytest.fixture(scope = 'module')
def before_all() -> None:
datachannel_module.pre_check()
# TODO: add test_parse_sdp_payload_types
def test_build_media_description() -> None:
assert rtc.build_media_description('audio', 111, 'opus/48000/2', 'sendonly', 1) == b'm=audio 9 UDP/TLS/RTP/SAVPF 111\r\na=rtpmap:111 opus/48000/2\r\na=rtcp-fb:111 nack\r\na=rtcp-fb:111 nack pli\r\na=sendonly\r\na=mid:1\r\na=rtcp-mux\r\n'
assert rtc.build_media_description('video', 96, 'VP8/90000', 'recvonly', 0) == b'm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=rtpmap:96 VP8/90000\r\na=rtcp-fb:96 nack\r\na=rtcp-fb:96 nack pli\r\na=recvonly\r\na=mid:0\r\na=rtcp-mux\r\n'
# TODO: enable again
@pytest.mark.skip
def test_create_peer_connection() -> None:
peer_connection = rtc.create_peer_connection()
datachannel_library = datachannel_module.create_static_library()
assert peer_connection > 0
assert datachannel_library.rtcDeletePeerConnection(peer_connection) == 0
# TODO: enable again
@pytest.mark.skip
def test_add_audio_track() -> None:
peer_connection = rtc.create_peer_connection()
assert rtc.add_audio_track(peer_connection, 'sendonly', 111) > 0
datachannel_module.create_static_library().rtcDeletePeerConnection(peer_connection)
# TODO: enable again
@pytest.mark.skip
def test_add_video_track() -> None:
peer_connection = rtc.create_peer_connection()
assert rtc.add_video_track(peer_connection, 'sendonly', 96) > 0
datachannel_module.create_static_library().rtcDeletePeerConnection(peer_connection)
# TODO: enable again
@pytest.mark.skip
def test_negotiate_sdp() -> None:
datachannel_library = datachannel_module.create_static_library()
sender_connection = rtc.create_peer_connection()
rtc.add_video_track(sender_connection, 'sendonly', 96)
rtc.add_audio_track(sender_connection, 'sendonly', 111)
sdp_offer = rtc.create_sdp(sender_connection)
receiver_connection = rtc.create_peer_connection()
rtc.add_video_track(receiver_connection, 'recvonly', 96)
rtc.add_audio_track(receiver_connection, 'recvonly', 111)
sdp_answer = rtc.negotiate_sdp(receiver_connection, sdp_offer)
assert sdp_answer
assert 'm=video' in sdp_answer
assert 'VP8/90000' in sdp_answer
assert 'm=audio' in sdp_answer
assert 'opus/48000/2' in sdp_answer
assert datachannel_library.rtcDeletePeerConnection(sender_connection) == 0
assert datachannel_library.rtcDeletePeerConnection(receiver_connection) == 0
# TODO: enable again
@pytest.mark.skip
def test_delete_peers() -> None:
datachannel_library = datachannel_module.create_static_library()
peer_connection = rtc.create_peer_connection()
rtc_peers : List[RtcPeer] =\
[
{
'peer_connection': peer_connection,
'video_track': 0,
'audio_track': 0
}
]
rtc.delete_peers(rtc_peers)
assert datachannel_library.rtcDeletePeerConnection(peer_connection) < 0