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
5 changes: 5 additions & 0 deletions pecha_api/search/search_response_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,3 +77,8 @@ class MultilingualSearchResponse(BaseModel):
skip: int
limit: int
total: int


class SegmentLinkResponse(BaseModel):
text_id: str
segment_id: str
29 changes: 18 additions & 11 deletions pecha_api/search/search_service.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
from elastic_transport import ObjectApiResponse
from pecha_api.plans.response_message import NO_SEGMENTATION_IDS_RETURNED, PECHA_SEGMENT_NOT_FOUND
from fastapi import HTTPException
from starlette import status

from pecha_api.plans.response_message import NO_SEGMENTATION_IDS_RETURNED
from .search_enums import SearchType
from .search_client import search_client
from pecha_api.config import get
Expand All @@ -20,7 +23,8 @@
ExternalSearchResponse,
MultilingualSegmentMatch,
MultilingualSourceResult,
MultilingualSearchResponse
MultilingualSearchResponse,
SegmentLinkResponse,
)

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -446,17 +450,20 @@ async def build_multilingual_sources(segments: List[Segment], results_map: Dict[

return sources

async def get_url_link(pecha_segment_id: str) -> str:

async def get_url_link(pecha_segment_id: str) -> SegmentLinkResponse:
try:
segment = await Segment.get_segment_by_pecha_segment_id(pecha_segment_id=pecha_segment_id)

if not segment:
return PECHA_SEGMENT_NOT_FOUND

url = f"/chapter?text_id={segment.text_id}&segment_id={str(segment.id)}"
return url

raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pecha segment not found")

return SegmentLinkResponse(
text_id=segment.text_id,
segment_id=str(segment.id),
)

except HTTPException:
raise
except Exception as e:
logger.error(f"Error generating URL for pecha_segment_id {pecha_segment_id}: {str(e)}", exc_info=True)
return ""
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to retrieve segment link")
5 changes: 3 additions & 2 deletions pecha_api/search/search_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@

from .search_response_models import (
SearchResponse,
MultilingualSearchResponse
MultilingualSearchResponse,
SegmentLinkResponse,
)

search_router = APIRouter(
Expand Down Expand Up @@ -55,5 +56,5 @@ async def multilingual_search(
)

@search_router.get("/chat/{pecha_segment_id}", status_code=status.HTTP_200_OK)
async def get_url_link(pecha_segment_id: str) -> str:
async def get_url_link(pecha_segment_id: str) -> SegmentLinkResponse:
return await get_url_link_service(pecha_segment_id)
106 changes: 59 additions & 47 deletions tests/search/test_search_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -760,162 +760,174 @@ async def test_build_multilingual_sources_sorting():
async def test_get_url_link_success():
"""Test get_url_link service with valid pecha_segment_id"""
from pecha_api.search.search_service import get_url_link

mock_segment = Mock()
mock_segment.id = uuid4()
mock_segment.text_id = "text123"
mock_segment.pecha_segment_id = "pecha_seg_123"

with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock, return_value=mock_segment):
result = await get_url_link("pecha_seg_123")

assert result is not None
assert result == f"/chapter?text_id=text123&segment_id={str(mock_segment.id)}"
assert "text_id=" in result
assert "segment_id=" in result
assert result.text_id == "text123"
assert result.segment_id == str(mock_segment.id)


@pytest.mark.asyncio
async def test_get_url_link_segment_not_found():
"""Test get_url_link service when segment is not found"""
from fastapi import HTTPException
from starlette import status

from pecha_api.search.search_service import get_url_link
from pecha_api.plans.response_message import PECHA_SEGMENT_NOT_FOUND


with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock, return_value=None):
result = await get_url_link("nonexistent_segment_id")

assert result == PECHA_SEGMENT_NOT_FOUND
assert result == "Pecha segment not found"
with pytest.raises(HTTPException) as exc_info:
await get_url_link("nonexistent_segment_id")

assert exc_info.value.status_code == status.HTTP_404_NOT_FOUND
assert exc_info.value.detail == "Pecha segment not found"


@pytest.mark.asyncio
async def test_get_url_link_with_uuid_segment_id():
"""Test get_url_link service with UUID-formatted segment ID"""
from pecha_api.search.search_service import get_url_link

segment_uuid = uuid4()
text_uuid = uuid4()

mock_segment = Mock()
mock_segment.id = segment_uuid
mock_segment.text_id = str(text_uuid)
mock_segment.pecha_segment_id = str(uuid4())

with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock, return_value=mock_segment):
result = await get_url_link(mock_segment.pecha_segment_id)

assert result is not None
assert result == f"/chapter?text_id={str(text_uuid)}&segment_id={str(segment_uuid)}"
assert str(text_uuid) in result
assert str(segment_uuid) in result
assert result.text_id == str(text_uuid)
assert result.segment_id == str(segment_uuid)


@pytest.mark.asyncio
async def test_get_url_link_with_special_characters():
"""Test get_url_link service with special characters in pecha_segment_id"""
from pecha_api.search.search_service import get_url_link

mock_segment = Mock()
mock_segment.id = uuid4()
mock_segment.text_id = "text-abc-123"
mock_segment.pecha_segment_id = "pecha-seg_123-xyz"

with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock, return_value=mock_segment):
result = await get_url_link("pecha-seg_123-xyz")

assert result is not None
assert result == f"/chapter?text_id=text-abc-123&segment_id={str(mock_segment.id)}"
assert result.text_id == "text-abc-123"
assert result.segment_id == str(mock_segment.id)


@pytest.mark.asyncio
async def test_get_url_link_database_exception():
"""Test get_url_link service when database raises an exception"""
from fastapi import HTTPException
from starlette import status

from pecha_api.search.search_service import get_url_link

with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock, side_effect=Exception("Database connection error")):
result = await get_url_link("error_segment_id")

assert result == ""
with pytest.raises(HTTPException) as exc_info:
await get_url_link("error_segment_id")

assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert exc_info.value.detail == "Failed to retrieve segment link"


@pytest.mark.asyncio
async def test_get_url_link_with_long_pecha_segment_id():
"""Test get_url_link service with very long pecha_segment_id"""
from pecha_api.search.search_service import get_url_link

long_segment_id = "a" * 500

mock_segment = Mock()
mock_segment.id = uuid4()
mock_segment.text_id = "text123"
mock_segment.pecha_segment_id = long_segment_id

with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock, return_value=mock_segment):
result = await get_url_link(long_segment_id)

assert result is not None
assert result == f"/chapter?text_id=text123&segment_id={str(mock_segment.id)}"
assert result.text_id == "text123"
assert result.segment_id == str(mock_segment.id)


@pytest.mark.asyncio
async def test_get_url_link_with_empty_text_id():
"""Test get_url_link service when segment has empty text_id"""
from pecha_api.search.search_service import get_url_link

mock_segment = Mock()
mock_segment.id = uuid4()
mock_segment.text_id = ""
mock_segment.pecha_segment_id = "pecha_seg_123"

with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock, return_value=mock_segment):
result = await get_url_link("pecha_seg_123")

assert result is not None
assert result == f"/chapter?text_id=&segment_id={str(mock_segment.id)}"
assert result.text_id == ""
assert result.segment_id == str(mock_segment.id)


@pytest.mark.asyncio
async def test_get_url_link_multiple_calls():
"""Test get_url_link service with multiple sequential calls"""
from pecha_api.search.search_service import get_url_link

mock_segment1 = Mock()
mock_segment1.id = uuid4()
mock_segment1.text_id = "text1"
mock_segment1.pecha_segment_id = "seg1"

mock_segment2 = Mock()
mock_segment2.id = uuid4()
mock_segment2.text_id = "text2"
mock_segment2.pecha_segment_id = "seg2"

with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock) as mock_get:
mock_get.return_value = mock_segment1
result1 = await get_url_link("seg1")
assert result1 == f"/chapter?text_id=text1&segment_id={str(mock_segment1.id)}"

assert result1.text_id == "text1"
assert result1.segment_id == str(mock_segment1.id)

mock_get.return_value = mock_segment2
result2 = await get_url_link("seg2")
assert result2 == f"/chapter?text_id=text2&segment_id={str(mock_segment2.id)}"

assert result2.text_id == "text2"
assert result2.segment_id == str(mock_segment2.id)

assert mock_get.call_count == 2


@pytest.mark.asyncio
async def test_get_url_link_none_segment_id():
"""Test get_url_link service when segment.id is None"""
from pecha_api.search.search_service import get_url_link

mock_segment = Mock()
mock_segment.id = None
mock_segment.text_id = "text123"
mock_segment.pecha_segment_id = "pecha_seg_123"

with patch("pecha_api.search.search_service.Segment.get_segment_by_pecha_segment_id", new_callable=AsyncMock, return_value=mock_segment):
result = await get_url_link("pecha_seg_123")

assert result is not None
assert result == f"/chapter?text_id=text123&segment_id=None"
assert result.text_id == "text123"
assert result.segment_id == "None"

@pytest.mark.asyncio
async def test_get_multilingual_search_external_limit_calculation():
Expand Down
Loading
Loading