upload asset endpoint

facefusion/apis/asset_store.py

@@ -80,7 +80,7 @@ def clear_session(session_id : SessionId) -> None:
 		for asset in ASSET_STORE[session_id].values():
 			file_path = asset.get('path')
 
-			if file_path and is_file(file_path):
+			if is_file(file_path):
 				remove_file(file_path)
 
 		del ASSET_STORE[session_id]

facefusion/apis/endpoints/assets.py

@@ -25,12 +25,12 @@ async def upload_asset(request : Request) -> JSONResponse:
 		if asset_type == 'target':
 			files = files[:1]
 
-		prepared_files = await prepare_files(files)
+		media_files = await prepare_media_files(files)
 
-		if prepared_files:
+		if media_files:
 			asset_ids : List[str] = []
 
-			for file_path, media_type in prepared_files:
+			for file_path, media_type in media_files:
 				asset = asset_store.create_asset(session_id, asset_type, media_type, file_path) #type:ignore[arg-type]
 
 				if asset:
@@ -48,8 +48,8 @@ async def upload_asset(request : Request) -> JSONResponse:
 	return JSONResponse({}, status_code = HTTP_400_BAD_REQUEST)
 
 
-async def prepare_files(files : List[UploadFile]) -> List[Tuple[str, MediaType]]:
+async def prepare_media_files(files : List[UploadFile]) -> List[Tuple[str, MediaType]]:
-	prepared_files : List[Tuple[str, MediaType]] = []
+	media_files : List[Tuple[str, MediaType]] = []
 
 	for file in files:
 		file_extension = get_file_extension(file.filename)
@@ -69,9 +69,9 @@ async def prepare_files(files : List[UploadFile]) -> List[Tuple[str, MediaType]]
 			media_type = 'video'
 
 		if media_type:
-			prepared_files.append((file_path, media_type))
+			media_files.append((file_path, media_type))
 
 		if not media_type and is_file(file_path):
 			remove_file(file_path)
 
-	return prepared_files
+	return media_files

tests/test_api_assets.py

@@ -6,6 +6,17 @@ from starlette.testclient import TestClient
 from facefusion import metadata, session_manager
 from facefusion.apis import asset_store
 from facefusion.apis.core import create_api
+from facefusion.download import conditional_download
+from .helper import get_test_example_file, get_test_examples_directory
+
+
+@pytest.fixture(scope = 'module', autouse = True)
+def before_all() -> None:
+	conditional_download(get_test_examples_directory(),
+	[
+		'https://github.com/facefusion/facefusion-assets/releases/download/examples-3.0.0/source.jpg',
+		'https://github.com/facefusion/facefusion-assets/releases/download/examples-3.0.0/target-240p.mp4'
+	])
 
 
 @pytest.fixture(scope = 'module')
@@ -63,7 +74,7 @@ def test_upload_source_asset(test_client : TestClient) -> None:
 	})
 	create_session_body = create_session_response.json()
 
-	with open('.assets/examples/source.jpg', 'rb') as source_file:
+	with open(get_test_example_file('source.jpg'), 'rb') as source_file:
 		upload_response = test_client.post('/assets?type=source', headers =
 		{
 			'Authorization': 'Bearer ' + create_session_body.get('access_token')
@@ -84,16 +95,16 @@ def test_upload_multiple_source_assets(test_client : TestClient) -> None:
 	})
 	create_session_body = create_session_response.json()
 
-	with open('.assets/examples/source.jpg', 'rb') as source_file_1:
+	with open(get_test_example_file('source.jpg'), 'rb') as source_file:
-		with open('.assets/examples/source.jpg', 'rb') as source_file_2:
+		source_content = source_file.read()
-			upload_response = test_client.post('/assets?type=source', headers =
+		upload_response = test_client.post('/assets?type=source', headers =
-			{
+		{
-				'Authorization': 'Bearer ' + create_session_body.get('access_token')
+			'Authorization': 'Bearer ' + create_session_body.get('access_token')
-			}, files =
+		}, files =
-			[
+		[
-				('file', ('source1.jpg', source_file_1, 'image/jpeg')),
+			('file', ('source1.jpg', source_content, 'image/jpeg')),
-				('file', ('source2.jpg', source_file_2, 'image/jpeg'))
+			('file', ('source2.jpg', source_content, 'image/jpeg'))
-			])
+		])
 
 	assert upload_response.status_code == 201
 	assert upload_response.json().get('asset_ids')
@@ -107,7 +118,7 @@ def test_upload_target_asset(test_client : TestClient) -> None:
 	})
 	create_session_body = create_session_response.json()
 
-	with open('.assets/examples/target-240p.mp4', 'rb') as target_file:
+	with open(get_test_example_file('target-240p.mp4'), 'rb') as target_file:
 		upload_response = test_client.post('/assets?type=target', headers =
 		{
 			'Authorization': 'Bearer ' + create_session_body.get('access_token')