forked from omeryusufyagci/fast-music-remover
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add tests, refactor backend and improve logs
*Add: Add unit tests for all static functions in utils.py and media_handler.py *improve: Improve the logs, remove irrelevant logs *refactor: remove unused import statements, change structure <Signed off by Prakash([email protected])>
- Loading branch information
1 parent
3291b21
commit 24f1799
Showing
6 changed files
with
192 additions
and
37 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| import tempfile | ||
| import unittest | ||
| from unittest.mock import patch, MagicMock | ||
| from pathlib import Path | ||
| from media_handler import MediaHandler | ||
| import json | ||
|
|
||
| class TestMediaHandler(unittest.TestCase): | ||
|
|
||
| def setUp(self): | ||
| """Set up base directory and mock paths for testing.""" | ||
| self.base_directory = tempfile.mkdtemp() | ||
| self.video_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ" | ||
| self.config_path = "/path/to/config.json" | ||
| self.video_path = "/path/to/video.mp4" | ||
| self.input_data = { | ||
| "video_file_path": self.video_path, | ||
| "config_file_path": self.config_path | ||
| } | ||
|
|
||
| @patch("media_handler.yt_dlp.YoutubeDL") | ||
| def test_download_media(self, mock_yt_dlp): | ||
| # Mock YoutubeDL instance | ||
| mock_ydl_instance = MagicMock() | ||
|
|
||
| # Mock extract_info for download=False and download=True | ||
| mock_ydl_instance.extract_info.side_effect = [ | ||
| {"title": "Test Video", "ext": "mp4"}, # For download=False | ||
| {"ext": "mp4"} # For download=True | ||
| ] | ||
|
|
||
| # Set the return value when instantiating YoutubeDL | ||
| mock_yt_dl_context_manager = MagicMock() | ||
| mock_yt_dl_context_manager.__enter__.return_value = mock_ydl_instance | ||
| mock_yt_dlp.return_value = mock_yt_dl_context_manager | ||
|
|
||
| # Test download_media method | ||
| result = MediaHandler.download_media(self.video_url, self.base_directory) | ||
| expected_file = Path(self.base_directory) / "Test_Video.mp4" | ||
| self.assertEqual(result, str(expected_file.resolve())) | ||
|
|
||
| @patch("media_handler.subprocess.run") | ||
| @patch("media_handler.ResponseHandler.core_data_passer") | ||
| def test_process_with_media_processor_success(self, mock_core_data_passer, mock_subprocess_run): | ||
| # Mock response from core_data_passer | ||
| mock_core_data_passer.return_value = json.dumps( | ||
| { "status": "success", | ||
| "message": "the input data", | ||
| "data": self.input_data} | ||
| ) | ||
| # Mock subprocess.run to simulate a successful response | ||
| mock_subprocess_run.return_value = MagicMock( | ||
| returncode=0, | ||
| stdout=json.dumps({ | ||
| "status": "success", | ||
| "data": {"processed_video_path": "/path/to/processed_video.mp4"} | ||
| }), | ||
| stderr="" | ||
| ) | ||
|
|
||
| # Test process_with_media_processor | ||
| result = MediaHandler.process_with_media_processor( | ||
| self.video_path, self.base_directory, self.config_path | ||
| ) | ||
|
|
||
| self.assertEqual(result, "/path/to/processed_video.mp4") | ||
|
|
||
|
|
||
|
|
||
|
|
||
|
|
||
| @patch("media_handler.subprocess.run") | ||
| def test_process_with_media_processor_failure(self, mock_subprocess_run): | ||
| # Mock subprocess.run to simulate a failed response | ||
| mock_subprocess_run.return_value = MagicMock( | ||
| returncode=1, | ||
| stdout="", | ||
| stderr="Error processing video" | ||
| ) | ||
|
|
||
| # Test process_with_media_processor with failure | ||
| result = MediaHandler.process_with_media_processor( | ||
| self.video_path, self.base_directory, self.config_path | ||
| ) | ||
|
|
||
| self.assertIsNone(result) | ||
|
|
||
| @patch("media_handler.subprocess.run") | ||
| def test_process_with_media_processor_invalid_json(self, mock_subprocess_run): | ||
| # Mock subprocess.run to simulate invalid JSON output | ||
| mock_subprocess_run.return_value = MagicMock( | ||
| returncode=0, | ||
| stdout="Invalid JSON Output", | ||
| stderr="" | ||
| ) | ||
|
|
||
| # Test process_with_media_processor with invalid JSON | ||
| result = MediaHandler.process_with_media_processor( | ||
| self.video_path, self.base_directory, self.config_path | ||
| ) | ||
|
|
||
| self.assertIsNone(result) | ||
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| import unittest | ||
| import os | ||
| import tempfile | ||
| from pathlib import Path | ||
| from utils import Utils | ||
|
|
||
| class TestUtils(unittest.TestCase): | ||
|
|
||
| def setUp(self): | ||
| """Set up temporary directories and files for testing.""" | ||
| self.test_dir = tempfile.mkdtemp() | ||
| self.upload_folder = self.test_dir | ||
| self.base_filename = "test_file" | ||
|
|
||
| def tearDown(self): | ||
| """Clean up the temporary directory.""" | ||
| for root, dirs, files in os.walk(self.test_dir, topdown=False): | ||
| for name in files: | ||
| os.remove(os.path.join(root, name)) | ||
| for name in dirs: | ||
| os.rmdir(os.path.join(root, name)) | ||
|
|
||
| def test_ensure_dir_exists(self): | ||
| # Test creating a new directory | ||
| new_dir = Path(self.test_dir) / "new_folder" | ||
| Utils.ensure_dir_exists(str(new_dir)) | ||
| self.assertTrue(new_dir.exists()) | ||
|
|
||
| def test_remove_files_by_base(self): | ||
| # Create temporary files for testing | ||
| file_paths = [ | ||
| Path(self.upload_folder) / (f"{self.base_filename}.webm"), | ||
| Path(self.upload_folder) / (f"{self.base_filename}_isolated_audio.wav"), | ||
| Path(self.upload_folder) / (f"{self.base_filename}_processed_video.mp4") | ||
| ] | ||
| # Create the files | ||
| for file_path in file_paths: | ||
| file_path.touch() | ||
|
|
||
| # Ensure the files were created | ||
| for file_path in file_paths: | ||
| self.assertTrue(file_path.exists()) | ||
|
|
||
| # Remove files using the method | ||
| Utils.remove_files_by_base(self.base_filename, self.upload_folder) | ||
|
|
||
| # Check if files are removed | ||
| for file_path in file_paths: | ||
| self.assertFalse(file_path.exists()) | ||
|
|
||
|
|
||
| def test_sanitize_filename(self): | ||
| # Test filename sanitization | ||
| sanitized_filename = Utils.sanitize_filename("test@file!.mp4") | ||
| self.assertEqual(sanitized_filename, "test_file_.mp4") | ||
|
|
||
| sanitized_filename = Utils.sanitize_filename("valid_file-name.mp4") | ||
| self.assertEqual(sanitized_filename, "valid_file-name.mp4") | ||
|
|
||
| def test_validate_url(self): | ||
| # Test valid URLs | ||
| valid_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ" | ||
| self.assertTrue(Utils.validate_url(valid_url)) | ||
|
|
||
| # Test invalid URLs | ||
| invalid_url = "invalid_url" | ||
| self.assertFalse(Utils.validate_url(invalid_url)) | ||
|
|
||
| no_scheme_url = "www.youtube.com" | ||
| self.assertFalse(Utils.validate_url(no_scheme_url)) | ||
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters