diff --git a/README.md b/README.md
index 1eec390..af86978 100644
--- a/README.md
+++ b/README.md
@@ -92,6 +92,10 @@ To run your Bot in Heroku 24/7, you will need the Procfile located in root, then
- https://github.com/xrisk/heroku-opus.git
+## Testing
+The tests were written manually with no package due to problems with async function in other packages, to execute them type in root:
+`python run_tests.py`
+
## License
- This program is free software: you can redistribute it and/or modify it under the terms of the [MIT License](https://github.com/RafaelSolVargas/Vulkan/blob/master/LICENSE).
@@ -101,4 +105,4 @@ To run your Bot in Heroku 24/7, you will need the Procfile located in root, then
## Acknowledgment
- - See the DingoLingo [project](https://github.com/Raptor123471/DingoLingo) from Raptor123471, it helped me a lot to build Vulkan.
+ - See the DingoLingo [project](https://github.com/Raptor123471/DingoLingo) from Raptor123471, it helped me a lot to build Vulkan.
\ No newline at end of file
diff --git a/Tests/Colors.py b/Tests/Colors.py
new file mode 100644
index 0000000..d3d2a62
--- /dev/null
+++ b/Tests/Colors.py
@@ -0,0 +1,10 @@
+class Colors:
+ HEADER = '\033[95m'
+ OKBLUE = '\033[94m'
+ OKCYAN = '\033[96m'
+ OKGREEN = '\033[92m'
+ WARNING = '\033[93m'
+ FAIL = '\033[91m'
+ ENDC = '\033[0m'
+ BOLD = '\033[1m'
+ UNDERLINE = '\033[4m'
diff --git a/Tests/LoopRunner.py b/Tests/LoopRunner.py
new file mode 100644
index 0000000..660604a
--- /dev/null
+++ b/Tests/LoopRunner.py
@@ -0,0 +1,50 @@
+import asyncio
+from asyncio import AbstractEventLoop
+from threading import Thread
+from typing import Any, Coroutine, List
+
+
+class LoopRunner(Thread):
+ """
+ Class to help deal with asyncio coroutines and loops
+ Copyright: https://agariinc.medium.com/advanced-strategies-for-testing-async-code-in-python-6196a032d8d7
+ """
+
+ def __init__(self, loop: AbstractEventLoop) -> None:
+ # We ensure to always use the same loop
+ self.loop = loop
+ Thread.__init__(self, name='runner')
+
+ def run(self) -> None:
+ asyncio.set_event_loop(self.loop)
+ try:
+ self.loop.run_forever()
+ finally:
+ if self.loop.is_running():
+ self.loop.close()
+
+ def run_coroutine(self, coroutine: Coroutine) -> Any:
+ """Run a coroutine inside the loop and return the result, doesn't allow concurrency"""
+ result = asyncio.run_coroutine_threadsafe(coroutine, self.loop)
+ return result.result()
+
+ def _stop(self):
+ self.loop.stop()
+
+ def run_in_thread(self, callback, *args):
+ return self.loop.call_soon_threadsafe(callback, *args)
+
+ def stop(self):
+ return self.run_in_thread(self._stop)
+
+ def run_coroutines_list(self, coroutineList: List[Coroutine]) -> None:
+ """Create multiple tasks in the loop and wait for them, use concurrency"""
+ tasks = []
+ for coroutine in coroutineList:
+ tasks.append(self.loop.create_task(coroutine))
+
+ self.run_coroutine(self.__waitForMultipleTasks(tasks))
+
+ async def __waitForMultipleTasks(self, coroutines: List[Coroutine]) -> None:
+ """Function to trigger the await for asyncio.wait coroutines"""
+ await asyncio.wait(coroutines)
diff --git a/Tests/TestBase.py b/Tests/TestBase.py
new file mode 100644
index 0000000..358b42c
--- /dev/null
+++ b/Tests/TestBase.py
@@ -0,0 +1,86 @@
+import asyncio
+from time import time
+from typing import Callable, List, Tuple
+from Tests.Colors import Colors
+from Music.Downloader import Downloader
+from Music.Searcher import Searcher
+from Tests.TestsHelper import TestsConstants
+from Tests.LoopRunner import LoopRunner
+
+
+class VulkanTesterBase:
+ """My own module to execute asyncio tests"""
+
+ def __init__(self) -> None:
+ self._downloader = Downloader()
+ self._searcher = Searcher()
+ self._constants = TestsConstants()
+ # Get the list of methods objects of this class if start with test
+ self._methodsList: List[Callable] = [getattr(self, func) for func in dir(self) if callable(
+ getattr(self, func)) and func.startswith("test")]
+
+ def run(self) -> None:
+ self.__printSeparator()
+ methodsSummary: List[Tuple[Callable, bool]] = []
+ testsSuccessQuant = 0
+ testsStartTime = time()
+
+ for method in self._methodsList:
+ currentTestStartTime = time()
+ self.__printTestStart(method)
+ success = False
+ try:
+ self._setUp()
+ success = method()
+ except Exception as e:
+ success = False
+ print(f'ERROR -> {e}')
+ finally:
+ self._tearDown()
+
+ methodsSummary.append((method, success))
+ runTime = time() - currentTestStartTime # Get the run time of the current test
+ if success:
+ testsSuccessQuant += 1
+ self.__printTestSuccess(method, runTime)
+ else:
+ self.__printTestFailure(method, runTime)
+
+ self.__printSeparator()
+
+ testsRunTime = time() - testsStartTime
+ self.__printTestsSummary(methodsSummary, testsSuccessQuant, testsRunTime)
+
+ def _setUp(self) -> None:
+ self._runner = LoopRunner(asyncio.new_event_loop())
+ self._runner.start()
+
+ def _tearDown(self) -> None:
+ self._runner.stop()
+ self._runner.join()
+
+ def __printTestsSummary(self, methods: List[Tuple[Callable, bool]], totalSuccess: int, runTime: int) -> None:
+ for index, methodResult in enumerate(methods):
+ method = methodResult[0]
+ success = methodResult[1]
+
+ if success:
+ print(f'{Colors.OKGREEN} {index} -> {method.__name__} = Success {Colors.ENDC}')
+ else:
+ print(f'{Colors.FAIL} {index} -> {method.__name__} = Failed {Colors.ENDC}')
+
+ print()
+ print(
+ f'TESTS EXECUTED: {len(methods)} | SUCCESS: {totalSuccess} | FAILED: {len(methods) - totalSuccess} | TIME: {runTime:.2f}sec')
+
+ def __printTestStart(self, method: Callable) -> None:
+ print(f'🧪 - Starting {method.__name__}')
+
+ def __printTestSuccess(self, method: Callable, runTime: int) -> None:
+ print(f'{method.__name__} -> {Colors.OKGREEN} Success {Colors.ENDC} | ⏰ - {runTime:.2f}sec')
+
+ def __printTestFailure(self, method: Callable, runTime: int) -> None:
+ print(f'{method.__name__} -> {Colors.FAIL} Test Failed {Colors.ENDC} | ⏰ - {runTime:.2f}sec')
+
+ def __printSeparator(self) -> None:
+ print('=-=' * 15)
diff --git a/Tests/TestsDownload.py b/Tests/TestsDownload.py
deleted file mode 100644
index 598a156..0000000
--- a/Tests/TestsDownload.py
+++ /dev/null
@@ -1,103 +0,0 @@
-import asyncio
-from typing import List
-import unittest
-from Music.Downloader import Downloader
-from Music.Searcher import Searcher
-from Music.Playlist import Playlist
-from Music.Song import Song
-from Tests.TestsHelper import TestsConstants
-
-
-def myAsyncTest(coro):
- def wrapper(*args, **kwargs):
- loop = asyncio.new_event_loop()
- asyncio.set_event_loop(None)
- try:
- return loop.run_until_complete(coro(*args, **kwargs))
- finally:
- loop.close()
- return wrapper
-
-
-class TestDownloader(unittest.IsolatedAsyncioTestCase):
- def __init__(self, methodName: str = ...) -> None:
- self.downloader = Downloader()
- self.searcher = Searcher()
- self.constants = TestsConstants()
- super().__init__(methodName)
-
- @myAsyncTest
- async def test_emptyString(self):
- musicsList = await self.downloader.extract_info('')
-
- self.assertEqual(musicsList, [], self.constants.EMPTY_STRING_ERROR_MSG)
-
- @myAsyncTest
- async def test_YoutubeMusicUrl(self) -> None:
- musics = await self.searcher.search(self.constants.YT_MUSIC_URL)
-
- self.assertTrue(len(musics) > 0)
-
- @myAsyncTest
- async def test_YoutubePersonalPlaylist(self) -> None:
- musics = await self.searcher.search(self.constants.YT_PERSONAL_PLAYLIST_URL)
-
- self.assertTrue(len(musics) > 0)
-
- @myAsyncTest
- async def test_YoutubeChannelPlaylist(self) -> None:
- # Search the link to determine names
- musicsInfo = await self.searcher.search(self.constants.YT_CHANNEL_PLAYLIST_URL)
-
- self.assertTrue(len(musicsInfo) > 0)
-
- # Create and store songs in list
- playlist = Playlist()
- songsList = []
- for info in musicsInfo:
- song = Song(identifier=info, playlist=playlist, requester='')
- playlist.add_song(song)
- songsList.append(song)
-
- # We need to trigger and wait multiple tasks, so we create multiple tasks with asyncio
- # and then we await for each one of them. We use this because download_song is a Coroutine
- tasks: List[asyncio.Task] = []
- for song in songsList:
- task = asyncio.create_task(self.downloader.download_song(song))
- tasks.append(task)
-
- # Await for each task to finish
- for task in tasks:
- await task
-
- self.assertTrue(self.__verifySuccessfullyPreload(songsList))
-
- @myAsyncTest
- async def test_YoutubeMixPlaylist(self) -> None:
- music = await self.searcher.search(self.constants.YT_MIX_URL)
-
- # Musics from Mix should download only the first music
- self.assertTrue(len(music) == 1)
-
- @myAsyncTest
- async def test_musicTitle(self):
- playlist = Playlist()
- song = Song(self.constants.MUSIC_TITLE_STRING, playlist, '')
- playlist.add_song(song)
-
- task = asyncio.create_task(self.downloader.download_song(song))
- await task
-
- self.assertFalse(song.problematic)
-
- def __verifySuccessfullyPreload(self, songs: List[Song]) -> bool:
- for song in songs:
- if song.title == None:
- print('Song failed to download')
- return False
-
- return True
-
-
-if __name__ == 'main':
- unittest.main()
diff --git a/Tests/VDownloaderTests.py b/Tests/VDownloaderTests.py
new file mode 100644
index 0000000..e64b2f1
--- /dev/null
+++ b/Tests/VDownloaderTests.py
@@ -0,0 +1,91 @@
+from typing import List
+from Tests.TestBase import VulkanTesterBase
+from Music.Playlist import Playlist
+from Music.Song import Song
+from asyncio import Task
+
+
+class VulkanDownloaderTest(VulkanTesterBase):
+ def __init__(self) -> None:
+ super().__init__()
+
+ def test_emptyString(self) -> bool:
+ musicsList = self._runner.run_coroutine(self._downloader.extract_info(''))
+
+ if musicsList == []:
+ return True
+ else:
+ return False
+
+ def test_YoutubeMusicUrl(self) -> bool:
+ musicsList = self._runner.run_coroutine(self._searcher.search(self._constants.YT_MUSIC_URL))
+
+ if len(musicsList) > 0:
+ print(musicsList[0])
+ return True
+ else:
+ return False
+
+ def test_YoutubeChannelPlaylist(self) -> None:
+ # Search the link to determine names
+ musicsList = self._runner.run_coroutine(
+ self._searcher.search(self._constants.YT_CHANNEL_PLAYLIST_URL))
+
+ if len(musicsList) == 0:
+ return False
+
+ # Create and store songs in list
+ playlist = Playlist()
+ songsList: List[Song] = []
+ for info in musicsList:
+ song = Song(identifier=info, playlist=playlist, requester='')
+ playlist.add_song(song)
+ songsList.append(song)
+
+ # Create a list of coroutines without waiting for them
+ tasks: List[Task] = []
+ for song in songsList:
+ tasks.append(self._downloader.download_song(song))
+
+ # Send for runner to execute them concurrently
+ self._runner.run_coroutines_list(tasks)
+
+ for song in songsList:
+ if song.problematic or song.title == None:
+ return False
+
+ return True
+
+ def test_YoutubeMixPlaylist(self) -> None:
+ # Search the link to determine names
+ music = self._runner.run_coroutine(
+ self._searcher.search(self._constants.YT_MIX_URL))
+
+ # Musics from Mix should download only the first music
+ if len(music) == 1:
+ return True
+ else:
+ return False
+
+ def test_musicTitle(self):
+ playlist = Playlist()
+ song = Song(self._constants.MUSIC_TITLE_STRING, playlist, '')
+ playlist.add_song(song)
+
+ self._runner.run_coroutine(self._downloader.download_song(song))
+
+ if song.problematic:
+ return False
+ else:
+ print(song.title)
+ return True
+
+ def test_YoutubePersonalPlaylist(self) -> None:
+ musics = self._runner.run_coroutine(
+ self._searcher.search(self._constants.YT_PERSONAL_PLAYLIST_URL))
+
+ if len(musics) > 0:
+ print(musics)
+ return True
+ else:
+ return False
diff --git a/Tests/__init__.py b/Tests/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/requirements.txt b/requirements.txt
index cc7481c..2fe4355 100644
Binary files a/requirements.txt and b/requirements.txt differ
diff --git a/run_tests.py b/run_tests.py
new file mode 100644
index 0000000..430eeb5
--- /dev/null
+++ b/run_tests.py
@@ -0,0 +1,5 @@
+from Tests.VDownloaderTests import VulkanDownloaderTest
+
+
+tester = VulkanDownloaderTest()
+tester.run()