Skip to content
This repository has been archived by the owner on May 10, 2024. It is now read-only.

Commit

Permalink
add tests for routes.py
Browse files Browse the repository at this point in the history
  • Loading branch information
cainky committed Sep 23, 2023
1 parent ddb8ccb commit 4d1b92b
Show file tree
Hide file tree
Showing 3 changed files with 110 additions and 6 deletions.
12 changes: 10 additions & 2 deletions backend/routes.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import os
from flask import jsonify, request, send_from_directory
from flask import jsonify, request, Response
from pathlib import Path
from services.merge_service import merge_audio_video
from utils import get_uploads_dir
Expand All @@ -14,7 +14,15 @@ def allowed_file(filename):
def init_app(app):
@app.route("/uploads/<filename>", methods=["GET"])
def uploaded_file(filename):
return send_from_directory(get_uploads_dir(), filename)
uploads_dir = get_uploads_dir()
filepath = os.path.join(uploads_dir, filename)

try:
with open(filepath, "rb") as f:
file_content = f.read()
return Response(file_content, content_type="video/webm")
except FileNotFoundError:
return jsonify(error="File not found"), 404

@app.route("/api/merge", methods=["POST"])
def merge():
Expand Down
7 changes: 3 additions & 4 deletions backend/services/wav2lip_inference.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,9 @@


def run_command(cmd, workdir=None):
if workdir:
result = subprocess.run(cmd, text=True, cwd=workdir)
else:
result = subprocess.run(cmd, text=True)
result = subprocess.run(
cmd, text=True, cwd=workdir, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
if result.returncode != 0:
raise Exception(
f"Command {cmd[0]} failed with return code {result.returncode}."
Expand Down
97 changes: 97 additions & 0 deletions backend/tests/test_routes.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
import unittest, os
from unittest.mock import patch, MagicMock
from app import create_app
from werkzeug.datastructures import FileStorage


class TestRoutes(unittest.TestCase):
def setUp(self):
self.app = create_app()
self.client = self.app.test_client()
self.sample_audio_file = FileStorage(filename="test_audio.webm")
self.sample_video_file = FileStorage(filename="test_video.webm")

def tearDown(self):
file_path = "uploads/existing_file.webm"
if os.path.exists(file_path):
with open(file_path, "rb") as f:
f.close()
os.remove(file_path)

def test_get_existing_upload_file(self):
with open("uploads/existing_file.webm", "w") as f:
f.write("dummy content")

try:
response = self.client.get("/uploads/existing_file.webm")
self.assertEqual(response.status_code, 200)
finally:
os.remove("uploads/existing_file.webm")

def test_get_non_existing_upload_file(self):
response = self.client.get("/uploads/nonexistent_file.webm")
self.assertEqual(response.status_code, 404)

@patch("routes.merge_audio_video", return_value="uploads/output.mp4")
@patch("pathlib.Path.is_file", return_value=True)
def test_merge_with_valid_files(self, mock_is_file, mock_merge):
data = {
"audio": (self.sample_audio_file, "audio.webm"),
"video": (self.sample_video_file, "video.webm"),
}

response = self.client.post(
"/api/merge", content_type="multipart/form-data", data=data
)
self.assertEqual(response.status_code, 200, response.data)
self.assertIn("videoPath", response.json)

def test_merge_missing_audio_or_video(self):
data = {"audio": (self.sample_audio_file, "audio.webm")}
response = self.client.post(
"/api/merge", content_type="multipart/form-data", data=data
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.json["error"], "Missing audio or video file")

def test_merge_empty_filenames(self):
empty_audio_file = FileStorage(filename="")
data = {
"audio": (empty_audio_file, ""),
"video": (self.sample_video_file, "video.webm"),
}
response = self.client.post(
"/api/merge", content_type="multipart/form-data", data=data
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.json["error"], "No selected file")

def test_merge_disallowed_file_extensions(self):
disallowed_audio_file = FileStorage(filename="audio.txt")
data = {
"audio": (disallowed_audio_file, "audio.txt"),
"video": (self.sample_video_file, "video.webm"),
}
response = self.client.post(
"/api/merge", content_type="multipart/form-data", data=data
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.json["error"], "File type not allowed")

def test_merge_error_during_process(self):
error_msg_prefix = "Output video not found after processing."

with patch(
"services.merge_service.merge_audio_video",
side_effect=Exception("Error: Command ffmpeg failed with return code 1."),
):
data = {
"audio": (self.sample_audio_file, "audio.webm"),
"video": (self.sample_video_file, "video.webm"),
}
response = self.client.post(
"/api/merge", content_type="multipart/form-data", data=data
)

self.assertEqual(response.status_code, 500)
self.assertTrue(response.json["error"].startswith(error_msg_prefix))

0 comments on commit 4d1b92b

Please sign in to comment.