Skip to content
16 changes: 15 additions & 1 deletion amber/src/main/python/pytexera/storage/dataset_file_document.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,21 @@ def get_presigned_url(self) -> str:
f"Failed to get presigned URL: {response.status_code} {response.text}"
)

return response.json().get("presignedUrl")
try:
payload = response.json()
except ValueError as e:
raise RuntimeError(
f"Failed to get presigned URL: invalid JSON response: {response.text}"
) from e

presigned_url = payload.get("presignedUrl")
if not isinstance(presigned_url, str) or not presigned_url:
raise RuntimeError(
f"Failed to get presigned URL: 'presignedUrl' missing from "
f"response: {response.text}"
)

return presigned_url

def read_file(self) -> io.BytesIO:
"""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -137,13 +137,37 @@ def test_raises_runtime_error_with_status_and_body_on_failure(self, monkeypatch)
with pytest.raises(RuntimeError, match=r"403.*forbidden"):
doc.get_presigned_url()

def test_returns_none_when_response_body_lacks_presigned_url_key(self, monkeypatch):
# Pins current behavior: a 200 with no "presignedUrl" key yields None
# rather than raising. read_file() will then call requests.get(None).
def test_raises_when_response_body_lacks_presigned_url_key(self, monkeypatch):
doc = self._make_doc(monkeypatch)
with patch("pytexera.storage.dataset_file_document.requests.get") as mock_get:
mock_get.return_value = make_response(200, body={"other": "value"})
assert doc.get_presigned_url() is None
with pytest.raises(RuntimeError, match="'presignedUrl' missing"):
doc.get_presigned_url()

def test_raises_when_response_body_is_not_valid_json(self, monkeypatch):
doc = self._make_doc(monkeypatch)
with patch("pytexera.storage.dataset_file_document.requests.get") as mock_get:
response = MagicMock()
response.status_code = 200
response.json.side_effect = ValueError("Expecting value")
response.text = "<html>not json</html>"
mock_get.return_value = response
with pytest.raises(RuntimeError, match="invalid JSON response"):
doc.get_presigned_url()

def test_raises_when_presigned_url_is_empty_string(self, monkeypatch):
doc = self._make_doc(monkeypatch)
with patch("pytexera.storage.dataset_file_document.requests.get") as mock_get:
mock_get.return_value = make_response(200, body={"presignedUrl": ""})
with pytest.raises(RuntimeError, match="'presignedUrl' missing"):
doc.get_presigned_url()

def test_raises_when_presigned_url_is_not_a_string(self, monkeypatch):
doc = self._make_doc(monkeypatch)
with patch("pytexera.storage.dataset_file_document.requests.get") as mock_get:
mock_get.return_value = make_response(200, body={"presignedUrl": None})
with pytest.raises(RuntimeError, match="'presignedUrl' missing"):
doc.get_presigned_url()


class TestReadFile:
Expand Down
Loading