DX-2686 Add Media Integration Tests (#95)

* Write Media API Integration Test

Added FIxtures

* Update test to conform to binary format definition

* PEP8 format

* Remove unused libraries

* Add Media Integration Tests

* Utilize a steps function to run tests in a specific order

https://stackoverflow.com/questions/5387299/python-unittest-testcase-execution-order/5387956#5387956

* Test Cleanup

* Update mediaId

github actions runId is not unique

* wrong env

* Address PR Comments

Author: ajrice6713

.github/workflows/test.yaml

@@ -43,6 +43,7 @@ jobs:
         ATT_NUMBER: ${{ secrets.ATT_NUMBER }}
         T_MOBILE_NUMBER: ${{ secrets.T_MOBILE_NUMBER }}
         BASE_CALLBACK_URL: ${{ secrets.BASE_CALLBACK_URL }}
+        PYTHON_VERSION: ${{ matrix.python-version }}
       run: pytest
 
     - name: Notify Slack of Failures

.gitignore

@@ -6,7 +6,7 @@ __pycache__/
 # C extensions
 *.so
 
-# MacOS Files
+# MacOS Files 
 .DS_Store
 
 # Distribution / packaging
@@ -67,3 +67,9 @@ target/
 
 #Ipython Notebook
 .ipynb_checkpoints
+
+# VS Code
+.vscode
+
+# Test Fixtures
+test/fixtures/

test/fixtures/python_cat.jpeg

@@ -0,0 +1,134 @@
+"""
+Integration test for Bandwidth's Media API
+"""
+
+import os
+import filecmp
+import unittest
+
+import bandwidth
+from bandwidth.api import media_api
+from bandwidth.model.media import Media
+from bandwidth.exceptions import ApiException, NotFoundException
+
+
+class TestMedia(unittest.TestCase):
+    """Media API integration Test
+    """
+
+    def setUp(self) -> None:
+        configuration = bandwidth.Configuration(
+            username=os.environ['BW_USERNAME'],
+            password=os.environ['BW_PASSWORD']
+        )
+        api_client = bandwidth.ApiClient(configuration)
+        self.api_instance = media_api.MediaApi(api_client)
+        self.account_id = os.environ['BW_ACCOUNT_ID']
+        self.media_path = "./test/fixtures/"
+        self.media_file = "python_cat.jpeg"
+        self.media_id = os.environ['PYTHON_VERSION'] + "_" + os.environ['RUNNER_OS'] + "_" + os.environ['GITHUB_RUN_ID'] + "_" + self.media_file
+        self.download_file_path = "cat_download.jpeg"
+
+        self.original_file = open(self.media_path + self.media_file, "rb")
+
+    def uploadMedia(self) -> None:
+        """Test uploading media
+        """
+        media_id = self.media_id
+        content_type = "image/jpeg"
+        cache_control = "no-cache"
+
+        api_response_with_http_info = self.api_instance.upload_media(
+            account_id=self.account_id,
+            media_id=media_id,
+            body=self.original_file,
+            content_type=content_type,
+            cache_control=cache_control,
+            _return_http_data_only=False
+        )
+
+        self.assertEqual(api_response_with_http_info[1], 204)
+
+        # reopen the media file
+        # the client automatically closes any files passed into request bodies
+        reopened_file = open(self.media_path + self.media_file, "rb")
+
+        # returns void
+        self.api_instance.upload_media(
+            account_id=self.account_id,
+            media_id=media_id,
+            body=reopened_file,
+            content_type=content_type,
+            cache_control=cache_control,
+            _return_http_data_only=False
+        )
+
+    def listMedia(self) -> None:
+        """Test listing all media on the account
+        """
+        api_response_with_http_info = self.api_instance.list_media(
+            self.account_id, _return_http_data_only=False)
+
+        self.assertEqual(api_response_with_http_info[1], 200)
+
+        api_response = self.api_instance.list_media(self.account_id)
+
+        self.assertIs(type(api_response[0]), Media)
+        pass
+
+    def getMedia(self) -> None:
+        """Test downloading the media we uploaded in step 1
+        """
+        api_response_with_http_info = self.api_instance.get_media(
+            self.account_id, self.media_id, _return_http_data_only=False)
+
+        self.assertEqual(api_response_with_http_info[1], 200)
+
+        api_response = self.api_instance.get_media(
+            self.account_id, self.media_id, _preload_content=False)
+
+        with open(self.media_path + self.download_file_path, "wb") as download_file:
+            download_file.write(api_response.data)
+
+        self.assertTrue(filecmp.cmp(self.media_path + self.media_file,
+                        self.media_path + self.download_file_path))
+        download_file.close()
+
+    def deleteMedia(self) -> None:
+        """Test deleting the media that was uploaded in step 1
+        """
+        api_response_with_http_info = self.api_instance.delete_media(
+            self.account_id, self.media_id, _return_http_data_only=False)
+
+        self.assertEqual(api_response_with_http_info[1], 204)
+
+        # returns void
+        self.api_instance.delete_media(self.account_id, self.media_id)
+    
+    def _steps(self) -> None:
+        call_order = ['uploadMedia', 'listMedia', 'getMedia', 'deleteMedia']
+        for name in call_order: 
+            yield name, getattr(self, name)
+
+    def test_steps(self) -> None:
+        """Test each function from _steps.call_order in specified order
+        """
+        for name, step in self._steps():
+            try:
+                step()
+            except ApiException as e:
+                self.fail("{} failed ({}: {})".format(step, type(e), e))
+
+    @unittest.skip("API does not support url encoded characters in path")
+    def testGetMediaWithBandwidthId(self) -> None:
+        # use a nonexistent mediaId - results in a 404
+        media_id = "abcd1234-e5f6-1111-2222-3456ghi7890/image123456.jpg"
+
+        with self.assertRaises(NotFoundException) as context:
+            api_response = self.api_instance.get_media(
+                self.account_id,
+                media_id,
+                _preload_content=False
+            )
+
+        self.assertEqual(context.exception.status, 404)