diff --git a/docs/tree/getting_started.rst b/docs/tree/getting_started.rst index b98e4ff..90f56a0 100644 --- a/docs/tree/getting_started.rst +++ b/docs/tree/getting_started.rst @@ -72,17 +72,23 @@ The :class:`~.NucleiClient` can provide you with a list of `Nuclei` applications print(client.applications) +And can also fetch the available versions for an application for you: + +.. ipython:: python + + versions = print(client.get_versions(app="PileCore")) + And can also fetch the available endpoints for an application for you: .. ipython:: python - endpoints = print(client.get_endpoints("PileCore")) + endpoints = print(client.get_endpoints(app="PileCore", version="latest")) You can also check the applications to which you have full access: .. ipython:: python - endpoints = print(client.user_permissions) + permissions = print(client.user_permissions) If an application is not listed here, your usage of the app is limited. Check the documentation of the specific apps to see the limitations. diff --git a/requirements.txt b/requirements.txt index 24379e4..ebe467f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,25 +4,25 @@ # # pip-compile --extra=client --extra=docs --extra=lint --extra=test --output-file=requirements.txt pyproject.toml # -alabaster==0.7.13 +alabaster==0.7.16 # via sphinx asteroid-sphinx-theme==0.0.3 # via cems-nuclei (pyproject.toml) asttokens==2.4.1 # via stack-data -babel==2.13.1 +babel==2.15.0 # via sphinx -black==23.11.0 +black==24.4.2 # via cems-nuclei (pyproject.toml) -certifi==2023.11.17 +certifi==2024.7.4 # via requests charset-normalizer==3.3.2 # via requests click==8.1.7 # via black -coverage==6.5.0 +coverage[toml]==7.6.0 # via coveralls -coveralls==3.3.1 +coveralls==4.0.1 # via cems-nuclei (pyproject.toml) decorator==5.1.1 # via ipython @@ -33,72 +33,72 @@ docutils==0.18.1 # m2r2 # sphinx # sphinx-rtd-theme -exceptiongroup==1.2.0 +exceptiongroup==1.2.2 # via # ipython # pytest executing==2.0.1 # via stack-data -idna==3.4 +idna==3.7 # via requests imagesize==1.4.1 # via sphinx -importlib-metadata==6.8.0 +importlib-metadata==8.0.0 # via sphinx iniconfig==2.0.0 # via pytest -ipython==8.17.2 +ipython==8.18.1 # via cems-nuclei (pyproject.toml) -isort==5.12.0 +isort==5.13.2 # via cems-nuclei (pyproject.toml) jedi==0.19.1 # via ipython -jinja2==3.1.2 +jinja2==3.1.4 # via sphinx m2r2==0.3.2 # via cems-nuclei (pyproject.toml) -markupsafe==2.1.3 +markupsafe==2.1.5 # via jinja2 -matplotlib-inline==0.1.6 +matplotlib-inline==0.1.7 # via ipython mistune==0.8.4 # via m2r2 mypy-extensions==1.0.0 # via black -numpy==1.26.2 +numpy==1.26.4 # via cems-nuclei (pyproject.toml) -orjson==3.9.10 +orjson==3.10.6 # via cems-nuclei (pyproject.toml) -packaging==23.2 +packaging==24.1 # via # black # pytest # sphinx -parso==0.8.3 +parso==0.8.4 # via jedi -pathspec==0.11.2 +pathspec==0.12.1 # via black -pexpect==4.8.0 +pexpect==4.9.0 # via ipython -platformdirs==4.0.0 +platformdirs==4.2.2 # via black -pluggy==1.3.0 +pluggy==1.5.0 # via pytest -prompt-toolkit==3.0.41 +prompt-toolkit==3.0.47 # via ipython ptyprocess==0.7.0 # via pexpect pure-eval==0.2.2 # via stack-data -pygments==2.17.1 +pygments==2.18.0 # via # ipython # sphinx pyjwt==2.6.0 # via cems-nuclei (pyproject.toml) -pytest==7.4.3 +pytest==8.2.2 # via cems-nuclei (pyproject.toml) -requests==2.31.0 +requests==2.32.3 # via # cems-nuclei (pyproject.toml) # coveralls @@ -113,47 +113,43 @@ sphinx==6.1.3 # cems-nuclei (pyproject.toml) # sphinx-autodoc-typehints # sphinx-rtd-theme - # sphinxcontrib-applehelp - # sphinxcontrib-devhelp - # sphinxcontrib-htmlhelp # sphinxcontrib-jquery - # sphinxcontrib-qthelp - # sphinxcontrib-serializinghtml sphinx-autodoc-typehints==1.22 # via cems-nuclei (pyproject.toml) sphinx-rtd-theme==1.2.0 # via cems-nuclei (pyproject.toml) -sphinxcontrib-applehelp==1.0.7 +sphinxcontrib-applehelp==1.0.8 # via sphinx -sphinxcontrib-devhelp==1.0.5 +sphinxcontrib-devhelp==1.0.6 # via sphinx -sphinxcontrib-htmlhelp==2.0.4 +sphinxcontrib-htmlhelp==2.0.5 # via sphinx sphinxcontrib-jquery==4.1 # via sphinx-rtd-theme sphinxcontrib-jsmath==1.0.1 # via sphinx -sphinxcontrib-qthelp==1.0.6 +sphinxcontrib-qthelp==1.0.7 # via sphinx -sphinxcontrib-serializinghtml==1.1.9 +sphinxcontrib-serializinghtml==1.1.10 # via sphinx stack-data==0.6.3 # via ipython tomli==2.0.1 # via # black + # coverage # pytest -traitlets==5.13.0 +traitlets==5.14.3 # via # ipython # matplotlib-inline -typing-extensions==4.8.0 +typing-extensions==4.12.2 # via # black # ipython -urllib3==2.1.0 +urllib3==2.2.2 # via requests -wcwidth==0.2.11 +wcwidth==0.2.13 # via prompt-toolkit -zipp==3.17.0 +zipp==3.19.2 # via importlib-metadata diff --git a/src/nuclei/client/main.py b/src/nuclei/client/main.py index 9662ee8..39e5f06 100644 --- a/src/nuclei/client/main.py +++ b/src/nuclei/client/main.py @@ -6,7 +6,6 @@ from typing import Any, List, Literal, Optional, Union import jwt -import requests from nuclei import create_session @@ -23,10 +22,23 @@ ) ROUTING = { - "PileCore": "https://crux-nuclei.com/api/pilecore/v2", - "VibraCore": "https://crux-nuclei.com/api/vibracore/v2", - "CPT Core": "https://crux-nuclei.com/api/cptcore/v1", - "ShallowCore": "https://crux-nuclei.com/api/shallowcore/v1", + "PileCore": { + "v2": "https://crux-nuclei.com/api/pilecore/v2", + "v3": "https://crux-nuclei.com/api/pilecore/v3", + "latest": "https://crux-nuclei.com/api/pilecore/v3", + }, + "VibraCore": { + "v2": "https://crux-nuclei.com/api/vibracore/v2", + "latest": "https://crux-nuclei.com/api/vibracore/v2", + }, + "CPT Core": { + "v1": "https://crux-nuclei.com/api/cptcore/v1", + "latest": "https://crux-nuclei.com/api/cptcore/v1", + }, + "ShallowCore": { + "v1": "https://crux-nuclei.com/api/shallowcore/v1", + "latest": "https://crux-nuclei.com/api/shallowcore/v1", + }, } DEFAULT_REQUEST_TIMEOUT = 5 @@ -60,14 +72,17 @@ def __init__(self) -> None: # set default timeout self.timeout = DEFAULT_REQUEST_TIMEOUT - def get_url(self, app: str) -> str: + def get_url(self, app: str, version: str = "latest") -> str: """ Get API's url Parameters ---------- - app : str - Application name + app: str + Name of the API. Call `applications` to obtain a list with all applications. + version: str, optional + default is latest. + API version used. Call `get_versions` to obtain a list with all version of a specific application. Returns ------- @@ -76,20 +91,19 @@ def get_url(self, app: str) -> str: Raises ------- TypeError: - Wrong type for `app` argument + Wrong type for `app` or `version` argument ValueError: - Wrong value for `app` argument + Wrong value for `app` or `version` argument """ - if not isinstance(app, str): + if not isinstance(version, str): raise TypeError( - f"Expected positional argument `app` to be of type , but got type: {type(app)}" + f"Expected positional argument `version` to be of type , but got type: {type(version)}" ) - - if app in self.applications: - return self.routing[app] - raise ValueError( - f"Application not available, please select one of the following valid applications {self.applications}" - ) + if version not in self.get_versions(app): + raise ValueError( + f"Application version not available, please select one of the following valid versions {self.get_versions(app)}" + ) + return self.routing[app][version] @property def user_permissions(self) -> List[str | None]: @@ -119,24 +133,22 @@ def applications(self) -> List[str]: """ return list(self.routing.keys()) - @lru_cache(16) - def _get_app_specification(self, app: str) -> dict: + def get_versions(self, app: str) -> List[str]: """ - Private methode to get the JSON schema of the API documentation. + Provide available API's versions in the Nuclei landscape. Parameters ---------- - app : str - Name of the API. + app: str + Name of the API. Call `applications` to obtain a list with all applications. Returns ------- - dict + out : list[str] + Versions of the API. Raises ------- - ConnectionError: - Application not available TypeError: Wrong type for `app` argument ValueError: @@ -146,9 +158,40 @@ def _get_app_specification(self, app: str) -> dict: raise TypeError( f"Expected positional argument `app` to be of type , but got type: {type(app)}" ) + if app not in self.applications: + raise ValueError( + f"Application not available, please select one of the following valid applications {self.applications}" + ) + return list(self.routing[app].keys()) + + @lru_cache(16) + def _get_app_specification(self, app: str, version: str = "latest") -> dict: + """ + Private methode to get the JSON schema of the API documentation. + + Parameters + ---------- + app: str + Name of the API. Call `applications` to obtain a list with all applications. + version: str, optional + default is latest. + API version used. Call `get_versions` to obtain a list with all version of a specific application. - response = requests.get( - self.get_url(app) + "/openapi.json", timeout=self.timeout + Returns + ------- + dict + + Raises + ------- + ConnectionError: + Application not available + TypeError: + Wrong type for `app` or `version` argument + ValueError: + Wrong value for `app` or `version` argument + """ + response = self.session.get( + self.get_url(app, version) + "/openapi.json", timeout=self.timeout ) if response.status_code != 200: raise ConnectionError( @@ -158,14 +201,17 @@ def _get_app_specification(self, app: str) -> dict: ) return response.json() - def get_application_version(self, app: str) -> str: + def get_application_version(self, app: str, version: str = "latest") -> str: """ - Provide version of the API in the Nuclei landscape. + Provide the semanctic version of the API in the Nuclei landscape. Parameters ---------- - app : str - Name of the API. + app: str + Name of the API. Call `applications` to obtain a list with all applications. + version: str, optional + default is latest. + API version used. Call `get_versions` to obtain a list with all version of a specific application. Returns ------- @@ -177,29 +223,27 @@ def get_application_version(self, app: str) -> str: ConnectionError: Application not available TypeError: - Wrong type for `app` argument + Wrong type for `app` or `version` argument ValueError: - Wrong value for `app` argument + Wrong value for `app` or `version` argument """ - if not isinstance(app, str): - raise TypeError( - f"Expected positional argument `app` to be of type , but got type: {type(app)}" - ) + return self._get_app_specification(app, version)["info"]["version"] - return self._get_app_specification(app)["info"]["version"] - - def get_endpoints(self, app: str) -> List[str]: + def get_endpoints(self, app: str, version: str = "latest") -> List[str]: """ Get available endpoints of single API. Parameters ---------- - app : str - Name of the API. + app: str + Name of the API. Call `applications` to obtain a list with all applications. + version: str, optional + default is latest + API version used. Call `get_versions` to obtain a list with all version of a specific application. Returns ------- - out : list[str] + endpoints : list[str] Endpoint urls. Raises @@ -207,28 +251,32 @@ def get_endpoints(self, app: str) -> List[str]: ConnectionError: Application not available TypeError: - Wrong type for `app` argument + Wrong type for `app` or `version` argument ValueError: - Wrong value for `app` argument + Wrong value for `app` or `version` argument """ - if not isinstance(app, str): - raise TypeError( - f"Expected positional argument `app` to be of type , but got type: {type(app)}" - ) + return list(self._get_app_specification(app, version)["paths"].keys()) - return list(self._get_app_specification(app)["paths"].keys()) - - def get_endpoint_type(self, app: str, endpoint: str) -> str: + def get_endpoint_type( + self, app: str, endpoint: str, version: str = "latest" + ) -> List[str]: """ + Get a list of HTTP methods used for this endpoint. + Parameters ---------- - app - name of the app + app: str + Name of the API. Call `applications` to obtain a list with all applications. endpoint url of the endpoint. + version: str, optional + default is latest. + API version used. Call `get_versions` to obtain a list with all version of a specific application. + Returns ------- - "get" | "post" + methods: List[str] + List of HTTP methods used for endpoint Raises ------- @@ -239,20 +287,17 @@ def get_endpoint_type(self, app: str, endpoint: str) -> str: ValueError: Wrong value for an argument """ - if not isinstance(app, str): - raise TypeError( - f"Expected positional argument `app` to be of type , but got type: {type(app)}" - ) - if not isinstance(endpoint, str): raise TypeError( f"Expected positional argument `endpoint` to be of type , but got type: {type(endpoint)}" ) - if endpoint in self.get_endpoints(app): - return list(self._get_app_specification(app)["paths"][endpoint].keys())[0] + if endpoint in self.get_endpoints(app, version): + return list( + self._get_app_specification(app, version)["paths"][endpoint].keys() + ) raise ValueError( - f"Endpoint name not valid, please select on of the following valid endpoints {self.get_endpoints(app)}" + f"Endpoint name not valid, please select on of the following valid endpoints {self.get_endpoints(app, version)}" ) def call_endpoint( @@ -260,6 +305,7 @@ def call_endpoint( app: str, endpoint: str, methode: Literal["auto", "get", "post"] = "auto", + version: str = "latest", schema: Optional[Union[dict, str]] = None, return_response: bool = False, ) -> Any: @@ -275,13 +321,17 @@ def call_endpoint( Parameters ---------- app: str - Name of the API. call `get_applications` to obtain a list with all applications. + Name of the API. Call `applications` to obtain a list with all applications. endpoint: str Name of the API's endpoint. call `get_endpoints` to obtain a list with all applications for a given API. methode: str - default is auto + default is auto HTTP methode used to call endpoint. When auto methode is selected the HTTP methode is - obtained from the openapi docs. + obtained from the openapi docs. Please note that this is the first one. call `get_endpoint_type` + to obtain list with all methods related to the endpoint. + version: str, optional + default is latest. + API version used. Call `get_versions` to obtain a list with all versions of a specific application. schema: dict or json-string, optional Default is None The parameter schema for the API. Take a look at the API documentation. @@ -316,16 +366,16 @@ def call_endpoint( ValueError: Wrong value for an argument """ - if not isinstance(app, str): - raise TypeError( - f"Expected positional argument `app` to be of type , but got type: {type(app)}" - ) - if not isinstance(endpoint, str): raise TypeError( f"Expected positional argument `endpoint` to be of type , but got type: {type(endpoint)}" ) + if endpoint not in self.get_endpoints(app, version): + raise ValueError( + f"Endpoint name not valid, please select on of the following valid endpoints {self.get_endpoints(app, version)}" + ) + if not isinstance(methode, str): raise TypeError( f"Expected keyword-argument `methode` to be of type , but got type: {type(methode)}" @@ -346,7 +396,7 @@ def call_endpoint( ) if methode == "auto": - t = self.get_endpoint_type(app, endpoint) + t = self.get_endpoint_type(app=app, version=version, endpoint=endpoint)[0] else: t = methode @@ -368,17 +418,22 @@ def call_endpoint( if t.lower() == "get": response = self.session.get( - self.get_url(app) + endpoint, + self.get_url(app=app, version=version) + endpoint, params=utils.serialize_jsonifyable_object(schema), timeout=self.timeout, ) elif t.lower() == "post": response = self.session.post( - self.get_url(app) + endpoint, + self.get_url(app=app, version=version) + endpoint, json=utils.serialize_jsonifyable_object(schema), timeout=self.timeout, ) + # If the response contains one of the following statuses, retry the request. + # 429 Too Many Requests + # 502 Bad Gateway + # 503 Service Unavailable + # 504 Gateway Timeout status_codes_that_trigger_retry = [429, 502, 503, 504] if response.status_code in status_codes_that_trigger_retry: # If the call failed, and retry is a viable option, we sleep for a diff --git a/src/nuclei/client/utils.py b/src/nuclei/client/utils.py index c22e0ba..cad81c6 100644 --- a/src/nuclei/client/utils.py +++ b/src/nuclei/client/utils.py @@ -1,8 +1,6 @@ from __future__ import annotations import json -import warnings -from collections.abc import Collection, Mapping from typing import Any try: @@ -112,42 +110,3 @@ def serialize_jsonifyable_object( a JSON string """ return orjson.loads(serialize_json_bytes(obj)) - - -def python_types_to_message( - schema: dict | np.ndarray | list | str | float | int | bool, -) -> Any: - """ - Cast python types to jsonifyable message. - - DEPRECATED since 0.5.0 - - Parameters - ---------- - schema - - Returns - ------- - message - jsonifyable message. - """ - warnings.warn( - "This function has been deprecated and will be removed in the future. It is " - "recommended to use `nuclei.client.utils.serialize_jsonifyable_object` instead.", - DeprecationWarning, - stacklevel=2, - ) - if isinstance(schema, np.ndarray): - # no NaN in array - if all(isinstance(x, (np.floating, np.integer)) for x in schema.flatten()): - return json.loads(json.dumps(schema.tolist()).replace("NaN", "null")) - else: - return schema.tolist() - if isinstance(schema, float) and np.isnan(schema): - return None - # recurse and convert to jsonifyable types - if isinstance(schema, Mapping): - return {k: python_types_to_message(v) for k, v in schema.items()} - if isinstance(schema, Collection) and not isinstance(schema, str): - return [python_types_to_message(k) for k in schema] - return schema diff --git a/tests/conftest.py b/tests/conftest.py index 6b83037..6026359 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -111,7 +111,7 @@ def mock_png_response(test_png): @pytest.fixture def session_send_post_returns_json(monkeypatch, mock_json_response): def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): - if request.url == routing["PileCore"] + "/MockPostEndpoint": + if request.url == routing["PileCore"]["latest"] + "/MockPostEndpoint": return mock_json_response raise ValueError( @@ -124,7 +124,10 @@ def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): @pytest.fixture def session_send_get_returns_json(monkeypatch, mock_json_response): def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): - if request.url == routing["PileCore"] + "/MockGetEndpoint?somekey=somevalue": + if ( + request.url + == routing["PileCore"]["latest"] + "/MockGetEndpoint?somekey=somevalue" + ): return mock_json_response raise ValueError( @@ -137,7 +140,7 @@ def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): @pytest.fixture def session_send_post_returns_b64_png(monkeypatch, mock_png_b64_response): def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): - if request.url == routing["PileCore"] + "/MockPostEndpoint": + if request.url == routing["PileCore"]["latest"] + "/MockPostEndpoint": return mock_png_b64_response raise ValueError( @@ -150,7 +153,7 @@ def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): @pytest.fixture def session_send_post_returns_png(monkeypatch, mock_png_response): def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): - if request.url == routing["PileCore"] + "/MockPostEndpoint": + if request.url == routing["PileCore"]["latest"] + "/MockPostEndpoint": return mock_png_response raise ValueError( @@ -163,7 +166,7 @@ def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): @pytest.fixture def session_send_post_returns_text(monkeypatch, mock_text_response): def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): - if request.url == routing["PileCore"] + "/MockPostEndpoint": + if request.url == routing["PileCore"]["latest"] + "/MockPostEndpoint": return mock_text_response raise ValueError( @@ -175,7 +178,7 @@ def mock_session_send(self, request: requests.models.PreparedRequest, **kwargs): @pytest.fixture def get_app_specification_post(monkeypatch): - def mock_get_app_specification(self, app: str) -> dict: + def mock_get_app_specification(self, app: str, version: str) -> dict: if app == "PileCore": return { "paths": { @@ -194,7 +197,7 @@ def mock_get_app_specification(self, app: str) -> dict: @pytest.fixture def get_app_specification_get(monkeypatch): - def mock_get_app_specification(self, app: str) -> dict: + def mock_get_app_specification(self, app: str, version: str) -> dict: if app == "PileCore": return { "paths": { @@ -213,7 +216,7 @@ def mock_get_app_specification(self, app: str) -> dict: @pytest.fixture def get_app_specification_invalid_method(monkeypatch): - def mock_get_app_specification(self, app: str) -> dict: + def mock_get_app_specification(self, app: str, version: str) -> dict: if app == "PileCore": return { "paths": { @@ -244,7 +247,7 @@ def user_token_envvar(monkeypatch): @pytest.fixture def get_app_specification_version(monkeypatch): - def mock_get_app_specification(self, app: str) -> dict: + def mock_get_app_specification(self, app: str, version: str) -> dict: if app == "PileCore": return { "info": { diff --git a/tests/test_client.py b/tests/test_client.py index ab8bf7c..6577739 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -48,16 +48,18 @@ def test_applications_to_endpoints(app, user_token_envvar): assert app in apps # Test get_url - assert isinstance(client.get_url(app), str) + assert isinstance(client.get_url(app, version="latest"), str) # Test get_endpoints. Assumes pulic openapi docs for all APIs. - endpoints = client.get_endpoints(app) + endpoints = client.get_endpoints(app, version="latest") for endpoint in endpoints: assert isinstance(endpoint, str) assert endpoint[0] == r"/" - endpoint_type = client.get_endpoint_type(app, endpoint) + endpoint_type = client.get_endpoint_type( + app, version="latest", endpoint=endpoint + )[0] assert endpoint_type in ("get", "post") @@ -69,13 +71,27 @@ def test_get_wrong_url(user_token_envvar, app): client = NucleiClient() with pytest.raises(ValueError) as err: - client.get_url(app) + client.get_url(app, version="latest") assert str(err.value).startswith( "Application not available, please select one of the following valid applications" ) +@pytest.mark.parametrize("version", ["v0"]) +def test_get_wrong_version(user_token_envvar, version): + "Tests if a ValueError is raised after passing an erroneous app version" + + client = NucleiClient() + + with pytest.raises(ValueError) as err: + client.get_url("PileCore", version=version) + + assert str(err.value).startswith( + "Application version not available, please select one of the following valid versions" + ) + + def test_get_userclaims(user_token_envvar): """Tests if the get_user_claims endpoint returns the correct claims for a free user""" @@ -91,7 +107,9 @@ def test_wrong_endpoint_for_type(user_token_envvar): client = NucleiClient() with pytest.raises(ValueError) as err: - client.get_endpoint_type("PileCore", "/some/invalid/endpoint") + client.get_endpoint_type( + "PileCore", version="latest", endpoint="/some/invalid/endpoint" + ) assert str(err.value).startswith( "Endpoint name not valid, please select on of the following valid endpoints" @@ -263,4 +281,7 @@ def test_call_endpoint_version( client = NucleiClient() - assert client.get_application_version(app="PileCore") == "0.1.0-beta.1" + assert ( + client.get_application_version(app="PileCore", version="latest") + == "0.1.0-beta.1" + ) diff --git a/tests/test_utils.py b/tests/test_utils.py index 9f6a444..0068cf6 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,7 +1,7 @@ import numpy as np import pytest -from nuclei.client.utils import python_types_to_message, serialize_jsonifyable_object +from nuclei.client.utils import serialize_jsonifyable_object @pytest.mark.parametrize( @@ -74,26 +74,3 @@ def __init__(self): with pytest.raises(TypeError): serialize_jsonifyable_object(Custom()) - - -def test_numpy_all_types(): - a = np.array([[np.int16(1), 2.0], [np.nan, np.float32(4)]]) - rsp = python_types_to_message(a) - assert rsp == [[1.0, 2.0], [None, 4.0]] - - a = np.array([[np.int16(1), 2.0, np.inf], [np.nan, np.float32(4), "abc"]]) - rsp = python_types_to_message(a) - assert rsp == [["1", "2.0", "inf"], ["nan", "4.0", "abc"]] - - -def test_list_str(): - msg = {"str": "str", "list": [1, 2, 3]} - resp = python_types_to_message(msg) - assert msg == resp - - -def test_literals(): - s = "str" - assert python_types_to_message(s) == s - list_ = [1, 2, 3] - assert python_types_to_message(list_) == list_