Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions glpi_utils/aio.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
from typing import Any, AsyncIterator, Optional

from ._resource import AsyncItemProxy
from .api import DEFAULT_PAGE_SIZE, _ITEMTYPE_MAP, _boolify_params, _parse_content_range, _raise_for_glpi_error
from .api import DEFAULT_PAGE_SIZE, _ITEMTYPE_MAP, _boolify_params, _flatten_search_params, _parse_content_range, _raise_for_glpi_error
from .exceptions import GlpiAPIError, GlpiAuthError, GlpiConnectionError
from .logger import EmptyHandler, SensitiveFilter
from .version import GLPIVersion
Expand Down Expand Up @@ -425,7 +425,7 @@ async def iter_pages(
start += page_size

async def search(self, itemtype: str, **kwargs: Any) -> dict:
params = _boolify_params(kwargs)
params = _boolify_params(_flatten_search_params(kwargs))
return await self._request("GET", f"search/{itemtype}", params=params)

async def create_item(self, itemtype: str, input_data: Any, **kwargs: Any) -> Any:
Expand Down
28 changes: 27 additions & 1 deletion glpi_utils/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,32 @@ def _raise_for_glpi_error(response: Response) -> None:
raise GlpiAPIError(message, status_code=status, error_code=error_code)


def _flatten_search_params(params: dict) -> dict:
"""Flatten nested search parameters into PHP-style bracket notation.

GLPI expects ``criteria[0][field]=126&criteria[0][searchtype]=equals``
rather than Python list/dict structures that HTTP clients cannot
serialize correctly.
"""
flat: dict = {}

for key in ("criteria", "metacriteria"):
items = params.pop(key, None)
if not items:
continue
for i, crit in enumerate(items):
for sub_key, value in crit.items():
flat[f"{key}[{i}][{sub_key}]"] = value

forcedisplay = params.pop("forcedisplay", None)
if forcedisplay:
for i, field in enumerate(forcedisplay):
flat[f"forcedisplay[{i}]"] = field

flat.update(params)
return flat


def _boolify_params(params: dict) -> dict:
"""Convert Python ``bool`` values to GLPI-expected ``0``/``1`` integers."""
return {k: int(v) if isinstance(v, bool) else v for k, v in params.items()}
Expand Down Expand Up @@ -651,7 +677,7 @@ def search(self, itemtype: str, **kwargs: Any) -> dict:
``criteria``, ``metacriteria``, ``sort``, ``order``,
``range``, ``forcedisplay``, ``rawdata``, ``withindexes``.
"""
params = _boolify_params(kwargs)
params = _boolify_params(_flatten_search_params(kwargs))
return self._request("GET", f"search/{itemtype}", params=params)

def create_item(self, itemtype: str, input_data: Any, **kwargs: Any) -> Any:
Expand Down
115 changes: 115 additions & 0 deletions tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import requests

from glpi_utils import GlpiAPI, GlpiAPIError, GlpiAuthError, GlpiNotFoundError
from glpi_utils.api import _flatten_search_params
from glpi_utils.exceptions import GlpiConnectionError, GlpiPermissionError
from glpi_utils.version import GLPIVersion

Expand Down Expand Up @@ -382,6 +383,120 @@ def test_search_returns_dict(self, mock_req):
self.assertEqual(result["totalcount"], 5)


# ──────────────────────────────────────────────────────────────────────────────
# Search parameter flattening
# ──────────────────────────────────────────────────────────────────────────────


class TestFlattenSearchParams(unittest.TestCase):

def test_criteria_flattened(self):
params = {
"criteria": [
{"field": 1, "searchtype": "contains", "value": "test"},
],
}
flat = _flatten_search_params(params)
self.assertEqual(flat["criteria[0][field]"], 1)
self.assertEqual(flat["criteria[0][searchtype]"], "contains")
self.assertEqual(flat["criteria[0][value]"], "test")
self.assertNotIn("criteria", flat)

def test_multiple_criteria(self):
params = {
"criteria": [
{"field": 1, "searchtype": "contains", "value": "a"},
{"field": 5, "searchtype": "equals", "value": "b"},
],
}
flat = _flatten_search_params(params)
self.assertEqual(flat["criteria[0][field]"], 1)
self.assertEqual(flat["criteria[1][field]"], 5)
self.assertEqual(flat["criteria[1][value]"], "b")

def test_metacriteria_flattened(self):
params = {
"metacriteria": [
{"link": "AND", "itemtype": "User", "field": 1, "searchtype": "equals", "value": "admin"},
],
}
flat = _flatten_search_params(params)
self.assertEqual(flat["metacriteria[0][itemtype]"], "User")
self.assertEqual(flat["metacriteria[0][field]"], 1)

def test_forcedisplay_flattened(self):
params = {"forcedisplay": [1, 5, 126]}
flat = _flatten_search_params(params)
self.assertEqual(flat["forcedisplay[0]"], 1)
self.assertEqual(flat["forcedisplay[1]"], 5)
self.assertEqual(flat["forcedisplay[2]"], 126)
self.assertNotIn("forcedisplay", flat)

def test_scalar_params_preserved(self):
params = {"range": "0-49", "sort": 1, "order": "ASC"}
flat = _flatten_search_params(params)
self.assertEqual(flat["range"], "0-49")
self.assertEqual(flat["sort"], 1)
self.assertEqual(flat["order"], "ASC")

def test_combined_criteria_and_forcedisplay(self):
params = {
"criteria": [{"field": 126, "searchtype": "equals", "value": "10.0.0.1"}],
"forcedisplay": [1, 5, 126],
"range": "0-1999",
}
flat = _flatten_search_params(params)
self.assertEqual(flat["criteria[0][field]"], 126)
self.assertEqual(flat["forcedisplay[0]"], 1)
self.assertEqual(flat["range"], "0-1999")

def test_empty_criteria_ignored(self):
params = {"criteria": [], "range": "0-49"}
flat = _flatten_search_params(params)
self.assertNotIn("criteria[0][field]", flat)
self.assertEqual(flat["range"], "0-49")

def test_no_nested_params_is_passthrough(self):
params = {"range": "0-49", "rawdata": True}
flat = _flatten_search_params(params)
self.assertEqual(flat, {"range": "0-49", "rawdata": True})


class TestSearchFlattensParams(unittest.TestCase):

def setUp(self):
self.api = make_api()

@patch("requests.Session.request")
def test_search_flattens_criteria(self, mock_req):
mock_req.return_value = mock_response(200, {"totalcount": 1, "data": []})
self.api.search(
"Computer",
criteria=[{"field": 126, "searchtype": "equals", "value": "10.0.0.1"}],
forcedisplay=[1, 5, 126],
range="0-1999",
)
params = mock_req.call_args[1]["params"]
self.assertEqual(params["criteria[0][field]"], 126)
self.assertEqual(params["criteria[0][searchtype]"], "equals")
self.assertEqual(params["criteria[0][value]"], "10.0.0.1")
self.assertEqual(params["forcedisplay[0]"], 1)
self.assertEqual(params["forcedisplay[2]"], 126)
self.assertEqual(params["range"], "0-1999")
self.assertNotIn("criteria", params)
self.assertNotIn("forcedisplay", params)

@patch("requests.Session.request")
def test_search_proxy_flattens_criteria(self, mock_req):
mock_req.return_value = mock_response(200, {"totalcount": 0, "data": []})
self.api.computer.search(
criteria=[{"field": 1, "searchtype": "contains", "value": "SRV"}],
)
params = mock_req.call_args[1]["params"]
self.assertEqual(params["criteria[0][field]"], 1)
self.assertNotIn("criteria", params)


# ──────────────────────────────────────────────────────────────────────────────
# Sub-items
# ──────────────────────────────────────────────────────────────────────────────
Expand Down
Loading