123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373 |
- from time import sleep
- import pytest
- from facefusion.jobs.job_helper import get_step_output_path
- from facefusion.jobs.job_manager import add_step, clear_jobs, count_step_total, create_job, delete_job, delete_jobs, find_job_ids, get_steps, init_jobs, insert_step, move_job_file, remix_step, remove_step, set_step_status, set_steps_status, submit_job, submit_jobs
- from .helper import get_test_jobs_directory
- @pytest.fixture(scope = 'function', autouse = True)
- def before_each() -> None:
- clear_jobs(get_test_jobs_directory())
- init_jobs(get_test_jobs_directory())
- def test_create_job() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- assert create_job('job-test-create-job') is True
- assert create_job('job-test-create-job') is False
- add_step('job-test-submit-job', args_1)
- submit_job('job-test-create-job')
- assert create_job('job-test-create-job') is False
- def test_submit_job() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- assert submit_job('job-invalid') is False
- create_job('job-test-submit-job')
- assert submit_job('job-test-submit-job') is False
- add_step('job-test-submit-job', args_1)
- assert submit_job('job-test-submit-job') is True
- assert submit_job('job-test-submit-job') is False
- def test_submit_jobs() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- args_2 =\
- {
- 'source_path': 'source-2.jpg',
- 'target_path': 'target-2.jpg',
- 'output_path': 'output-2.jpg'
- }
- assert submit_jobs() is False
- create_job('job-test-submit-jobs-1')
- create_job('job-test-submit-jobs-2')
- assert submit_jobs() is False
- add_step('job-test-submit-jobs-1', args_1)
- add_step('job-test-submit-jobs-2', args_2)
- assert submit_jobs() is True
- assert submit_jobs() is False
- def test_delete_job() -> None:
- assert delete_job('job-invalid') is False
- create_job('job-test-delete-job')
- assert delete_job('job-test-delete-job') is True
- assert delete_job('job-test-delete-job') is False
- def test_delete_jobs() -> None:
- assert delete_jobs() is False
- create_job('job-test-delete-jobs-1')
- create_job('job-test-delete-jobs-2')
- assert delete_jobs() is True
- @pytest.mark.skip()
- def test_find_jobs() -> None:
- pass
- def test_find_job_ids() -> None:
- create_job('job-test-find-job-ids-1')
- sleep(0.5)
- create_job('job-test-find-job-ids-2')
- sleep(0.5)
- create_job('job-test-find-job-ids-3')
- assert find_job_ids('drafted') == [ 'job-test-find-job-ids-1', 'job-test-find-job-ids-2', 'job-test-find-job-ids-3' ]
- assert find_job_ids('queued') == []
- assert find_job_ids('completed') == []
- assert find_job_ids('failed') == []
- move_job_file('job-test-find-job-ids-1', 'queued')
- move_job_file('job-test-find-job-ids-2', 'queued')
- move_job_file('job-test-find-job-ids-3', 'queued')
- assert find_job_ids('drafted') == []
- assert find_job_ids('queued') == [ 'job-test-find-job-ids-1', 'job-test-find-job-ids-2', 'job-test-find-job-ids-3' ]
- assert find_job_ids('completed') == []
- assert find_job_ids('failed') == []
- move_job_file('job-test-find-job-ids-1', 'completed')
- assert find_job_ids('drafted') == []
- assert find_job_ids('queued') == [ 'job-test-find-job-ids-2', 'job-test-find-job-ids-3' ]
- assert find_job_ids('completed') == [ 'job-test-find-job-ids-1' ]
- assert find_job_ids('failed') == []
- move_job_file('job-test-find-job-ids-2', 'failed')
- assert find_job_ids('drafted') == []
- assert find_job_ids('queued') == [ 'job-test-find-job-ids-3' ]
- assert find_job_ids('completed') == [ 'job-test-find-job-ids-1' ]
- assert find_job_ids('failed') == [ 'job-test-find-job-ids-2' ]
- move_job_file('job-test-find-job-ids-3', 'completed')
- assert find_job_ids('drafted') == []
- assert find_job_ids('queued') == []
- assert find_job_ids('completed') == [ 'job-test-find-job-ids-1', 'job-test-find-job-ids-3' ]
- assert find_job_ids('failed') == [ 'job-test-find-job-ids-2' ]
- def test_add_step() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- args_2 =\
- {
- 'source_path': 'source-2.jpg',
- 'target_path': 'target-2.jpg',
- 'output_path': 'output-2.jpg'
- }
- assert add_step('job-invalid', args_1) is False
- create_job('job-test-add-step')
- assert add_step('job-test-add-step', args_1) is True
- assert add_step('job-test-add-step', args_2) is True
- steps = get_steps('job-test-add-step')
- assert steps[0].get('args') == args_1
- assert steps[1].get('args') == args_2
- assert count_step_total('job-test-add-step') == 2
- def test_remix_step() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- args_2 =\
- {
- 'source_path': 'source-2.jpg',
- 'target_path': 'target-2.jpg',
- 'output_path': 'output-2.jpg'
- }
- assert remix_step('job-invalid', 0, args_1) is False
- create_job('job-test-remix-step')
- add_step('job-test-remix-step', args_1)
- add_step('job-test-remix-step', args_2)
- assert remix_step('job-test-remix-step', 99, args_1) is False
- assert remix_step('job-test-remix-step', 0, args_2) is True
- assert remix_step('job-test-remix-step', -1, args_2) is True
- steps = get_steps('job-test-remix-step')
- assert steps[0].get('args') == args_1
- assert steps[1].get('args') == args_2
- assert steps[2].get('args').get('source_path') == args_2.get('source_path')
- assert steps[2].get('args').get('target_path') == get_step_output_path('job-test-remix-step', 0, args_1.get('output_path'))
- assert steps[2].get('args').get('output_path') == args_2.get('output_path')
- assert steps[3].get('args').get('source_path') == args_2.get('source_path')
- assert steps[3].get('args').get('target_path') == get_step_output_path('job-test-remix-step', 2, args_2.get('output_path'))
- assert steps[3].get('args').get('output_path') == args_2.get('output_path')
- assert count_step_total('job-test-remix-step') == 4
- def test_insert_step() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- args_2 =\
- {
- 'source_path': 'source-2.jpg',
- 'target_path': 'target-2.jpg',
- 'output_path': 'output-2.jpg'
- }
- args_3 =\
- {
- 'source_path': 'source-3.jpg',
- 'target_path': 'target-3.jpg',
- 'output_path': 'output-3.jpg'
- }
- assert insert_step('job-invalid', 0, args_1) is False
- create_job('job-test-insert-step')
- add_step('job-test-insert-step', args_1)
- add_step('job-test-insert-step', args_1)
- assert insert_step('job-test-insert-step', 99, args_1) is False
- assert insert_step('job-test-insert-step', 0, args_2) is True
- assert insert_step('job-test-insert-step', -1, args_3) is True
- steps = get_steps('job-test-insert-step')
- assert steps[0].get('args') == args_2
- assert steps[1].get('args') == args_1
- assert steps[2].get('args') == args_3
- assert steps[3].get('args') == args_1
- assert count_step_total('job-test-insert-step') == 4
- def test_remove_step() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- args_2 =\
- {
- 'source_path': 'source-2.jpg',
- 'target_path': 'target-2.jpg',
- 'output_path': 'output-2.jpg'
- }
- args_3 =\
- {
- 'source_path': 'source-3.jpg',
- 'target_path': 'target-3.jpg',
- 'output_path': 'output-3.jpg'
- }
- assert remove_step('job-invalid', 0) is False
- create_job('job-test-remove-step')
- add_step('job-test-remove-step', args_1)
- add_step('job-test-remove-step', args_2)
- add_step('job-test-remove-step', args_1)
- add_step('job-test-remove-step', args_3)
- assert remove_step('job-test-remove-step', 99) is False
- assert remove_step('job-test-remove-step', 0) is True
- assert remove_step('job-test-remove-step', -1) is True
- steps = get_steps('job-test-remove-step')
- assert steps[0].get('args') == args_2
- assert steps[1].get('args') == args_1
- assert count_step_total('job-test-remove-step') == 2
- def test_get_steps() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- args_2 =\
- {
- 'source_path': 'source-2.jpg',
- 'target_path': 'target-2.jpg',
- 'output_path': 'output-2.jpg'
- }
- assert get_steps('job-invalid') == []
- create_job('job-test-get-steps')
- add_step('job-test-get-steps', args_1)
- add_step('job-test-get-steps', args_2)
- steps = get_steps('job-test-get-steps')
- assert steps[0].get('args') == args_1
- assert steps[1].get('args') == args_2
- assert count_step_total('job-test-get-steps') == 2
- def test_set_step_status() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- args_2 =\
- {
- 'source_path': 'source-2.jpg',
- 'target_path': 'target-2.jpg',
- 'output_path': 'output-2.jpg'
- }
- assert set_step_status('job-invalid', 0, 'completed') is False
- create_job('job-test-set-step-status')
- add_step('job-test-set-step-status', args_1)
- add_step('job-test-set-step-status', args_2)
- assert set_step_status('job-test-set-step-status', 99, 'completed') is False
- assert set_step_status('job-test-set-step-status', 0, 'completed') is True
- assert set_step_status('job-test-set-step-status', 1, 'failed') is True
- steps = get_steps('job-test-set-step-status')
- assert steps[0].get('status') == 'completed'
- assert steps[1].get('status') == 'failed'
- assert count_step_total('job-test-set-step-status') == 2
- def test_set_steps_status() -> None:
- args_1 =\
- {
- 'source_path': 'source-1.jpg',
- 'target_path': 'target-1.jpg',
- 'output_path': 'output-1.jpg'
- }
- args_2 =\
- {
- 'source_path': 'source-2.jpg',
- 'target_path': 'target-2.jpg',
- 'output_path': 'output-2.jpg'
- }
- assert set_steps_status('job-invalid', 'queued') is False
- create_job('job-test-set-steps-status')
- add_step('job-test-set-steps-status', args_1)
- add_step('job-test-set-steps-status', args_2)
- assert set_steps_status('job-test-set-steps-status', 'queued') is True
- steps = get_steps('job-test-set-steps-status')
- assert steps[0].get('status') == 'queued'
- assert steps[1].get('status') == 'queued'
- assert count_step_total('job-test-set-steps-status') == 2
|