CalvinBackup/facefusion
1
0
Fork 0
mirror of https://github.com/facefusion/facefusion.git synced 2026-06-02 10:51:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

hash based test for test_stream_image (#1104)

Browse Source 
* hash based test for test_stream_image

* new todo for test_video_stream

* new todo for test_video_stream
This commit is contained in:
Henry Ruhs
2026-05-12 16:21:19 +02:00
committed by GitHub
parent 5e39c60b5c
commit 9453a042a1
1 changed files with 6 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_api_stream.py
+6 -24
View File
@@ -2,16 +2,15 @@ import tempfile
import threading
from typing import Iterator
import cv2
import numpy
import pytest
from starlette.testclient import TestClient
from facefusion import metadata, session_manager, state_manager
from facefusion.apis import asset_store
from facefusion.apis.core import create_api
from facefusion.core import common_pre_check, processors_pre_check
from facefusion.core import common_pre_check
from facefusion.download import conditional_download
from facefusion.hash_helper import create_hash
from .assert_helper import get_test_example_file, get_test_examples_directory
from .stream_helper import create_sdp_offer, open_websocket_stream
@@ -22,22 +21,9 @@ def before_all() -> None:
state_manager.init_item('execution_providers', [ 'cpu' ])
state_manager.init_item('download_providers', [ 'github', 'huggingface' ])
state_manager.init_item('temp_path', tempfile.gettempdir())
state_manager.init_item('processors', [ 'face_swapper' ])
state_manager.init_item('face_detector_model', 'yolo_face')
state_manager.init_item('face_detector_size', '640x640')
state_manager.init_item('face_detector_score', 0.5)
state_manager.init_item('face_detector_angles', [ 0 ])
state_manager.init_item('face_detector_margin', [ 0, 0, 0, 0 ])
state_manager.init_item('face_landmarker_model', '2dfan4')
state_manager.init_item('face_landmarker_score', 0.5)
state_manager.init_item('face_mask_types', [ 'box' ])
state_manager.init_item('face_mask_blur', 0.3)
state_manager.init_item('face_mask_padding', [ 0, 0, 0, 0 ])
state_manager.init_item('face_swapper_model', 'hyperswap_1a_256')
state_manager.init_item('face_swapper_pixel_boost', '256x256')
state_manager.init_item('processors', [])
common_pre_check()
processors_pre_check()
conditional_download(get_test_examples_directory(),
[
@@ -92,14 +78,12 @@ def test_stream_image(test_client : TestClient) -> None:
'access_token.' + access_token
]) as websocket:
websocket.send_bytes(source_content)
output_bytes = websocket.receive_bytes()
output_vision_frame = cv2.imdecode(numpy.frombuffer(output_bytes, numpy.uint8), cv2.IMREAD_COLOR)
output_buffer = websocket.receive_bytes()
# TODO: can we test if bytes have been passed? what does this actual test than just the handshake?
# TODO: compare to the video test - no status code check?
assert output_vision_frame.shape == (1024, 1024, 3)
assert create_hash(output_buffer) == '0142782f'
#TODO: this test only checks the handshake and sdp offer but no stream of video bytes
def test_stream_video(test_client : TestClient) -> None:
create_session_response = test_client.post('/session', json =
{
@@ -130,7 +114,6 @@ def test_stream_video(test_client : TestClient) -> None:
ready_event = threading.Event()
stop_event = threading.Event()
#TODO: use asyncio
stream_thread = threading.Thread(target = open_websocket_stream, args = (test_client, [ 'access_token.' + access_token ], source_content, ready_event, stop_event))
stream_thread.start()
ready_event.wait(timeout = 10)
@@ -144,7 +127,6 @@ def test_stream_video(test_client : TestClient) -> None:
'Content-Type': 'application/sdp'
})
# TODO: can we test if bytes have been passed? what does this actual test than just the handshake?
assert stream_response.status_code == 201
assert stream_response.text