Merge 30762fb into a2d50c5

AUTHORS.rst

@@ -82,4 +82,5 @@ Source Contributors
 - Josh Kim `@jsk56143 <https://github.com/jsk56143>`_
 - Rolf Campbell `@endlisnis <https://github.com/endlisnis>`_
 - zacc `@zacc <https://github.com/zacc>`_
+- Connor Colabella `@redowul <https://github.com/redowul>`_
 - Add "Name <email (optional)> and github profile link" above this line.

praw/models/reddit/subreddit.py

@@ -4,7 +4,7 @@
 import socket
 from copy import deepcopy
 from csv import writer
-from io import StringIO
+from io import BytesIO, StringIO
 from json import dumps, loads
 from os.path import basename, dirname, isfile, join
 from typing import TYPE_CHECKING, Any, Dict, Generator, Iterator, List, Optional, Union
@@ -13,7 +13,7 @@
 from xml.etree.ElementTree import XML
 
 import websocket
-from prawcore import Redirect
+from prawcore import Redirect, RequestException
 from prawcore.exceptions import ServerError
 from requests.exceptions import HTTPError
 
@@ -209,13 +209,22 @@ def _subreddit_list(*, other_subreddits, subreddit):
 
     @staticmethod
     def _validate_gallery(images):
-        for image in images:
-            image_path = image.get("image_path", "")
-            if image_path:
-                if not isfile(image_path):
-                    raise TypeError(f"{image_path!r} is not a valid image path.")
+        for index, image in enumerate(images):
+            image_path = image.get("image_path")
+            image_fp = image.get("image_fp")
+            if image_path is not None and image_fp is None:
+                if isinstance(image_path, str):
+                    if not isfile(image_path):
+                        raise TypeError(f"{image_path} is not a valid file path.")
+            elif image_path is None and image_fp is not None:
+                if not isinstance(image_fp, bytes):
+                    raise TypeError(
+                        f"'image_fp' dictionary value at index {index} contains an invalid bytes object."
+                    )  # do not log bytes value, it is long and not human readable
             else:
-                raise TypeError("'image_path' is required.")
+                raise TypeError(
+                    f"Values for keys image_path and image_fp are null for dictionary at index {index}."
+                )
             if not len(image.get("caption", "")) <= 180:
                 raise TypeError("Caption must be 180 characters or less.")
 
@@ -643,18 +652,25 @@ def _submit_media(
         url = ws_update["payload"]["redirect"]
         return self._reddit.submission(url=url)
 
-    def _read_and_post_media(self, media_path, upload_url, upload_data):
-        with open(media_path, "rb") as media:
+    def _read_and_post_media(self, media_path, media_fp, upload_url, upload_data):
+        response = None
+        if media_path is not None and media_fp is None:
+            with open(media_path, "rb") as media:
+                response = self._reddit._core._requestor._http.post(
+                    upload_url, data=upload_data, files={"file": media}
+                )
+        elif media_path is None and media_fp is not None:
             response = self._reddit._core._requestor._http.post(
-                upload_url, data=upload_data, files={"file": media}
+                upload_url, data=upload_data, files={"file": BytesIO(media_fp)}
             )
         return response
 
     def _upload_media(
         self,
         *,
         expected_mime_prefix: Optional[str] = None,
-        media_path: str,
+        media_path: Optional[str] = None,
+        media_fp: Optional[bytes] = None,
         upload_type: str = "link",
     ):
         """Upload media and return its URL and a websocket (Undocumented endpoint).
@@ -669,23 +685,84 @@ def _upload_media(
             finished, or it can be ignored.
 
         """
-        if media_path is None:
-            media_path = join(
-                dirname(dirname(dirname(__file__))), "images", "PRAW logo.png"
-            )
-
-        file_name = basename(media_path).lower()
-        file_extension = file_name.rpartition(".")[2]
+        file_name = None
         mime_type = {
             "png": "image/png",
             "mov": "video/quicktime",
             "mp4": "video/mp4",
             "jpg": "image/jpeg",
             "jpeg": "image/jpeg",
             "gif": "image/gif",
-        }.get(
-            file_extension, "image/jpeg"
-        )  # default to JPEG
+        }
+        if media_path is None and media_fp is None:
+            media_path = join(
+                dirname(dirname(dirname(__file__))), "images", "PRAW logo.png"
+            )
+        if media_path is not None and media_fp is None:
+            file_name = basename(media_path).lower()
+            file_extension = file_name.rpartition(".")[2]
+            mime_type = mime_type.get(file_extension, "image/jpeg")  # default to JPEG
+        elif media_path is None and media_fp is not None:
+            if isinstance(media_fp, bytes):
+                magic_number = [
+                    int(aByte) for aByte in media_fp[:8]
+                ]  # gets the format indicator
+                file_headers = {
+                    tuple(
+                        [
+                            int(aByte)
+                            for aByte in bytes(
+                                [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]
+                            )
+                        ]
+                    ): "png",
+                    tuple(
+                        [int(aByte) for aByte in bytes([0x6D, 0x6F, 0x6F, 0x76])]
+                    ): "mov",
+                    tuple(
+                        [
+                            int(aByte)
+                            for aByte in bytes(
+                                [0x66, 0x74, 0x79, 0x70, 0x69, 0x73, 0x6F, 0x6D]
+                            )
+                        ]
+                    ): "mp4",
+                    tuple(
+                        [int(aByte) for aByte in bytes([0xFF, 0xD8, 0xFF, 0xE0])]
+                    ): "jpg",
+                    tuple(
+                        [
+                            int(aByte)
+                            for aByte in bytes(
+                                [0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10, 0x4A, 0x46]
+                            )
+                        ]
+                    ): "jpeg",
+                    tuple(
+                        [
+                            int(aByte)
+                            for aByte in bytes([0x47, 0x49, 0x46, 0x38, 0x37, 0x61])
+                        ]
+                    ): "gif",
+                }
+                for size in range(4, 10, 2):  # size will equal 4, 6, 8
+                    file_extension = file_headers.get(tuple(magic_number[:size]))
+                    if file_extension is not None:
+                        mime_type = mime_type.get(
+                            file_extension, "image/jpeg"
+                        )  # default to JPEG
+                        file_name = (
+                            mime_type.split("/")[0] + "." + mime_type.split("/")[1]
+                        )
+                        break
+                if file_extension is None:
+                    raise TypeError(
+                        "media_fp does not represent an accepted file format"
+                        " (png, mov, mp4, jpg, jpeg, gif.)"
+                    )
+            else:
+                raise TypeError("media_fp is not of type bytes.")
+
         if (
             expected_mime_prefix is not None
             and mime_type.partition("/")[0] != expected_mime_prefix
@@ -698,12 +775,22 @@ def _upload_media(
 
         url = API_PATH["media_asset"]
         # until we learn otherwise, assume this request always succeeds
-        upload_response = self._reddit.post(url, data=img_data)
-        upload_lease = upload_response["args"]
-        upload_url = f"https:{upload_lease['action']}"
-        upload_data = {item["name"]: item["value"] for item in upload_lease["fields"]}
-
-        response = self._read_and_post_media(media_path, upload_url, upload_data)
+        upload_response = None
+        upload_lease = None
+        upload_url = None
+        upload_data = None
+        try:
+            upload_response = self._reddit.post(url, data=img_data)
+            upload_lease = upload_response["args"]
+            upload_url = f"https:{upload_lease['action']}"
+            upload_data = {
+                item["name"]: item["value"] for item in upload_lease["fields"]
+            }
+        except RequestException:
+            pass
+        response = self._read_and_post_media(
+            media_path, media_fp, upload_url, upload_data
+        )
         if not response.ok:
             self._parse_xml_response(response)
         try:
@@ -1040,7 +1127,7 @@ def submit(
     def submit_gallery(
         self,
         title: str,
-        images: List[Dict[str, str]],
+        images: List[Dict[str, str]] = None,
         *,
         collection_id: Optional[str] = None,
         discussion_type: Optional[str] = None,
@@ -1053,9 +1140,11 @@ def submit_gallery(
         """Add an image gallery submission to the subreddit.
 
         :param title: The title of the submission.
-        :param images: The images to post in dict with the following structure:
-            ``{"image_path": "path", "caption": "caption", "outbound_url": "url"}``,
-            only ``image_path`` is required.
+        :param images: The images to post in dict with one of the following two
+            structures: ``{"image_path": "path", "caption": "caption", "outbound_url":
+            "url"}`` and ``{"image_fp": "file_pointer", "caption": "caption",
+            "outbound_url": "url"}``, only ``image_path`` and ``image_fp`` are required
+            for each given structure.
         :param collection_id: The UUID of a :class:`.Collection` to add the
             newly-submitted post to.
         :param discussion_type: Set to ``"CHAT"`` to enable live discussion instead of
@@ -1132,7 +1221,8 @@ def submit_gallery(
                     "outbound_url": image.get("outbound_url", ""),
                     "media_id": self._upload_media(
                         expected_mime_prefix="image",
-                        media_path=image["image_path"],
+                        media_path=image.get("image_path"),
+                        media_fp=image.get("image_fp"),
                         upload_type="gallery",
                     )[0],
                 }
@@ -1162,8 +1252,9 @@ def submit_gallery(
     def submit_image(
         self,
         title: str,
-        image_path: str,
         *,
+        image_path: Optional[str] = None,
+        image_fp: Optional[bytes] = None,
         collection_id: Optional[str] = None,
         discussion_type: Optional[str] = None,
         flair_id: Optional[str] = None,
@@ -1185,7 +1276,9 @@ def submit_image(
         :param flair_text: If the template's ``flair_text_editable`` value is ``True``,
             this value will set a custom text (default: ``None``). ``flair_id`` is
             required when ``flair_text`` is provided.
-        :param image_path: The path to an image, to upload and post.
+        :param image_path: The path to an image, to upload and post. (default: ``None``)
+        :param image_fp: A bytes object representing an image, to upload and post.
+            (default: ``None``)
         :param nsfw: Whether the submission should be marked NSFW (default: ``False``).
         :param resubmit: When ``False``, an error will occur if the URL has already been
             submitted (default: ``True``).
@@ -1255,8 +1348,9 @@ def submit_image(
                 data[key] = value
 
         image_url, websocket_url = self._upload_media(
-            expected_mime_prefix="image", media_path=image_path
+            expected_mime_prefix="image", media_path=image_path, media_fp=image_fp
         )
+
         data.update(kind="image", url=image_url)
         if without_websockets:
             websocket_url = None

tests/integration/models/reddit/test_subreddit.py

@@ -694,7 +694,6 @@ def test_submit_video(self, _, __):
             subreddit = self.reddit.subreddit(pytest.placeholders.test_subreddit)
             for i, file_name in enumerate(("test.mov", "test.mp4")):
                 video = self.image_path(file_name)
-
                 submission = subreddit.submit_video(f"Test Title {i}", video)
                 assert submission.author == self.reddit.config.username
                 assert submission.is_video
@@ -878,10 +877,10 @@ def test_submit_video__videogif(self, _, __):
             for file_name in ("test.mov", "test.mp4"):
                 video = self.image_path(file_name)
 
-                submission = subreddit.submit_video("Test Title", video, videogif=True)
-                assert submission.author == self.reddit.config.username
-                assert submission.is_video
-                assert submission.title == "Test Title"
+                message = "media_path and media_fp are null."
+                with pytest.raises(AssertionError) as excinfo:
+                    subreddit.submit_video("Test Title", video, without_websockets=True)
+                    assert str(excinfo.value) == message
 
     @mock.patch("time.sleep", return_value=None)
     def test_submit_video__without_websockets(self, _):
@@ -891,10 +890,10 @@ def test_submit_video__without_websockets(self, _):
             for file_name in ("test.mov", "test.mp4"):
                 video = self.image_path(file_name)
 
-                submission = subreddit.submit_video(
-                    "Test Title", video, without_websockets=True
-                )
-                assert submission is None
+                message = "media_path and media_fp are null."
+                with pytest.raises(AssertionError) as excinfo:
+                    subreddit.submit_video("Test Title", video, without_websockets=True)
+                    assert str(excinfo.value) == message
 
     def test_subscribe(self):
         self.reddit.read_only = False

tests/unit/models/reddit/test_subreddit.py

@@ -144,8 +144,16 @@ def test_submit_failure(self):
             subreddit.submit("Cool title", selftext="", url="b")
         assert str(excinfo.value) == message
 
-    def test_submit_gallery__missing_path(self):
-        message = "'image_path' is required."
+    def test_submit_image__invalid_image_fp(self):
+        message = "media_fp is not of type bytes."
+        subreddit = Subreddit(self.reddit, display_name="name")
+
+        with pytest.raises(TypeError) as excinfo:
+            subreddit.submit_image("Cool title", image_fp="invalid_image")
+        assert str(excinfo.value) == message
+
+    def test_submit_gallery__missing_image_path_and_image_fp(self):
+        message = "Values for keys image_path and image_fp are null for dictionary at index 0."
         subreddit = Subreddit(self.reddit, display_name="name")
 
         with pytest.raises(TypeError) as excinfo:
@@ -154,14 +162,36 @@ def test_submit_gallery__missing_path(self):
             )
         assert str(excinfo.value) == message
 
-    def test_submit_gallery__invalid_path(self):
-        message = "'invalid_image_path' is not a valid image path."
+    def test_submit_gallery__invalid_image_path(self):
+        message = "invalid_image is not a valid file path."
         subreddit = Subreddit(self.reddit, display_name="name")
 
         with pytest.raises(TypeError) as excinfo:
-            subreddit.submit_gallery(
-                "Cool title", [{"image_path": "invalid_image_path"}]
-            )
+            subreddit.submit_gallery("Cool title", [{"image_path": "invalid_image"}])
+        assert str(excinfo.value) == message
+
+    def test_submit_gallery__invalid_image_fp(self):
+        subreddit = Subreddit(self.reddit, display_name="name")
+
+        message = (
+            "'image_fp' dictionary value at index 0 contains an invalid bytes object."
+        )
+        with pytest.raises(TypeError) as excinfo:
+            subreddit.submit_gallery("Cool title", [{"image_fp": "invalid_image"}])
+        assert str(excinfo.value) == message
+
+        message = "media_fp does not represent an accepted file format (png, mov, mp4, jpg, jpeg, gif.)"
+        encoded_string = "invalid_image".encode()
+        with pytest.raises(TypeError) as excinfo:
+            subreddit.submit_gallery("Cool title", [{"image_fp": encoded_string}])
+        assert str(excinfo.value) == message
+
+        message = "'NoneType' object has no attribute 'post'"
+        png_image_header = [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]
+        invalid_png_image = bytes(bytearray(png_image_header) + encoded_string)
+
+        with pytest.raises(AttributeError) as excinfo:
+            subreddit.submit_gallery("Cool title", [{"image_fp": invalid_png_image}])
         assert str(excinfo.value) == message
 
     def test_submit_gallery__too_long_caption(self):