import asyncio import uuid from datetime import datetime from types import SimpleNamespace from unittest.mock import AsyncMock, Mock, patch import pytest from fastapi import HTTPException from api.v1.ppt.endpoints import presentation as presentation_endpoint from constants.presentation import MAX_NUMBER_OF_SLIDES from models.generate_presentation_request import GeneratePresentationRequest from models.presentation_and_path import PresentationAndPath from models.presentation_from_template import EditPresentationRequest, SlideContentUpdate from models.presentation_outline_model import SlideOutlineModel from models.presentation_structure_model import PresentationStructureModel from models.sql.presentation import PresentationModel, PresentationVersion from models.sql.slide import SlideModel from models.sql.template_v2 import TemplateV2 from templates.presentation_layout import PresentationLayoutModel, SlideLayoutModel from tests.conftest import FakeAsyncSession class FakeRequest: def __init__(self): self.headers: dict[str, str] = {} self.cookies: dict[str, str] = {} self.state = SimpleNamespace() def _run(coro): return asyncio.run(coro) def _mock_layout() -> PresentationLayoutModel: return PresentationLayoutModel( name="general", ordered=False, slides=[ SlideLayoutModel(id="layout-1", name="Title", json_schema={"title": "title"}), SlideLayoutModel(id="layout-2", name="Body", json_schema={"title": "body"}), ], ) def _template_v2_layout_payload() -> dict: return { "layouts": [ { "id": "template-layout-1", "description": "Hero layout", "components": [ { "id": "hero", "description": "Hero content", "elements": [ { "type": "text", "decorative": False, "name": "headline", "runs": [{"text": "Original headline"}], } ], } ], } ] } def test_generate_presentation_handler_full_flow_uses_mocked_dependencies(fake_async_session): request = GeneratePresentationRequest( content="Create a two-slide deck about renewable energy.", n_slides=2, language="English", export_as="pptx", template="general", ) presentation_id = uuid.uuid4() async def fake_outline_stream(*_args, **_kwargs): yield '{"slides":[{"content":"## Intro"},{"content":"## Action Plan"}]}' get_slide_content = AsyncMock( side_effect=[ {"title": "Intro", "points": ["A"]}, {"title": "Action Plan", "points": ["B"]}, ] ) with patch.object( presentation_endpoint.MEM0_PRESENTATION_MEMORY_SERVICE, "store_generation_context", new=AsyncMock(), ), patch.object( presentation_endpoint.MEM0_PRESENTATION_MEMORY_SERVICE, "store_generated_outlines", new=AsyncMock(), ), patch.object( presentation_endpoint, "generate_ppt_outline", side_effect=fake_outline_stream, ), patch.object( presentation_endpoint, "get_layout_by_name", new=AsyncMock(return_value=_mock_layout()), ), patch.object( presentation_endpoint, "generate_presentation_structure", new=AsyncMock(return_value=PresentationStructureModel(slides=[0, 1])), ), patch.object( presentation_endpoint, "get_slide_content_from_type_and_outline", get_slide_content, ), patch.object( presentation_endpoint, "process_slide_and_fetch_assets", new=AsyncMock(return_value=[]), ), patch.object( presentation_endpoint, "get_images_directory", return_value="/tmp", ), patch.object( presentation_endpoint, "ImageGenerationService", return_value=Mock(), ), patch.object( presentation_endpoint, "export_presentation", new=AsyncMock( return_value=PresentationAndPath( presentation_id=presentation_id, path="/tmp/generated/deck.pptx", ) ), ), patch.object( presentation_endpoint.CONCURRENT_SERVICE, "run_task", new=Mock(), ), patch.object( presentation_endpoint, "random", new=Mock(randint=Mock(return_value=0)), ): response = _run( presentation_endpoint.generate_presentation_handler( request=request, presentation_id=presentation_id, async_status=None, sql_session=fake_async_session, ) ) assert response.path.endswith(".pptx") assert response.edit_path == f"/presentation?id={presentation_id}" assert len(fake_async_session.added_all) == 2 assert all(slide.presentation == presentation_id for slide in fake_async_session.added_all) assert all(slide.ui is None for slide in fake_async_session.added_all) def test_generate_presentation_handler_uses_template_v2_layout(): template_id = str(uuid.uuid4()) presentation_id = uuid.uuid4() template = TemplateV2( id=template_id, name="Custom V2", layouts=_template_v2_layout_payload(), assets={"fonts": {"Inter": "https://example.com/inter.css"}}, ) session = FakeAsyncSession(get_results={template_id: template}) request = GeneratePresentationRequest( content="Create a one-slide deck about climate risk.", n_slides=1, language="English", export_as="pptx", template=f"custom-{template_id}", ) async def fake_outline_stream(*_args, **_kwargs): yield '{"slides":[{"content":"## Climate risk"}]}' with patch.object( presentation_endpoint.MEM0_PRESENTATION_MEMORY_SERVICE, "store_generation_context", new=AsyncMock(), ), patch.object( presentation_endpoint.MEM0_PRESENTATION_MEMORY_SERVICE, "store_generated_outlines", new=AsyncMock(), ), patch.object( presentation_endpoint, "generate_ppt_outline", side_effect=fake_outline_stream, ), patch.object( presentation_endpoint, "generate_presentation_structure", new=AsyncMock(return_value=PresentationStructureModel(slides=[0])), ), patch.object( presentation_endpoint, "get_slide_content_from_type_and_outline", new=AsyncMock(return_value={"hero": {"headline": "V2 headline"}}), ), patch.object( presentation_endpoint, "process_slide_and_fetch_assets", new=AsyncMock(return_value=[]), ), patch.object( presentation_endpoint, "get_images_directory", return_value="/tmp", ), patch.object( presentation_endpoint, "ImageGenerationService", return_value=Mock(), ), patch.object( presentation_endpoint, "export_presentation", new=AsyncMock( return_value=PresentationAndPath( presentation_id=presentation_id, path="/tmp/generated/deck.pptx", ) ), ), patch.object( presentation_endpoint.CONCURRENT_SERVICE, "run_task", new=Mock(), ), patch.object( presentation_endpoint, "random", new=Mock(randint=Mock(return_value=0)), ): _run( presentation_endpoint.generate_presentation_handler( request=request, presentation_id=presentation_id, async_status=None, sql_session=session, ) ) presentation = next( item for item in session.added if isinstance(item, PresentationModel) ) slide = next(item for item in session.added_all if isinstance(item, SlideModel)) assert presentation.version == PresentationVersion.V2_STANDARD assert presentation.layout == _template_v2_layout_payload() assert presentation.fonts == {"Inter": "https://example.com/inter.css"} assert slide.layout_group == f"template-v2-{template_id}" assert slide.ui["components"][0]["elements"][0]["runs"][0]["text"] == "V2 headline" def test_create_presentation_stores_current_version(fake_async_session): presentation = _run( presentation_endpoint.create_presentation( content="Create a short deck.", sql_session=fake_async_session, ) ) assert presentation.version == PresentationVersion.V2_STANDARD assert presentation.content == "Create a short deck." assert fake_async_session.added == [presentation] assert fake_async_session.commit_count == 1 def test_check_api_request_accepts_custom_prefixed_template_v2_id(): template_id = "general-template" template = TemplateV2( id=template_id, name="Custom V2", layouts=_template_v2_layout_payload(), ) session = FakeAsyncSession(get_results={template_id: template}) request = GeneratePresentationRequest( content="Create a deck.", n_slides=1, template=f"custom-{template_id}", ) _run(presentation_endpoint.check_if_api_request_is_valid(request, session)) assert request.template == f"template-v2-{template_id}" def test_check_api_request_rejects_custom_id_without_template_v2(fake_async_session): request = GeneratePresentationRequest( content="Create a deck.", n_slides=1, template=f"custom-{uuid.uuid4()}", ) with pytest.raises(HTTPException) as exc: _run( presentation_endpoint.check_if_api_request_is_valid( request, fake_async_session, ) ) assert exc.value.status_code == 400 assert exc.value.detail == "Template not found. Please use a valid template." def test_prepare_presentation_preserves_payload_icon_weight(): presentation_id = uuid.uuid4() presentation = PresentationModel( id=presentation_id, version=PresentationVersion.V1_STANDARD, content="deck", n_slides=1, language="English", tone="default", verbosity="standard", instructions=None, ) session = FakeAsyncSession(get_results={presentation_id: presentation}) layout = PresentationLayoutModel( name="swift", ordered=False, icon_weight="thin", slides=[ SlideLayoutModel( id="swift:feature", name="Feature", description="Feature slide", json_schema={"title": "Feature"}, ) ], ) with patch.object( presentation_endpoint, "generate_presentation_structure", new=AsyncMock(return_value=PresentationStructureModel(slides=[0])), ), patch.object( presentation_endpoint.MEM0_PRESENTATION_MEMORY_SERVICE, "store_generated_outlines", new=AsyncMock(), ): response = _run( presentation_endpoint.prepare_presentation( presentation_id=presentation_id, outlines=[SlideOutlineModel(content="## Causes")], layout=layout, sql_session=session, ) ) assert response.layout["icon_weight"] == "thin" assert response.get_layout().icon_weight == "thin" assert response.language == "" def test_prepare_presentation_clears_stale_language_for_reviewed_outlines(): presentation_id = uuid.uuid4() presentation = PresentationModel( id=presentation_id, content="global warming deck", n_slides=1, language="Spanish (Español)", tone="default", verbosity="standard", instructions=None, ) session = FakeAsyncSession(get_results={presentation_id: presentation}) layout = PresentationLayoutModel( name="swift", ordered=False, slides=[ SlideLayoutModel( id="swift:feature", name="Feature", description="Feature slide", json_schema={"title": "Feature"}, ) ], ) with patch.object( presentation_endpoint, "generate_presentation_structure", new=AsyncMock(return_value=PresentationStructureModel(slides=[0])), ), patch.object( presentation_endpoint.MEM0_PRESENTATION_MEMORY_SERVICE, "store_generated_outlines", new=AsyncMock(), ): response = _run( presentation_endpoint.prepare_presentation( presentation_id=presentation_id, outlines=[SlideOutlineModel(content="## 全球变暖的原因")], layout=layout, sql_session=session, ) ) assert response.language == "" def test_prepare_presentation_rejects_too_many_outlines(): presentation_id = uuid.uuid4() presentation = PresentationModel( id=presentation_id, content="deck", n_slides=MAX_NUMBER_OF_SLIDES, language="English", tone="default", verbosity="standard", instructions=None, ) session = FakeAsyncSession(get_results={presentation_id: presentation}) with pytest.raises(HTTPException) as exc_info: _run( presentation_endpoint.prepare_presentation( presentation_id=presentation_id, outlines=[ SlideOutlineModel(content=f"## Slide {index}") for index in range(MAX_NUMBER_OF_SLIDES + 1) ], layout=_mock_layout(), sql_session=session, ) ) assert exc_info.value.status_code == 400 assert ( exc_info.value.detail == f"Number of outlines cannot be greater than {MAX_NUMBER_OF_SLIDES}" ) def test_check_api_request_rejects_too_many_slides_markdown(fake_async_session): request = GeneratePresentationRequest( content="", slides_markdown=[ f"## Slide {index}" for index in range(MAX_NUMBER_OF_SLIDES + 1) ], export_as="pdf", template="general", ) with pytest.raises(HTTPException) as exc_info: _run( presentation_endpoint.check_if_api_request_is_valid( request=request, sql_session=fake_async_session, ) ) assert exc_info.value.status_code == 400 assert ( exc_info.value.detail == f"Number of slides cannot be greater than {MAX_NUMBER_OF_SLIDES}" ) def test_prepare_presentation_accepts_template_v2_layout_id(): presentation_id = uuid.uuid4() template_id = "strategy-template" presentation = PresentationModel( id=presentation_id, version=PresentationVersion.V2_STANDARD, content="deck", n_slides=1, language="English", tone="default", verbosity="standard", instructions=None, ) template_layouts = { "layouts": [ { "id": "template-layout-1", "description": "Hero layout", "components": [ { "id": "hero", "description": "Hero content", "elements": [ { "type": "text", "decorative": False, "name": "headline", } ], } ], } ] } template = TemplateV2( id=template_id, name="Custom V2", layouts=template_layouts, assets={"fonts": {"Inter": "https://example.com/inter.css"}}, ) session = FakeAsyncSession( get_results={presentation_id: presentation, template_id: template} ) generate_structure = AsyncMock( return_value=PresentationStructureModel(slides=[0]) ) with patch.object( presentation_endpoint, "generate_presentation_structure", new=generate_structure, ), patch.object( presentation_endpoint.MEM0_PRESENTATION_MEMORY_SERVICE, "store_generated_outlines", new=AsyncMock(), ): response = _run( presentation_endpoint.prepare_presentation( presentation_id=presentation_id, outlines=[SlideOutlineModel(content="## Causes")], layout=str(template_id), sql_session=session, ) ) assert response.layout == template_layouts assert response.structure == {"slides": [0]} assert response.fonts == {"Inter": "https://example.com/inter.css"} structure_layout = generate_structure.await_args.kwargs["presentation_layout"] assert structure_layout.name == f"template-v2-{template_id}" assert structure_layout.slides[0].id == "template-layout-1" def test_get_presentation_preserves_template_v2_detail_payload(): presentation_id = uuid.uuid4() now = datetime.now() template_layouts = { "layouts": [ { "id": "slide_1", "description": "Full slide layout converted from PPTX slide 1.", "components": [], } ] } template_components = { "cluster_count": 1, "component_count": 1, "components": [ { "id": "hero", "description": "Reusable hero content component.", "position": {"x": 24, "y": 32}, "size": {"width": 640, "height": 200}, "elements": [ { "type": "text", "position": {"x": 0, "y": 0}, "size": {"width": 640, "height": 80}, "runs": [{"text": "Hero"}], } ], } ], } structure = {"slides": [0]} presentation = PresentationModel( id=presentation_id, version=PresentationVersion.V2_STANDARD, content="deck", n_slides=1, language="English", title="Deck", layout=template_layouts, structure=structure, tone="default", verbosity="standard", instructions=None, created_at=now, updated_at=now, ) template = TemplateV2( id=str(uuid.uuid4()), name="presentation", layouts=template_layouts, assets={ "fonts": { "Inter": "https://example.com/inter.css", "Brand Sans": "/app_data/fonts/brand-sans.ttf", } }, components=template_components, merged_components=template_components, ) slides = [ SlideModel( presentation=presentation_id, layout_group=f"template-v2-{template.id}", layout="slide_1", index=0, content={}, ) ] class _TemplateV2ComponentSession(FakeAsyncSession): async def scalars(self, *_args, **_kwargs): return slides async def execute(self, *_args, **_kwargs): class _RowsResult: def all(self): return [(template.id, template.layouts, template.components)] return _RowsResult() session = _TemplateV2ComponentSession( get_results={presentation_id: presentation, template.id: template} ) response = _run( presentation_endpoint.get_presentation( id=presentation_id, request=FakeRequest(), sql_session=session, ) ) assert response.version == PresentationVersion.V2_STANDARD assert response.layout == template_layouts assert response.structure == structure assert not hasattr(response, "components") assert response.fonts == { "Inter": "https://example.com/inter.css", "Brand Sans": "/app_data/fonts/brand-sans.ttf", } assert response.merged_components == template_components def test_stream_presentation_uses_template_v2_schema_for_content_generation(): presentation_id = uuid.uuid4() now = datetime.now() template_layouts = { "layouts": [ { "id": "template-layout-1", "description": "Hero layout", "components": [ { "id": "hero", "description": "Hero content", "elements": [ { "type": "text", "decorative": False, "name": "headline", "min_length": 2, "max_length": 32, } ], } ], } ] } presentation = PresentationModel( id=presentation_id, version=PresentationVersion.V2_STANDARD, content="deck", n_slides=1, language="English", title="Deck", outlines={"slides": [{"content": "## Causes"}]}, layout=template_layouts, structure={"slides": [0]}, tone="default", verbosity="standard", instructions=None, created_at=now, updated_at=now, ) session = FakeAsyncSession(get_results={presentation_id: presentation}) generated_layouts: list[SlideLayoutModel] = [] async def fake_slide_content(slide_layout, *_args, **_kwargs): generated_layouts.append(slide_layout) return { "hero": {"headline": "Causes"}, "__speaker_note__": "Speaker note for this generated slide.", } async def consume_stream(): response = await presentation_endpoint.stream_presentation( id=presentation_id, sql_session=session, ) chunks = [] async for chunk in response.body_iterator: chunks.append(chunk) return chunks with patch.object( presentation_endpoint, "get_slide_content_from_type_and_outline", new=fake_slide_content, ), patch.object( presentation_endpoint, "process_slide_and_fetch_assets", new=AsyncMock(return_value=[]), ), patch.object( presentation_endpoint, "get_images_directory", return_value="/tmp", ), patch.object( presentation_endpoint, "ImageGenerationService", return_value=Mock(), ): chunks = _run(consume_stream()) assert chunks assert len(generated_layouts) == 1 generated_layout = generated_layouts[0] assert generated_layout.id == "template-layout-1" assert generated_layout.name == "template-layout-1" assert generated_layout.description == "Hero layout" assert generated_layout.json_schema["title"] == "template-layout-1" assert generated_layout.json_schema["properties"]["hero"] == { "type": "object", "additionalProperties": False, "properties": { "headline": { "type": "string", "minLength": 2, "maxLength": 32, } }, "required": ["headline"], } generated_slides = [ item for item in session.added_all if isinstance(item, SlideModel) ] assert len(generated_slides) == 1 assert ( generated_slides[0].ui["components"][0]["elements"][0]["runs"][0]["text"] == "Causes" ) assert template_layouts["layouts"][0]["components"][0]["elements"][0].get( "runs" ) is None def test_generate_presentation_sync_rejects_invalid_slide_count(fake_async_session): request = GeneratePresentationRequest( content="deck", n_slides=0, language="English", export_as="pdf", template="general", ) with pytest.raises(HTTPException) as exc: _run( presentation_endpoint.generate_presentation_sync( request_http=Mock(cookies={}), request=request, sql_session=fake_async_session, ) ) assert exc.value.status_code == 400 assert "Number of slides must be greater than 0" in exc.value.detail def test_generate_presentation_handler_rejects_invalid_llm_json(fake_async_session): request = GeneratePresentationRequest( content="Generate a small deck", n_slides=2, language="English", export_as="pdf", template="general", ) async def fake_outline_stream(*_args, **_kwargs): yield "{invalid-json" with patch.object( presentation_endpoint.MEM0_PRESENTATION_MEMORY_SERVICE, "store_generation_context", new=AsyncMock(), ), patch.object( presentation_endpoint, "generate_ppt_outline", side_effect=fake_outline_stream, ), patch.object( presentation_endpoint.CONCURRENT_SERVICE, "run_task", new=Mock(), ): with pytest.raises(HTTPException) as exc: _run( presentation_endpoint.generate_presentation_handler( request=request, presentation_id=uuid.uuid4(), async_status=None, sql_session=fake_async_session, ) ) assert exc.value.status_code == 400 assert "Failed to generate presentation outlines" in exc.value.detail class _SlidesAsyncSession(FakeAsyncSession): def __init__(self, *args, slides: list[SlideModel], **kwargs): super().__init__(*args, **kwargs) self._slides = slides async def scalars(self, *_args, **_kwargs): return self._slides async def _fake_export_presentation(presentation_id, *_args, **_kwargs): return PresentationAndPath( presentation_id=presentation_id, path="/tmp/generated/deck.pptx", ) def test_edit_presentation_hydrates_template_v2_slide_ui(): presentation_id = uuid.uuid4() template_id = str(uuid.uuid4()) layout_payload = _template_v2_layout_payload() presentation = PresentationModel( id=presentation_id, version=PresentationVersion.V2_STANDARD, content="deck", n_slides=1, language="English", layout=layout_payload, tone="default", verbosity="standard", instructions=None, ) slide = SlideModel( presentation=presentation_id, layout_group=f"template-v2-{template_id}", layout="template-layout-1", index=0, content={"hero": {"headline": "Old headline"}}, ui=layout_payload["layouts"][0], ) session = _SlidesAsyncSession( get_results={presentation_id: presentation}, slides=[slide], ) request = EditPresentationRequest( presentation_id=presentation_id, slides=[ SlideContentUpdate( index=0, content={"hero": {"headline": "Updated headline"}}, ) ], ) with patch.object( presentation_endpoint, "export_presentation", new=_fake_export_presentation, ): _run( presentation_endpoint.edit_presentation_with_new_content( request_http=FakeRequest(), data=request, sql_session=session, ) ) new_slide = next(item for item in session.added_all if isinstance(item, SlideModel)) title_element = new_slide.ui["components"][0]["elements"][0] assert title_element["runs"][0]["text"] == "Updated headline" def test_derive_presentation_hydrates_template_v2_slide_ui(): presentation_id = uuid.uuid4() template_id = str(uuid.uuid4()) layout_payload = _template_v2_layout_payload() presentation = PresentationModel( id=presentation_id, version=PresentationVersion.V2_STANDARD, content="deck", n_slides=1, language="English", layout=layout_payload, tone="default", verbosity="standard", instructions=None, ) slide = SlideModel( presentation=presentation_id, layout_group=f"template-v2-{template_id}", layout="template-layout-1", index=0, content={"hero": {"headline": "Old headline"}}, ui=layout_payload["layouts"][0], ) session = _SlidesAsyncSession( get_results={presentation_id: presentation}, slides=[slide], ) request = EditPresentationRequest( presentation_id=presentation_id, slides=[ SlideContentUpdate( index=0, content={"hero": {"headline": "Derived headline"}}, ) ], ) with patch.object( presentation_endpoint, "export_presentation", new=_fake_export_presentation, ): _run( presentation_endpoint.derive_presentation_from_existing_one( request_http=FakeRequest(), data=request, sql_session=session, ) ) new_presentation = next( item for item in session.added if isinstance(item, PresentationModel) ) new_slide = next(item for item in session.added_all if isinstance(item, SlideModel)) title_element = new_slide.ui["components"][0]["elements"][0] assert new_presentation.version == PresentationVersion.V2_STANDARD assert new_slide.presentation == new_presentation.id assert title_element["runs"][0]["text"] == "Derived headline"