Skip to content
Open
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
18 changes: 16 additions & 2 deletions src/mcp_server_qdrant/mcp_server.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import json
import logging
from typing import Annotated, Any
from typing import Annotated, Any, Union

from fastmcp import Context, FastMCP
from pydantic import Field
Expand Down Expand Up @@ -76,7 +76,7 @@ async def store(
# If we set it to be optional, some of the MCP clients, like Cursor, cannot
# handle the optional parameter correctly.
metadata: Annotated[
Metadata | None,
Union[Metadata, str, None],
Field(
description="Extra metadata stored along with memorised information. Any json is accepted."
),
Expand All @@ -92,6 +92,20 @@ async def store(
:return: A message indicating that the information was stored.
"""
await ctx.debug(f"Storing information {information} in Qdrant")

# Десериализация metadata если это строка
if isinstance(metadata, str):
try:
if metadata.lower().strip() in ['null', 'none', '']:
metadata = None
else:
metadata = json.loads(metadata)
except json.JSONDecodeError:
raise ValueError(f"Invalid JSON in metadata: {metadata}")

# Валидация типа metadata после десериализации
if metadata is not None and not isinstance(metadata, dict):
raise TypeError(f"Metadata must be a dictionary, got {type(metadata)}")

entry = Entry(content=information, metadata=metadata)

Expand Down
2 changes: 1 addition & 1 deletion src/mcp_server_qdrant/qdrant.py
Original file line number Diff line number Diff line change
Expand Up @@ -131,7 +131,7 @@ async def search(

return [
Entry(
content=result.payload["document"],
content=result.payload.get("document") or result.payload.get("text", ""),
metadata=result.payload.get("metadata"),
)
for result in search_results.points
Expand Down
176 changes: 176 additions & 0 deletions tests/test_metadata_deserialization.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,176 @@
"""
Тесты для десериализации metadata в MCP Qdrant Server.
"""

import json
import pytest
from typing import Dict, Any, Union

from mcp_server_qdrant.qdrant import Metadata


class TestMetadataDeserialization:
"""Тестирование десериализации metadata"""

def test_metadata_dict_passthrough(self):
"""Тест: словарь metadata должен проходить без изменений"""
metadata = {"key": "value", "number": 123}
result = self._deserialize_metadata(metadata)
assert result == metadata

def test_metadata_none_passthrough(self):
"""Тест: None metadata должен проходить без изменений"""
metadata = None
result = self._deserialize_metadata(metadata)
assert result is None

def test_metadata_json_string_deserialization(self):
"""Тест: JSON строка должна десериализоваться в словарь"""
metadata = '{"key": "value", "number": 123}'
result = self._deserialize_metadata(metadata)
assert result == {"key": "value", "number": 123}

def test_metadata_empty_dict_string(self):
"""Тест: пустой JSON объект как строка"""
metadata = '{}'
result = self._deserialize_metadata(metadata)
assert result == {}

def test_metadata_null_string(self):
"""Тест: 'null' строка должна стать None"""
metadata = 'null'
result = self._deserialize_metadata(metadata)
assert result is None

def test_metadata_none_string(self):
"""Тест: 'none' строка должна стать None"""
metadata = 'none'
result = self._deserialize_metadata(metadata)
assert result is None

def test_metadata_None_string(self):
"""Тест: 'None' строка должна стать None"""
metadata = 'None'
result = self._deserialize_metadata(metadata)
assert result is None

def test_metadata_empty_string(self):
"""Тест: пустая строка должна стать None"""
metadata = ''
result = self._deserialize_metadata(metadata)
assert result is None

def test_metadata_complex_json(self):
"""Тест: сложный JSON объект"""
metadata = '{"nested": {"key": "value"}, "array": [1, 2, 3], "bool": true}'
result = self._deserialize_metadata(metadata)
assert result == {
"nested": {"key": "value"},
"array": [1, 2, 3],
"bool": True
}

def test_metadata_invalid_json_raises_error(self):
"""Тест: невалидный JSON должен вызывать ValueError"""
invalid_cases = [
'{invalid json}',
'{"unclosed": "quote}',
'not json at all',
'{"key": value}', # без кавычек вокруг value
]

for invalid in invalid_cases:
with pytest.raises(ValueError, match="Invalid JSON in metadata"):
self._deserialize_metadata(invalid)

def test_metadata_non_dict_after_deserialization_raises_error(self):
"""Тест: metadata не являющийся словарем после десериализации должен вызывать TypeError"""
invalid_cases = [
'"string value"', # строка в JSON
'123', # число в JSON
'true', # boolean в JSON
'[1, 2, 3]', # массив в JSON
]

for invalid in invalid_cases:
with pytest.raises(TypeError, match="Metadata must be a dictionary"):
self._deserialize_metadata(invalid)

def _deserialize_metadata(self, metadata: Union[Metadata, str, None]) -> Union[Metadata, None]:
"""
Внутренняя функция для тестирования логики десериализации.
Имитирует логику из store функции.
"""
# Десериализация metadata если это строка
if isinstance(metadata, str):
try:
if metadata.lower() in ['null', 'none', '']:
metadata = None
else:
metadata = json.loads(metadata)
except json.JSONDecodeError:
raise ValueError(f"Invalid JSON in metadata: {metadata}")

# Валидация типа metadata после десериализации
if metadata is not None and not isinstance(metadata, dict):
raise TypeError(f"Metadata must be a dictionary, got {type(metadata)}")

return metadata


class TestMetadataEdgeCases:
"""Тестирование граничных случаев"""

def test_metadata_whitespace_variations(self):
"""Тест: различные варианты пробелов"""
test_cases = [
(' null ', None),
(' none ', None),
(' None ', None),
(' "" ', ""), # пустая строка в JSON
(' {} ', {}),
]

for input_val, expected in test_cases:
if expected == "":
# Пустая строка в JSON должна вызвать TypeError
with pytest.raises(TypeError):
self._deserialize_metadata(input_val)
else:
result = self._deserialize_metadata(input_val)
assert result == expected

def test_metadata_case_sensitivity(self):
"""Тест: чувствительность к регистру для специальных значений"""
test_cases = [
('NULL', None),
('Null', None),
('NONE', None),
('None', None),
('nOnE', None),
]

for input_val, expected in test_cases:
result = self._deserialize_metadata(input_val)
assert result == expected

def _deserialize_metadata(self, metadata: Union[Metadata, str, None]) -> Union[Metadata, None]:
"""
Внутренняя функция для тестирования логики десериализации.
Имитирует логику из store функции.
"""
# Десериализация metadata если это строка
if isinstance(metadata, str):
try:
if metadata.lower().strip() in ['null', 'none', '']:
metadata = None
else:
metadata = json.loads(metadata)
except json.JSONDecodeError:
raise ValueError(f"Invalid JSON in metadata: {metadata}")

# Валидация типа metadata после десериализации
if metadata is not None and not isinstance(metadata, dict):
raise TypeError(f"Metadata must be a dictionary, got {type(metadata)}")

return metadata