Mar 12, 2023 · Mar 4, 2023 · Mar 5, 2023
diff --git a/gitlab/_backends/requests_backend.py b/gitlab/_backends/requests_backend.py
 from __future__ import annotations

 from typing import Any, Dict, Optional, Tuple, TYPE_CHECKING, Union
 import dataclasses
 from typing import Any, Dict, Optional, TYPE_CHECKING, Union

 import requests
 from requests.structures import CaseInsensitiveDict
 from . import protocol


 @dataclasses.dataclass
 class SendData:
    content_type: str
    data: Optional[Union[Dict[str, Any], MultipartEncoder]] = None
    json: Optional[Union[Dict[str, Any], bytes]] = None

    def __post_init__(self) -> None:
        if self.json is not None and self.data is not None:
            raise ValueError(
                f"`json` and `data` are mutually exclusive. Only one can be set. "
                f"json={self.json!r}  data={self.data!r}"
            )


 class RequestsResponse(protocol.BackendResponse):
    def __init__(self, response: requests.Response) -> None:
        self._response: requests.Response = response
        files: Optional[Dict[str, Any]] = None,
        post_data: Optional[Union[Dict[str, Any], bytes]] = None,
        raw: bool = False,
    ) -> Tuple[
        Optional[Union[Dict[str, Any], bytes]],
        Optional[Union[Dict[str, Any], MultipartEncoder]],
        str,
    ]:
    ) -> SendData:
        if files:
            if post_data is None:
                post_data = {}
            else:
                # booleans does not exists for data (neither for MultipartEncoder):
                # cast to string int to avoid: 'bool' object has no attribute 'encode'
                # When creating a `MultipartEncoder` instance with data-types
                # which don't have an `encode` method it will cause an error:
                #       object has no attribute 'encode'
                # So convert common non-string types into strings.
                if TYPE_CHECKING:
                    assert isinstance(post_data, dict)
                for k, v in post_data.items():
                    if isinstance(v, bool):
                        post_data[k] = str(int(v))
                        v = int(v)
                    if isinstance(v, (complex, float, int)):
                        post_data[k] = str(v)
            post_data["file"] = files.get("file")
            post_data["avatar"] = files.get("avatar")

            data = MultipartEncoder(post_data)
            return(None,data,data.content_type)
            data = MultipartEncoder(fields=post_data)
            returnSendData(data=data, content_type=data.content_type)

        if raw and post_data:
            return(None,post_data, "application/octet-stream")
            returnSendData(data=post_data,content_type="application/octet-stream")

        return(post_data,None,"application/json")
        returnSendData(json=post_data,content_type="application/json")

    def http_request(
        self,
diff --git a/gitlab/client.py b/gitlab/client.py
            retry_transient_errors = self.retry_transient_errors

        # We need to deal with json vs. data when uploading files
        json, data, content_type = self._backend.prepare_send_data(
            files, post_data, raw
        )
        opts["headers"]["Content-type"] = content_type
        send_data = self._backend.prepare_send_data(files, post_data, raw)
        opts["headers"]["Content-type"] = send_data.content_type

        cur_retries = 0
        while True:
            try:
                result = self._backend.http_request(
                    method=verb,
                    url=url,
                    json=json,
                    data=data,
                    json=send_data.json,
                    data=send_data.data,
                    params=params,
                    timeout=timeout,
                    verify=verify,
diff --git a/gitlab/v4/objects/groups.py b/gitlab/v4/objects/groups.py
        file: BinaryIO,
        path: str,
        name: str,
        parent_id: Optional[str] = None,
        parent_id: Optional[Union[int,str]] = None,
        **kwargs: Any,
    ) -> Union[Dict[str, Any], requests.Response]:
        """Import a group from an archive file.
            A representation of the import status.
        """
        files = {"file": ("file.tar.gz", file, "application/octet-stream")}
        data = {"path": path, "name": name}
        data: Dict[str, Any] = {"path": path, "name": name}
        if parent_id is not None:
            data["parent_id"] = parent_id

diff --git a/tests/unit/_backends/__init__.py b/tests/unit/_backends/__init__.py
diff --git a/tests/unit/_backends/test_requests_backend.py b/tests/unit/_backends/test_requests_backend.py
 import pytest
 from requests_toolbelt.multipart.encoder import MultipartEncoder  # type: ignore

 from gitlab._backends import requests_backend


 class TestSendData:
    def test_senddata_json(self) -> None:
        result = requests_backend.SendData(
            json={"a": 1}, content_type="application/json"
        )
        assert result.data is None

    def test_senddata_data(self) -> None:
        result = requests_backend.SendData(
            data={"b": 2}, content_type="application/octet-stream"
        )
        assert result.json is None

    def test_senddata_json_and_data(self) -> None:
        with pytest.raises(ValueError, match=r"json={'a': 1}  data={'b': 2}"):
            requests_backend.SendData(
                json={"a": 1}, data={"b": 2}, content_type="application/json"
            )


 class TestRequestsBackend:
    @pytest.mark.parametrize(
        "test_data,expected",
        [
            (False, "0"),
            (True, "1"),
            ("12", "12"),
            (12, "12"),
            (12.0, "12.0"),
            (complex(-2, 7), "(-2+7j)"),
        ],
    )
    def test_prepare_send_data_non_strings(self, test_data, expected) -> None:
        assert isinstance(expected, str)
        files = {"file": ("file.tar.gz", "12345", "application/octet-stream")}
        post_data = {"test_data": test_data}

        result = requests_backend.RequestsBackend.prepare_send_data(
            files=files, post_data=post_data, raw=False
        )
        assert result.json is None
        assert result.content_type.startswith("multipart/form-data")
        assert isinstance(result.data, MultipartEncoder)
        assert isinstance(result.data.fields["test_data"], str)
        assert result.data.fields["test_data"] == expected
Original file line number	Diff line number	Diff line change
		@@ -1,6 +1,7 @@
		from __future__ import annotations

		from typing import Any, Dict, Optional, Tuple, TYPE_CHECKING, Union
		import dataclasses
		from typing import Any, Dict, Optional, TYPE_CHECKING, Union

		import requests
		from requests.structures import CaseInsensitiveDict
Expand All		@@ -9,6 +10,20 @@
		from . import protocol


		@dataclasses.dataclass
		class SendData:
		content_type: str
		data: Optional[Union[Dict[str, Any], MultipartEncoder]] = None
		json: Optional[Union[Dict[str, Any], bytes]] = None

		def __post_init__(self) -> None:
		if self.json is not None and self.data is not None:
		raise ValueError(
		f"`json` and `data` are mutually exclusive. Only one can be set. "
		f"json={self.json!r} data={self.data!r}"
		)


		class RequestsResponse(protocol.BackendResponse):
		def __init__(self, response: requests.Response) -> None:
		self._response: requests.Response = response
Expand DownExpand Up		@@ -50,32 +65,32 @@ def prepare_send_data(
		files: Optional[Dict[str, Any]] = None,
		post_data: Optional[Union[Dict[str, Any], bytes]] = None,
		raw: bool = False,
		) -> Tuple[
		Optional[Union[Dict[str, Any], bytes]],
		Optional[Union[Dict[str, Any], MultipartEncoder]],
		str,
		]:
		) -> SendData:
		if files:
		if post_data is None:
		post_data = {}
		else:
		# booleans does not exists for data (neither for MultipartEncoder):
		# cast to string int to avoid: 'bool' object has no attribute 'encode'
		# When creating a `MultipartEncoder` instance with data-types
		# which don't have an `encode` method it will cause an error:
		# object has no attribute 'encode'
		# So convert common non-string types into strings.
		if TYPE_CHECKING:
		assert isinstance(post_data, dict)
		for k, v in post_data.items():
		if isinstance(v, bool):
		post_data[k] = str(int(v))
		v = int(v)
		if isinstance(v, (complex, float, int)):
		post_data[k] = str(v)
		post_data["file"] = files.get("file")
		post_data["avatar"] = files.get("avatar")

		data = MultipartEncoder(post_data)
		return(None,data,data.content_type)
		data = MultipartEncoder(fields=post_data)
		returnSendData(data=data, content_type=data.content_type)

		if raw and post_data:
		return(None,post_data, "application/octet-stream")
		returnSendData(data=post_data,content_type="application/octet-stream")

		return(post_data,None,"application/json")
		returnSendData(json=post_data,content_type="application/json")

		def http_request(
		self,
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -716,19 +716,17 @@ def http_request(
		retry_transient_errors = self.retry_transient_errors

		# We need to deal with json vs. data when uploading files
		json, data, content_type = self._backend.prepare_send_data(
		files, post_data, raw
		)
		opts["headers"]["Content-type"] = content_type
		send_data = self._backend.prepare_send_data(files, post_data, raw)
		opts["headers"]["Content-type"] = send_data.content_type

		cur_retries = 0
		while True:
		try:
		result = self._backend.http_request(
		method=verb,
		url=url,
		json=json,
		data=data,
		json=send_data.json,
		data=send_data.data,
		params=params,
		timeout=timeout,
		verify=verify,
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -378,7 +378,7 @@ def import_group(
		file: BinaryIO,
		path: str,
		name: str,
		parent_id: Optional[str] = None,
		parent_id: Optional[Union[int,str]] = None,
		**kwargs: Any,
		) -> Union[Dict[str, Any], requests.Response]:
		"""Import a group from an archive file.
Expand All		@@ -399,7 +399,7 @@ def import_group(
		A representation of the import status.
		"""
		files = {"file": ("file.tar.gz", file, "application/octet-stream")}
		data = {"path": path, "name": name}
		data: Dict[str, Any] = {"path": path, "name": name}
		if parent_id is not None:
		data["parent_id"] = parent_id

Expand Down
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,51 @@
		import pytest
		from requests_toolbelt.multipart.encoder import MultipartEncoder # type: ignore

		from gitlab._backends import requests_backend


		class TestSendData:
		def test_senddata_json(self) -> None:
		result = requests_backend.SendData(
		json={"a": 1}, content_type="application/json"
		)
		assert result.data is None

		def test_senddata_data(self) -> None:
		result = requests_backend.SendData(
		data={"b": 2}, content_type="application/octet-stream"
		)
		assert result.json is None

		def test_senddata_json_and_data(self) -> None:
		with pytest.raises(ValueError, match=r"json={'a': 1} data={'b': 2}"):
		requests_backend.SendData(
		json={"a": 1}, data={"b": 2}, content_type="application/json"
		)


		class TestRequestsBackend:
		@pytest.mark.parametrize(
		"test_data,expected",
		[
		(False, "0"),
		(True, "1"),
		("12", "12"),
		(12, "12"),
		(12.0, "12.0"),
		(complex(-2, 7), "(-2+7j)"),
		],
		)
		def test_prepare_send_data_non_strings(self, test_data, expected) -> None:
		assert isinstance(expected, str)
		files = {"file": ("file.tar.gz", "12345", "application/octet-stream")}
		post_data = {"test_data": test_data}

		result = requests_backend.RequestsBackend.prepare_send_data(
		files=files, post_data=post_data, raw=False
		)
		assert result.json is None
		assert result.content_type.startswith("multipart/form-data")
		assert isinstance(result.data, MultipartEncoder)
		assert isinstance(result.data.fields["test_data"], str)
		assert result.data.fields["test_data"] == expected