Now using own test module to test vulkan downloader methods

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: <br>
+`python run_tests.py`<br>
+
 ## 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.

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'

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)

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)

Tests/TestsDownload.py

@@ -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()

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

run_tests.py

@@ -0,0 +1,5 @@
+from Tests.VDownloaderTests import VulkanDownloaderTest
+
+
+tester = VulkanDownloaderTest()
+tester.run()