diff --git a/src/osw/controller/file/memory.py b/src/osw/controller/file/memory.py index 26e5de7..5bd9658 100644 --- a/src/osw/controller/file/memory.py +++ b/src/osw/controller/file/memory.py @@ -1,27 +1,26 @@ import shutil -from io import StringIO +from io import BytesIO from typing import IO, Any, Dict, List, Optional +from pydantic.v1 import Field + from osw.controller.file.base import FileController from osw.core import model class InMemoryController(FileController, model.LocalFile): - """File controller for local files""" + """File controller for in-memory streams""" label: Optional[List[model.Label]] = [model.Label(text="Unnamed stream")] """the label of the stream, e.g., the name of the file the stream originates from. Defaults to 'Unnamed stream'.""" - stream: IO - """the stream to the file""" + stream: Any = Field(default_factory=BytesIO) + """the stream to the file, any file-like object. Defaults to an empty + binary buffer. Byte-oriented, to match the get/put counterparts.""" class Config: arbitrary_types_allowed = True - def __init__(self, **kwargs): - self.stream = StringIO() - super().__init__(**kwargs) - def get(self) -> IO: return self.stream diff --git a/src/osw/express.py b/src/osw/express.py index a865a82..a54aa72 100644 --- a/src/osw/express.py +++ b/src/osw/express.py @@ -231,7 +231,7 @@ def download_file( def upload_file( self, - source: Union["LocalFileController", "WikiFileController", str, Path], + source: Union["LocalFileController", "WikiFileController", str, Path, IO], url_or_title: Optional[str] = None, overwrite: OVERWRITE_CLASS_OPTIONS = OverwriteOptions.true, delete_after_use: bool = False, @@ -247,7 +247,7 @@ def upload_file( ---------- source The source file to upload. Can be a LocalFileController, WikiFileController, - str or Path. + str, Path or an open file-like object. url_or_title The URL or full page title of the WikiFile page to upload the file to. Used to overwrite autogenerated full page title on the target domain. If it is @@ -635,7 +635,9 @@ def __init__( data["path"] = Path(source) data["source"] = Path(source) data["source_file_controller"] = LocalFileController(path=data.get("path")) - elif isinstance(source, IO): + # duck-typed: typing.IO is not runtime-checkable, isinstance(BytesIO(), IO) + # is False, so an explicit isinstance check would never match a stream + elif hasattr(source, "read"): data["source_file_controller"] = InMemoryController(stream=source) else: raise ValueError( @@ -736,7 +738,7 @@ def __init__( def osw_upload_file( - source: Union[LocalFileController, WikiFileController, str, Path], + source: Union[LocalFileController, WikiFileController, str, Path, IO], url_or_title: Optional[str] = None, overwrite: OVERWRITE_CLASS_OPTIONS = OverwriteOptions.true, delete_after_use: bool = False, @@ -756,7 +758,7 @@ def osw_upload_file( ---------- source The source file to upload. Can be a LocalFileController, WikiFileController, - str or Path. + str, Path or an open file-like object. url_or_title The URL or full page title of the WikiFile page to upload the file to. Used to overwrite autogenerated full page title on the target domain. If it is diff --git a/tests/test_in_memory_upload.py b/tests/test_in_memory_upload.py new file mode 100644 index 0000000..9713c06 --- /dev/null +++ b/tests/test_in_memory_upload.py @@ -0,0 +1,76 @@ +"""Unit tests for uploading a file from an in-memory stream (issue #140). + +Fully offline: no network, no live wiki. The upload path is cut short by +replacing ``WikiFileController.from_other`` with a stub that records the +source controller it was handed, so the dispatch in ``UploadFileResult`` can +be checked without touching a wiki. +""" + +from __future__ import annotations + +from io import BytesIO +from unittest.mock import MagicMock + +import pytest + +import osw.express +from osw.controller.file.memory import InMemoryController + + +def test_controller_accepts_a_caller_supplied_stream(): + stream = BytesIO(b"payload") + controller = InMemoryController(stream=stream) + assert controller.get() is stream + assert controller.get().read() == b"payload" + + +def test_controller_defaults_to_an_empty_binary_buffer(): + controller = InMemoryController() + assert isinstance(controller.stream, BytesIO) + assert controller.stream.getvalue() == b"" + + +def test_controller_put_copies_into_the_stream(): + controller = InMemoryController() + controller.put(BytesIO(b"payload")) + assert controller.stream.getvalue() == b"payload" + + +def test_upload_wraps_a_stream_in_an_in_memory_controller(monkeypatch): + """A BytesIO must reach WikiFileController as an InMemoryController. + + Guards the duck-typed source check in UploadFileResult.__init__: an + ``isinstance(source, IO)`` test never matches, because typing.IO is not + runtime-checkable. + """ + stream = BytesIO(b"payload") + seen = {} + + class _StopBeforeUpload(Exception): + pass + + def _fake_from_other(other, osw, **data): + seen["source_file_controller"] = other + raise _StopBeforeUpload + + monkeypatch.setattr( + osw.express.WikiFileController, + "from_other", + staticmethod(_fake_from_other), + ) + + with pytest.raises(_StopBeforeUpload): + osw.express.UploadFileResult( + source=stream, + osw_express=MagicMock(), + target_fpt="File:Test.bin", + ) + + controller = seen["source_file_controller"] + assert isinstance(controller, InMemoryController) + assert controller.get() is stream + + +def test_upload_rejects_a_source_that_is_not_file_like(): + with pytest.raises(ValueError, match="must be a LocalFileController"): + osw.express.UploadFileResult(source=object(), osw_express=MagicMock())