diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..6af015b --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,24 @@ +name: CI + +on: + pull_request: + branches: [main] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: astral-sh/setup-uv@v6 + with: + python-version: "3.13" + + - name: Install dependencies + run: uv sync --extra dev + + - name: Lint + run: uv run ruff check . + + - name: Test + run: uv run pytest tests/ -x -q diff --git a/.gitignore b/.gitignore index c7d4ad7..331b1ee 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,6 @@ __pycache__/ *.mp3 *.mp4 *.wav +.venv/ +.pytest_cache/ +uv.lock diff --git a/README.md b/README.md index 4bd88d7..f262842 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,13 @@ # prax-plugins -Example plugin collection for [Prax](https://github.com/praxagent/prax). Each subfolder is a self-contained plugin with its own `plugin.py`. +Plugin collection for [Prax](https://github.com/praxagent/prax). Each subfolder is a self-contained plugin with its own `plugin.py`. ## Available plugins -| Plugin | Description | -|--------|-------------| -| [`pdf2presentation`](pdf2presentation/) | PDF → narrated video presentation (Beamer + TTS + ffmpeg) | -| [`flight_search`](flight_search/) | Search for the cheapest flights between airports (Amadeus API) | +| Plugin | Version | Description | +|--------|---------|-------------| +| [`pdf2presentation`](pdf2presentation/) | 4 | PDF → narrated video presentation (Beamer + TTS + ffmpeg) | +| [`flight_search`](flight_search/) | 1 | Search for the cheapest flights between airports (Amadeus API) | ## Installing plugins @@ -36,7 +36,43 @@ cd /path/to/prax/workspaces//plugins/shared/ git submodule add https://github.com/praxagent/prax-plugins.git prax-plugins ``` -### How it works +## Updating plugins + +Once installed, ask Prax to pull the latest version: + +> "Prax, please update the prax-plugins plugin" + +or more specifically: + +> "Update the pdf2presentation plugin" + +Prax runs `plugin_import_update("prax-plugins")` under the hood, which: + +1. Pulls the latest commit from this repo via `git submodule update --remote --merge` +2. Re-scans the updated code for security warnings +3. If clean, hot-reloads the plugin tools immediately — no restart needed +4. If new security concerns are found, shows them and waits for your confirmation + +You can also check the current plugin version at any time: + +> "What version of the pdf2presentation plugin am I running?" + +Prax will call `plugin_status("prax-plugins")` and show the active version, health status, and failure count. + +### Checking for updates manually + +If you prefer manual control: + +```bash +cd /path/to/prax/workspaces//plugins/shared/prax-plugins/ +git pull origin main +``` + +Then tell Prax to reload: + +> "Reload plugins" + +## How it works When you import a plugin repo, Prax: @@ -48,7 +84,7 @@ When you import a plugin repo, Prax: All plugin lifecycle events (import, activate, block, rollback, remove, security warnings) are recorded in the workspace trace log and searchable via `search_trace`. -#### Trust tiers +### Trust tiers Prax tags every plugin with a trust tier based on its origin: @@ -62,15 +98,13 @@ Imported plugins default to the least-trusted tier. Trust tiers are visible in ` When you import a specific subfolder from a multi-plugin repo, Prax writes a filter file (`.reponame_plugin_filter`) next to the submodule so only that subfolder's `plugin.py` is activated. The filter lives outside the submodule to avoid modifying its git working tree. -### Plugin structure +### Plugin failure tracking -Each plugin subfolder should contain: +Prax monitors every plugin tool invocation. If a tool fails 3 times consecutively, the plugin is automatically rolled back to its previous version. You'll see a message like: -``` -my-plugin/ -├── plugin.py # Required — tools + register() -└── README.md # Recommended — docs for users -``` +> "Plugin pdf2presentation auto-rolled back after 3 consecutive failures." + +You can check health status with `plugin_status` and manually roll back with `plugin_rollback` if needed. --- @@ -85,6 +119,15 @@ PDF → Markdown → Beamer LaTeX + speaker notes (LLM) → slide images → TTS | `pdf_to_presentation` | Full pipeline: PDF → narrated video (.mp4) | | `pdf_to_slides` | Lighter: PDF → Beamer slide deck + speaker notes (no video) | +### Input validation + +The plugin validates that the source is actually a PDF before processing: + +- **Content-Type check** — HTTP responses with `text/html` or other non-PDF content types are rejected immediately with a clear error message +- **Magic bytes check** — Downloaded files are verified to start with `%PDF`. HTML pages, JSON responses, and other non-PDF content are detected and rejected with guidance (e.g., "use fetch_url_content to extract text first") + +This prevents cryptic parser crashes when a URL returns an HTML page instead of a PDF. + ### Requirements **System dependencies:** @@ -153,7 +196,9 @@ Once installed, just talk to Prax: ### Architecture ``` -PDF file +PDF file / URL + │ + ├─ Content-Type + magic bytes validation │ ├─ opendataloader-pdf / pymupdf / pdftotext ▼ @@ -292,6 +337,32 @@ from prax.agent.llm_factory import build_llm # LLM from prax.settings import settings # Settings (NOT os.environ) ``` +--- + +## Development + +### Setup + +```bash +uv sync --extra dev +``` + +### Running tests + +```bash +uv run pytest tests/ -x -q +``` + +### Linting + +```bash +uv run ruff check . +``` + +### CI + +Pull requests run lint + tests automatically via GitHub Actions. Merges to `main` trigger [release-please](https://github.com/googleapis/release-please) for automated semantic versioning. + ## License Apache 2.0 diff --git a/flight_search/plugin.py b/flight_search/plugin.py index 80fd1fd..8022180 100644 --- a/flight_search/plugin.py +++ b/flight_search/plugin.py @@ -309,7 +309,7 @@ def flight_search( f"Date: {departure_date}" + (f" → {return_date}" if return_date else "") + f" | {adults} adult{'s' if adults > 1 else ''}" - + (f" | Nonstop only" if nonstop_only else "") + + (" | Nonstop only" if nonstop_only else "") + (f" | {cabin_class.replace('_', ' ').title()}" if cabin_class else "") + f"\n\nFound {len(offers)} option{'s' if len(offers) != 1 else ''} (cheapest first):\n" ) diff --git a/pdf2presentation/plugin.py b/pdf2presentation/plugin.py index 0eea2a1..715e051 100644 --- a/pdf2presentation/plugin.py +++ b/pdf2presentation/plugin.py @@ -19,7 +19,7 @@ """ from __future__ import annotations -PLUGIN_VERSION = "3" +PLUGIN_VERSION = "4" PLUGIN_DESCRIPTION = "Convert a PDF into a narrated video presentation" import json @@ -28,8 +28,6 @@ import re import subprocess import tempfile -import time -from pathlib import Path from langchain_core.tools import tool @@ -139,21 +137,66 @@ def _extract_text_from_pdf(pdf_path: str) -> str: def _download_pdf(url: str, dest_dir: str) -> str: - """Download a PDF from a URL. Returns the local file path.""" + """Download a PDF from a URL. Returns the local file path. + + Validates that the response is actually a PDF (by content-type header + and magic bytes) so that HTML pages, error pages, etc. are rejected + early with a clear error instead of crashing the PDF parser. + """ # Prefer Prax's PDF service which handles arXiv URLs etc. try: from prax.services.pdf_service import download_pdf - return download_pdf(url) - except Exception: + path = download_pdf(url) + _validate_pdf(path, url) + return path + except ImportError: pass - # Fallback: urllib. - import urllib.request + # Fallback: requests (preferred) or urllib. dest = os.path.join(dest_dir, "input.pdf") - urllib.request.urlretrieve(url, dest) + try: + import requests as _req + resp = _req.get(url, timeout=60, allow_redirects=True) + resp.raise_for_status() + content_type = resp.headers.get("Content-Type", "").lower() + if "html" in content_type: + raise ValueError( + f"URL returned HTML, not a PDF (Content-Type: {content_type}). " + f"If this is a web article, use fetch_url_content or web_summary_tool " + f"to extract the text first, then pass the text to pdf_to_slides." + ) + with open(dest, "wb") as f: + f.write(resp.content) + except ImportError: + import urllib.request + urllib.request.urlretrieve(url, dest) + + _validate_pdf(dest, url) return dest +def _validate_pdf(path: str, source_url: str = "") -> None: + """Check that a file is actually a PDF by reading its magic bytes.""" + try: + with open(path, "rb") as f: + header = f.read(16) + except OSError: + return # Can't read — let downstream handle it. + if not header.startswith(b"%PDF"): + # Try to detect what we actually got. + if header.startswith((b"=3.11" +license = "Apache-2.0" + +dependencies = [ + "langchain-core>=0.3", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "ruff>=0.9", + "requests>=2.31", +] + +[tool.ruff] +target-version = "py311" +line-length = 120 + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] +ignore = ["E501"] + +[tool.ruff.lint.per-file-ignores] +# Plugin files must declare PLUGIN_VERSION before imports — E402 is by design. +"*/plugin.py" = ["E402", "I001"] +# Test files need sys.modules patching before imports. +"tests/*.py" = ["E402", "I001"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_flight_search.py b/tests/test_flight_search.py new file mode 100644 index 0000000..5d409a4 --- /dev/null +++ b/tests/test_flight_search.py @@ -0,0 +1,252 @@ +"""Tests for the flight_search plugin. + +All Amadeus API calls are mocked. +""" +from __future__ import annotations + +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +# Mock Prax imports that flight_search may try. +sys.modules.setdefault("prax", MagicMock()) +sys.modules.setdefault("prax.settings", MagicMock()) + +from flight_search import plugin # noqa: E402 + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def _clear_token_cache(): + """Reset the token cache between tests.""" + plugin._TOKEN_CACHE["token"] = "" + plugin._TOKEN_CACHE["expires_at"] = 0.0 + + +# --------------------------------------------------------------------------- +# Input validation tests +# --------------------------------------------------------------------------- + +class TestInputValidation: + def test_invalid_departure_date(self): + result = plugin.flight_search.invoke({ + "origin": "JFK", + "destination": "CDG", + "departure_date": "March 15", + }) + assert "Invalid" in result + assert "YYYY-MM-DD" in result + + def test_invalid_return_date(self): + result = plugin.flight_search.invoke({ + "origin": "JFK", + "destination": "CDG", + "departure_date": "2026-03-15", + "return_date": "15-03-2026", + }) + assert "Invalid" in result + + def test_invalid_origin_code(self): + result = plugin.flight_search.invoke({ + "origin": "X", + "destination": "CDG", + "departure_date": "2026-03-15", + }) + assert "Invalid origin" in result + + def test_invalid_destination_code(self): + result = plugin.flight_search.invoke({ + "origin": "JFK", + "destination": "1234", + "departure_date": "2026-03-15", + }) + assert "Invalid destination" in result + + def test_origin_case_normalized(self): + """IATA codes should be uppercased.""" + with patch.object(plugin, "_search_flights", return_value=[]) as mock_search: + plugin.flight_search.invoke({ + "origin": "jfk", + "destination": "cdg", + "departure_date": "2026-03-15", + }) + mock_search.assert_called_once() + args = mock_search.call_args + assert args.kwargs["origin"] == "JFK" + assert args.kwargs["destination"] == "CDG" + + +# --------------------------------------------------------------------------- +# Duration formatting tests +# --------------------------------------------------------------------------- + +class TestFormatDuration: + def test_hours_and_minutes(self): + assert plugin._format_duration("PT2H30M") == "2h 30m" + + def test_hours_only(self): + assert plugin._format_duration("PT5H") == "5h" + + def test_minutes_only(self): + assert plugin._format_duration("PT45M") == "45m" + + def test_zero_minutes(self): + assert plugin._format_duration("PT0M") == "0m" + + +# --------------------------------------------------------------------------- +# Offer formatting tests +# --------------------------------------------------------------------------- + +class TestFormatOffer: + SAMPLE_OFFER = { + "price": {"grandTotal": "450.00", "currency": "USD"}, + "itineraries": [{ + "duration": "PT8H30M", + "segments": [{ + "departure": {"iataCode": "JFK", "at": "2026-03-15T08:00:00"}, + "arrival": {"iataCode": "CDG", "at": "2026-03-15T20:30:00"}, + "carrierCode": "AF", + "number": "123", + "duration": "PT8H30M", + }], + }], + "travelerPricings": [{ + "fareDetailsBySegment": [{"cabin": "ECONOMY"}], + }], + } + + def test_formats_price(self): + result = plugin._format_offer(self.SAMPLE_OFFER, 1) + assert "USD 450.00" in result + + def test_formats_route(self): + result = plugin._format_offer(self.SAMPLE_OFFER, 1) + assert "JFK" in result + assert "CDG" in result + assert "AF123" in result + + def test_formats_cabin(self): + result = plugin._format_offer(self.SAMPLE_OFFER, 1) + assert "Economy" in result + + def test_formats_nonstop(self): + result = plugin._format_itinerary(self.SAMPLE_OFFER["itineraries"][0], "Outbound") + assert "nonstop" in result + + def test_formats_with_stop(self): + two_seg = { + "duration": "PT12H", + "segments": [ + {"departure": {"iataCode": "JFK", "at": "2026-03-15T08:00"}, + "arrival": {"iataCode": "LHR", "at": "2026-03-15T14:00"}, + "carrierCode": "BA", "number": "1", "duration": "PT6H"}, + {"departure": {"iataCode": "LHR", "at": "2026-03-15T16:00"}, + "arrival": {"iataCode": "CDG", "at": "2026-03-15T20:00"}, + "carrierCode": "BA", "number": "2", "duration": "PT4H"}, + ], + } + result = plugin._format_itinerary(two_seg, "Outbound") + assert "1 stop" in result + + +# --------------------------------------------------------------------------- +# No results formatting +# --------------------------------------------------------------------------- + +class TestNoResults: + def test_no_flights_one_way(self): + with patch.object(plugin, "_search_flights", return_value=[]): + result = plugin.flight_search.invoke({ + "origin": "JFK", + "destination": "CDG", + "departure_date": "2026-03-15", + }) + assert "No flights found" in result + assert "one-way" in result + + def test_no_flights_round_trip(self): + with patch.object(plugin, "_search_flights", return_value=[]): + result = plugin.flight_search.invoke({ + "origin": "JFK", + "destination": "CDG", + "departure_date": "2026-03-15", + "return_date": "2026-03-22", + }) + assert "No flights found" in result + assert "round-trip" in result + + +# --------------------------------------------------------------------------- +# API error handling +# --------------------------------------------------------------------------- + +class TestApiErrors: + def test_missing_credentials(self, monkeypatch): + monkeypatch.delenv("AMADEUS_API_KEY", raising=False) + monkeypatch.delenv("AMADEUS_API_SECRET", raising=False) + # Also ensure Prax settings fallback is empty. + mock_settings = MagicMock() + mock_settings.amadeus_api_key = "" + mock_settings.amadeus_api_secret = "" + with patch.dict(sys.modules, {"prax.settings": MagicMock(settings=mock_settings)}): + result = plugin.flight_search.invoke({ + "origin": "JFK", + "destination": "CDG", + "departure_date": "2026-03-15", + }) + assert "credentials" in result.lower() or "failed" in result.lower() + + def test_api_failure_returns_error(self): + with patch.object(plugin, "_search_flights", side_effect=Exception("API down")): + result = plugin.flight_search.invoke({ + "origin": "JFK", + "destination": "CDG", + "departure_date": "2026-03-15", + }) + assert "failed" in result.lower() + + +# --------------------------------------------------------------------------- +# Airport lookup tests +# --------------------------------------------------------------------------- + +class TestAirportLookup: + def test_formats_results(self): + mock_results = [ + {"iata": "CDG", "name": "Charles de Gaulle", "city": "Paris", "country": "FR"}, + {"iata": "ORY", "name": "Orly", "city": "Paris", "country": "FR"}, + ] + with patch.object(plugin, "_search_airports", return_value=mock_results): + result = plugin.airport_lookup.invoke({"query": "Paris"}) + assert "CDG" in result + assert "ORY" in result + assert "Charles de Gaulle" in result + + def test_no_results(self): + with patch.object(plugin, "_search_airports", return_value=[]): + result = plugin.airport_lookup.invoke({"query": "xyznoexist"}) + assert "No airports found" in result + + +# --------------------------------------------------------------------------- +# Registration tests +# --------------------------------------------------------------------------- + +class TestRegistration: + def test_register_returns_tools(self): + tools = plugin.register() + assert len(tools) == 2 + names = {t.name for t in tools} + assert "flight_search" in names + assert "airport_lookup" in names + + def test_plugin_version(self): + assert plugin.PLUGIN_VERSION == "1" diff --git a/tests/test_pdf2presentation.py b/tests/test_pdf2presentation.py new file mode 100644 index 0000000..645b1bc --- /dev/null +++ b/tests/test_pdf2presentation.py @@ -0,0 +1,336 @@ +"""Tests for the pdf2presentation plugin. + +All external calls (LLM, TTS, system commands, Prax services) are mocked. +""" +from __future__ import annotations + +import json +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + +# Add the repo root to sys.path so we can import plugins directly. +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +# We need to mock Prax imports before importing the plugin module, +# because it tries `from prax.utils.shell import ...` at module level. +sys.modules["prax"] = MagicMock() +sys.modules["prax.utils"] = MagicMock() +sys.modules["prax.utils.shell"] = MagicMock() +sys.modules["prax.services"] = MagicMock() +sys.modules["prax.services.pdf_service"] = MagicMock() +sys.modules["prax.agent"] = MagicMock() +sys.modules["prax.agent.llm_factory"] = MagicMock() +sys.modules["prax.agent.user_context"] = MagicMock() +sys.modules["prax.services.workspace_service"] = MagicMock() +sys.modules["prax.settings"] = MagicMock() + +from pdf2presentation import plugin # noqa: E402 + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture() +def tmp_work_dir(tmp_path): + return str(tmp_path) + + +@pytest.fixture() +def fake_pdf(tmp_path): + """Create a minimal valid PDF file.""" + pdf = tmp_path / "test.pdf" + pdf.write_bytes(b"%PDF-1.4 fake pdf content for testing") + return str(pdf) + + +@pytest.fixture() +def fake_html(tmp_path): + """Create a file with HTML content (not a PDF).""" + html = tmp_path / "not_a_pdf.pdf" + html.write_bytes(b"Not a PDF") + return str(html) + + +@pytest.fixture() +def fake_json_file(tmp_path): + """Create a file with JSON content (not a PDF).""" + j = tmp_path / "response.pdf" + j.write_bytes(b'{"error": "not found"}') + return str(j) + + +# --------------------------------------------------------------------------- +# _validate_pdf tests +# --------------------------------------------------------------------------- + +class TestValidatePdf: + def test_valid_pdf_passes(self, fake_pdf): + # Should not raise. + plugin._validate_pdf(fake_pdf) + + def test_html_file_raises(self, fake_html): + with pytest.raises(ValueError, match="HTML page"): + plugin._validate_pdf(fake_html) + + def test_json_file_raises(self, fake_json_file): + with pytest.raises(ValueError, match="JSON"): + plugin._validate_pdf(fake_json_file) + + def test_unknown_binary_raises(self, tmp_path): + f = tmp_path / "garbage.pdf" + f.write_bytes(b"\x00\x01\x02\x03\x04\x05\x06\x07") + with pytest.raises(ValueError, match="does not start with %PDF"): + plugin._validate_pdf(str(f)) + + def test_includes_url_in_error(self, fake_html): + with pytest.raises(ValueError, match="example.com"): + plugin._validate_pdf(str(fake_html), source_url="https://example.com/page") + + def test_nonexistent_file_does_not_raise(self): + # Can't read — let downstream handle it. + plugin._validate_pdf("/nonexistent/file.pdf") + + def test_empty_file_raises(self, tmp_path): + f = tmp_path / "empty.pdf" + f.write_bytes(b"") + with pytest.raises(ValueError, match="does not start with %PDF"): + plugin._validate_pdf(str(f)) + + +# --------------------------------------------------------------------------- +# _download_pdf tests +# --------------------------------------------------------------------------- + +class TestDownloadPdf: + """Test the requests-based download path. + + The Prax PDF service is mocked at module level, so we force it to raise + ImportError to exercise the requests fallback. + """ + + @staticmethod + def _make_response(content_type: str, content: bytes): + resp = MagicMock() + resp.headers = {"Content-Type": content_type} + resp.status_code = 200 + resp.content = content + resp.raise_for_status = MagicMock() + return resp + + @staticmethod + def _patch_prax_unavailable(): + """Make the Prax PDF service raise ImportError so requests path is used.""" + mock_mod = MagicMock() + mock_mod.download_pdf.side_effect = ImportError("no prax") + return patch.dict(sys.modules, {"prax.services.pdf_service": mock_mod}) + + def test_rejects_html_response(self, tmp_work_dir): + """A URL returning HTML should fail with a clear error.""" + import requests as _req_mod + fake_resp = self._make_response("text/html; charset=utf-8", b"Not a PDF") + + with self._patch_prax_unavailable(), \ + patch.object(_req_mod, "get", return_value=fake_resp): + with pytest.raises(ValueError, match="HTML"): + plugin._download_pdf("https://example.com/page.html", tmp_work_dir) + + def test_accepts_pdf_response(self, tmp_work_dir): + """A URL returning application/pdf should succeed.""" + import requests as _req_mod + fake_resp = self._make_response("application/pdf", b"%PDF-1.4 test content") + + with self._patch_prax_unavailable(), \ + patch.object(_req_mod, "get", return_value=fake_resp): + path = plugin._download_pdf("https://example.com/paper.pdf", tmp_work_dir) + assert os.path.isfile(path) + with open(path, "rb") as f: + assert f.read().startswith(b"%PDF") + + def test_accepts_octet_stream(self, tmp_work_dir): + """application/octet-stream with valid PDF content should succeed.""" + import requests as _req_mod + fake_resp = self._make_response("application/octet-stream", b"%PDF-1.4 binary pdf") + + with self._patch_prax_unavailable(), \ + patch.object(_req_mod, "get", return_value=fake_resp): + path = plugin._download_pdf("https://example.com/file", tmp_work_dir) + assert os.path.isfile(path) + + def test_rejects_html_body_even_with_pdf_content_type(self, tmp_work_dir): + """Even if content-type says PDF, validate magic bytes.""" + import requests as _req_mod + fake_resp = self._make_response("application/pdf", b"") + + with self._patch_prax_unavailable(), \ + patch.object(_req_mod, "get", return_value=fake_resp): + with pytest.raises(ValueError, match="HTML"): + plugin._download_pdf("https://example.com/fake.pdf", tmp_work_dir) + + +# --------------------------------------------------------------------------- +# _resolve_pdf tests +# --------------------------------------------------------------------------- + +class TestResolvePdf: + def test_local_file(self, fake_pdf): + path = plugin._resolve_pdf(fake_pdf, "/tmp") + assert path == fake_pdf + + def test_file_not_found(self): + with pytest.raises(FileNotFoundError, match="not found"): + plugin._resolve_pdf("nonexistent_file.pdf", "/tmp") + + def test_url_triggers_download(self, tmp_work_dir): + """HTTP URLs should go through _download_pdf.""" + with patch.object(plugin, "_download_pdf", return_value="/tmp/downloaded.pdf") as mock_dl: + path = plugin._resolve_pdf("https://example.com/paper.pdf", tmp_work_dir) + mock_dl.assert_called_once_with("https://example.com/paper.pdf", tmp_work_dir) + assert path == "/tmp/downloaded.pdf" + + +# --------------------------------------------------------------------------- +# _generate_beamer_and_notes tests +# --------------------------------------------------------------------------- + +class TestGenerateBeamerAndNotes: + VALID_LLM_RESPONSE = json.dumps({ + "title": "Test Presentation", + "author": "Test Author", + "latex": "\\documentclass{beamer}\n\\begin{document}\n\\end{document}", + "slides": [ + {"title": "Intro", "notes": "Welcome to the presentation."}, + {"title": "Main Point", "notes": "Here is the key insight."}, + ], + }) + + @staticmethod + def _patch_llm(response_content: str): + """Patch the LLM factory to return a mock that produces the given content.""" + mock_llm = MagicMock() + mock_llm.invoke.return_value = MagicMock(content=response_content) + mock_factory = MagicMock() + mock_factory.build_llm.return_value = mock_llm + return patch.dict(sys.modules, {"prax.agent.llm_factory": mock_factory}) + + def test_parses_valid_json(self): + mock_llm = MagicMock() + mock_llm.invoke.return_value = MagicMock(content=self.VALID_LLM_RESPONSE) + mock_factory = MagicMock() + mock_factory.build_llm.return_value = mock_llm + + with patch.dict(sys.modules, {"prax.agent.llm_factory": mock_factory}): + result = plugin._generate_beamer_and_notes("sample text", "Test", "academic") + + assert result["title"] == "Test Presentation" + assert len(result["slides"]) == 2 + assert "beamer" in result["latex"] + + def test_strips_markdown_fences(self): + fenced = f"```json\n{self.VALID_LLM_RESPONSE}\n```" + mock_llm = MagicMock() + mock_llm.invoke.return_value = MagicMock(content=fenced) + mock_factory = MagicMock() + mock_factory.build_llm.return_value = mock_llm + + with patch.dict(sys.modules, {"prax.agent.llm_factory": mock_factory}): + result = plugin._generate_beamer_and_notes("text", "", "academic") + + assert result["title"] == "Test Presentation" + + def test_strips_surrounding_text(self): + wrapped = f"Here is the JSON:\n{self.VALID_LLM_RESPONSE}\nHope this helps!" + mock_llm = MagicMock() + mock_llm.invoke.return_value = MagicMock(content=wrapped) + mock_factory = MagicMock() + mock_factory.build_llm.return_value = mock_llm + + with patch.dict(sys.modules, {"prax.agent.llm_factory": mock_factory}): + result = plugin._generate_beamer_and_notes("text", "", "academic") + + assert result["title"] == "Test Presentation" + + def test_rejects_missing_keys(self): + bad_json = json.dumps({"title": "No slides key"}) + mock_llm = MagicMock() + mock_llm.invoke.return_value = MagicMock(content=bad_json) + mock_factory = MagicMock() + mock_factory.build_llm.return_value = mock_llm + + with patch.dict(sys.modules, {"prax.agent.llm_factory": mock_factory}): + with pytest.raises(ValueError, match="missing"): + plugin._generate_beamer_and_notes("text", "", "academic") + + +# --------------------------------------------------------------------------- +# _check_system_deps tests +# --------------------------------------------------------------------------- + +class TestCheckSystemDeps: + def test_all_present(self): + with patch.object(plugin, "_which_cmd", return_value=True): + assert plugin._check_system_deps(need_ffmpeg=True) == [] + + def test_missing_pdflatex(self): + def fake_which(name): + return name != "pdflatex" + + with patch.object(plugin, "_which_cmd", side_effect=fake_which): + missing = plugin._check_system_deps(need_ffmpeg=False) + assert "pdflatex" in missing + assert "ffmpeg" not in missing + + def test_skip_ffmpeg_check(self): + def fake_which(name): + return name not in ("ffmpeg", "ffprobe") + + with patch.object(plugin, "_which_cmd", side_effect=fake_which): + missing = plugin._check_system_deps(need_ffmpeg=False) + assert "ffmpeg" not in missing + + +# --------------------------------------------------------------------------- +# TTS config tests +# --------------------------------------------------------------------------- + +class TestTtsConfig: + def test_default_openai(self, monkeypatch): + monkeypatch.delenv("PRESENTATION_TTS_PROVIDER", raising=False) + monkeypatch.delenv("PRESENTATION_TTS_VOICE", raising=False) + monkeypatch.setenv("OPENAI_KEY", "sk-test") + cfg = plugin._get_tts_config() + assert cfg["provider"] == "openai" + assert cfg["voice"] == "nova" + assert cfg["api_key"] == "sk-test" + + def test_elevenlabs(self, monkeypatch): + monkeypatch.setenv("PRESENTATION_TTS_PROVIDER", "elevenlabs") + monkeypatch.setenv("PRESENTATION_TTS_VOICE", "Adam") + monkeypatch.setenv("ELEVENLABS_API_KEY", "el-test") + cfg = plugin._get_tts_config() + assert cfg["provider"] == "elevenlabs" + assert cfg["voice"] == "Adam" + assert cfg["api_key"] == "el-test" + + +# --------------------------------------------------------------------------- +# Plugin registration tests +# --------------------------------------------------------------------------- + +class TestRegistration: + def test_register_returns_tools(self): + tools = plugin.register() + assert len(tools) == 2 + names = {t.name for t in tools} + assert "pdf_to_presentation" in names + assert "pdf_to_slides" in names + + def test_plugin_version(self): + assert plugin.PLUGIN_VERSION == "4" + + def test_plugin_description(self): + assert plugin.PLUGIN_DESCRIPTION + assert isinstance(plugin.PLUGIN_DESCRIPTION, str)