mirror of
https://github.com/facefusion/facefusion.git
synced 2026-11-12 21:34:27 +01:00
Henry RuhsandClaude Fable 5.1
9b00b79182
* stop every session on exit Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Cd6JuxiVXYNPCDii9go2eh * log api start before serving, test api route Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Cd6JuxiVXYNPCDii9go2eh * move all to its place * move all to its place --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
103 lines
2.1 KiB
Python
103 lines
2.1 KiB
Python
from typing import Iterator
|
|
|
|
import pytest
|
|
|
|
from facefusion import session_context, session_manager
|
|
from facefusion.process_manager import clear, end_process, get_state, init, is_pending, is_processing, is_stopping, set_state, start, stop_process
|
|
|
|
|
|
@pytest.fixture(scope = 'function', autouse = True)
|
|
def before_each() -> Iterator[None]:
|
|
local_id = session_context.resolve_local_id()
|
|
|
|
session_context.set_session_id(local_id)
|
|
clear()
|
|
session_context.set_session_id('session-a')
|
|
clear()
|
|
session_context.set_session_id('session-b')
|
|
clear()
|
|
session_context.set_session_id('session-a')
|
|
|
|
yield
|
|
|
|
session_context.set_session_id(local_id)
|
|
|
|
|
|
def test_init() -> None:
|
|
set_state('processing')
|
|
init()
|
|
|
|
assert get_state() == 'pending'
|
|
|
|
|
|
def test_get_state() -> None:
|
|
set_state('processing')
|
|
session_context.set_session_id('session-b')
|
|
set_state('stopping')
|
|
|
|
assert get_state() == 'stopping'
|
|
|
|
session_context.set_session_id('session-a')
|
|
|
|
assert get_state() == 'processing'
|
|
|
|
fork_id = session_manager.fork_session()
|
|
clear()
|
|
set_state('processing')
|
|
|
|
assert get_state() == 'processing'
|
|
|
|
session_context.set_session_id('session-a')
|
|
set_state('stopping')
|
|
session_context.set_session_id(fork_id)
|
|
|
|
assert get_state() == 'stopping'
|
|
|
|
session_context.set_session_id('session-a')
|
|
set_state('processing')
|
|
session_context.set_session_id(fork_id)
|
|
|
|
assert get_state() == 'processing'
|
|
|
|
local_id = session_context.resolve_local_id()
|
|
session_context.set_session_id(local_id)
|
|
set_state('stopping')
|
|
session_context.set_session_id(fork_id)
|
|
|
|
assert get_state() == 'stopping'
|
|
|
|
session_context.set_session_id('session-b')
|
|
|
|
assert get_state() == 'stopping'
|
|
|
|
session_context.set_session_id(fork_id)
|
|
session_manager.join_session()
|
|
|
|
|
|
def test_start() -> None:
|
|
set_state('pending')
|
|
start()
|
|
|
|
assert is_processing()
|
|
|
|
|
|
def test_stop_process() -> None:
|
|
set_state('processing')
|
|
stop_process()
|
|
|
|
assert is_stopping()
|
|
|
|
|
|
def test_end_process() -> None:
|
|
set_state('processing')
|
|
end_process()
|
|
|
|
assert is_pending()
|
|
|
|
|
|
def test_clear() -> None:
|
|
set_state('processing')
|
|
clear()
|
|
|
|
assert get_state() == 'pending'
|