Tests for the job runner part3

facefusion/job_helper.py

@@ -0,0 +1,12 @@
+import os
+
+from facefusion.filesystem import is_directory
+
+
+def get_step_output_path(job_id : str, step_index : int, output_path : str) -> str:
+	if is_directory(output_path):
+		return os.path.join(output_path, job_id + '-' + str(step_index))
+
+	output_directory_path, file_name_with_extension = os.path.split(output_path)
+	output_file_name, output_file_extension = os.path.splitext(file_name_with_extension)
+	return os.path.join(output_directory_path, output_file_name + '-' + job_id + '-' + str(step_index) + output_file_extension)

facefusion/job_runner.py

@@ -1,7 +1,6 @@
-import os
-
 from facefusion.ffmpeg import concat_video
-from facefusion.filesystem import is_video, move_file, is_directory
+from facefusion.filesystem import is_video, move_file
+from facefusion.job_helper import get_step_output_path
 from facefusion.job_manager import find_job_ids, get_steps, set_step_status, move_job_file
 from facefusion.typing import ProcessStep, JobMergeSet
 
@@ -61,14 +60,3 @@ def collect_merge_set(job_id : str) -> JobMergeSet:
 		if step_output_path:
 			merge_set.setdefault(output_path, []).append(step_output_path)
 	return merge_set
-
-
-def get_step_output_path(job_id : str, step_index : int, output_path : str) -> str:
-	if is_directory(output_path):
-		output_directory_path, _ = os.path.split(output_path)
-		output_directory_name = os.path.basename(output_directory_path)
-		return os.path.join(output_directory_path, output_directory_name + '-' + job_id + '-' + str(step_index))
-
-	output_directory_path, file_name_with_extension = os.path.split(output_path)
-	output_file_name, output_file_extension = os.path.splitext(file_name_with_extension)
-	return os.path.join(output_directory_path, output_file_name + '-' + job_id + '-' + str(step_index) + output_file_extension)

tests/test_job_runner.py

@@ -8,7 +8,7 @@
 from facefusion.download import conditional_download
 from facefusion.job_manager import init_jobs, clear_jobs, create_job, add_step
 from facefusion.job_runner import run_job, run_jobs, collect_merge_set
-from .helper import get_test_jobs_directory, get_test_examples_directory, get_test_example_file, get_test_output_file
+from .helper import get_test_jobs_directory, get_test_examples_directory, get_test_example_file, get_test_outputs_directory, prepare_test_output_directory
 
 
 @pytest.fixture(scope = 'module', autouse = True)
@@ -27,6 +27,7 @@ def before_all() -> None:
 def before_each() -> None:
 	clear_jobs(get_test_jobs_directory())
 	init_jobs(get_test_jobs_directory())
+	prepare_test_output_directory()
 
 
 def process_step(step_args : Args) -> bool:
@@ -129,7 +130,7 @@ def test_collect_merge_set() -> None:
 	{
 		'source_path': get_test_example_file('source.jpg'),
 		'target_path': get_test_example_file('target-1080p.jpg'),
-		'output_path': get_test_example_file('output')
+		'output_path': get_test_outputs_directory()
 	}
 
 	create_job('job-collect-merge-set')
@@ -149,9 +150,9 @@ def test_collect_merge_set() -> None:
 		[
 			os.path.join(tempfile.gettempdir(), 'test-examples', 'output-job-collect-merge-set-2.jpg')
 		],
-		os.path.join(get_test_examples_directory(), 'output'):
+		get_test_outputs_directory():
 		[
-			os.path.join(get_test_examples_directory(), 'output-job-collect-merge-set-3')
+			os.path.join(tempfile.gettempdir(), 'test-outputs', 'job-collect-merge-set-3')
 		]
 	}