Expand test suite

Author: gleachkr

tests/test.db

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from pdf2sqlite.mcp_server.config import ServerConfig
+
+
+def test_from_cli_reads_env_when_database_missing(tmp_path, monkeypatch):
+    db = tmp_path / "t.db"
+    db.write_bytes(b"SQLite format 3\0")
+    monkeypatch.setenv("PDF2SQLITE_MCP_DATABASE", str(db))
+
+    cfg = ServerConfig.from_cli(database=None)
+
+    assert cfg.database_path == db.resolve()
+
+
+def test_from_cli_validates_limits(tmp_path, monkeypatch):
+    db = tmp_path / "t.db"
+    db.write_bytes(b"SQLite format 3\0")
+
+    # positive defaults
+    cfg = ServerConfig.from_cli(str(db), default_limit=10, max_limit=20)
+    assert cfg.clamp_limit(None) == 10
+
+    # default limit > max limit is rejected
+    with pytest.raises(ValueError):
+        ServerConfig.from_cli(str(db), default_limit=30, max_limit=20)
+
+    # zero means default; values over max are rejected
+    assert cfg.clamp_limit(0) == 10
+    with pytest.raises(ValueError):
+        cfg.clamp_limit(9999)
+
+
+def test_from_cli_env_overrides(tmp_path, monkeypatch):
+    db = tmp_path / "t.db"
+    db.write_bytes(b"SQLite format 3\0")
+
+    monkeypatch.setenv("PDF2SQLITE_MCP_MAX_BLOB_BYTES", "1024")
+    monkeypatch.setenv("PDF2SQLITE_MCP_DEFAULT_LIMIT", "5")
+    monkeypatch.setenv("PDF2SQLITE_MCP_MAX_LIMIT", "6")
+
+    cfg = ServerConfig.from_cli(str(db))
+
+    assert cfg.max_blob_bytes == 1024
+    assert cfg.default_limit == 5
+    assert cfg.max_limit == 6
+
+
+def test_from_cli_rejects_missing_database(monkeypatch):
+    monkeypatch.delenv("PDF2SQLITE_MCP_DATABASE", raising=False)
+    with pytest.raises(ValueError):
+        ServerConfig.from_cli(database=None)

tests/test_mcp_config.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from pdf2sqlite.mcp_server.db import Database
+
+
+TEST_DB = Path("tests/test.db").resolve()
+
+
+@pytest.mark.skipif(not TEST_DB.exists(), reason="tests/test.db missing")
+def test_db_pdf_counts_pages_and_blobs():
+    db = Database(TEST_DB)
+
+    pdfs = asyncio_run(db.get_pdf_counts(10, 0))
+    assert isinstance(pdfs, list)
+    assert pdfs, "expected at least one pdf in tests/test.db"
+
+    first_pdf_id = int(pdfs[0]["id"])  # type: ignore[index]
+
+    pages = asyncio_run(db.get_pdf_pages(first_pdf_id, 1, 0))
+    assert pages, "expected at least one page for first pdf"
+
+    page = pages[0]
+    page_id = int(page["id"])  # type: ignore[index]
+    page_num = int(page["page_number"])  # type: ignore[index]
+
+    blob = asyncio_run(db.get_page_blob(first_pdf_id, page_num))
+    assert isinstance(blob, (bytes, bytearray))
+    assert blob.startswith(b"%PDF"), "page blob should be a PDF"
+
+    blob2 = asyncio_run(db.get_page_blob_by_id(page_id))
+    assert blob2.startswith(b"%PDF")
+
+
+@pytest.mark.skipif(not TEST_DB.exists(), reason="tests/test.db missing")
+def test_db_schema_nonempty():
+    db = Database(TEST_DB)
+
+    stmts = asyncio_run(db.get_schema())
+    assert any("CREATE TABLE" in s for s in stmts)
+    assert any("pdfs" in s for s in stmts)
+
+
+@pytest.mark.skipif(not TEST_DB.exists(), reason="tests/test.db missing")
+def test_db_page_assets_optional():
+    db = Database(TEST_DB)
+
+    pdfs = asyncio_run(db.get_pdf_counts(1, 0))
+    if not pdfs:
+        pytest.skip("no pdfs present in test db")
+    pdf_id = int(pdfs[0]["id"])  # type: ignore[index]
+
+    pages = asyncio_run(db.get_pdf_pages(pdf_id, 1, 0))
+    if not pages:
+        pytest.skip("no pages present in test db")
+    page_id = int(pages[0]["id"])  # type: ignore[index]
+
+    figures = asyncio_run(db.get_figures_for_page(page_id))
+    tables = asyncio_run(db.get_tables_for_page(page_id))
+
+    assert isinstance(figures, list)
+    assert isinstance(tables, list)
+
+    if figures:
+        fid = int(figures[0]["id"])  # type: ignore[index]
+        blob, mime = asyncio_run(db.get_figure_blob(fid))
+        assert isinstance(blob, (bytes, bytearray))
+        assert mime is None or isinstance(mime, str)
+
+    if tables:
+        tid = int(tables[0]["id"])  # type: ignore[index]
+        img = asyncio_run(db.get_table_image_blob(tid))
+        assert isinstance(img, (bytes, bytearray))
+
+
+# helpers
+import asyncio
+
+
+def asyncio_run(awaitable):
+    return asyncio.get_event_loop().run_until_complete(awaitable)

tests/test_mcp_db.py

@@ -1,17 +1,9 @@
 from __future__ import annotations
 
 from pathlib import Path
-import sys
 
 import pytest
 
-ROOT = Path(__file__).resolve().parents[1]
-SITE_PACKAGES = ROOT / ".venv" / "lib" / "python3.13" / "site-packages"
-if SITE_PACKAGES.exists():
-    sys.path.append(str(SITE_PACKAGES))
-
-pytest.importorskip("mcp.server.fastmcp")
-
 from pdf2sqlite.mcp_server.config import ServerConfig
 from pdf2sqlite.mcp_server.server import build_server
 

tests/test_mcp_resources.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from pdf2sqlite.mcp_server.config import ServerConfig
+from pdf2sqlite.mcp_server.db import Database
+from pdf2sqlite.mcp_server.resources import (
+    ResourceService,
+    ResourceTooLargeError,
+    build_page_payload,
+)
+from mcp.server.fastmcp.utilities.types import Image as MCPImage
+from pdf2sqlite.mcp_server.uri import PdfResource
+
+
+TEST_DB = Path("tests/test.db").resolve()
+
+
+@pytest.mark.skipif(not TEST_DB.exists(), reason="tests/test.db missing")
+def test_resource_service_pdf_blobs_and_limits():
+    # tiny limit to force overflow on any real PDF blob
+    cfg = ServerConfig(
+        database_path=TEST_DB,
+        max_blob_bytes=1,
+        default_limit=10,
+        max_limit=10,
+    )
+    db = Database(TEST_DB)
+    svc = ResourceService(database=db, config=cfg)
+
+    # pick first pdf id
+    pdfs = asyncio_run(db.get_pdf_counts(1, 0))
+    if not pdfs:
+        pytest.skip("no pdfs present in test db")
+    pdf_id = int(pdfs[0]["id"])  # type: ignore[index]
+
+    # page blob should exceed limit and raise
+    page = asyncio_run(db.get_pdf_pages(pdf_id, 1, 0))[0]
+    page_num = int(page["page_number"])  # type: ignore[index]
+
+    with pytest.raises(ResourceTooLargeError):
+        asyncio_run(svc.load_pdf_blob(PdfResource(pdf_id, page_num)))
+
+    # bump limit so full-pdf works and returns a valid PDF
+    cfg.max_blob_bytes = 10_000_000
+    data = asyncio_run(svc.load_pdf_blob(PdfResource(pdf_id)))
+    assert data.startswith(b"%PDF")
+
+    embed = asyncio_run(svc.make_embedded_pdf(
+        f"pdf2sqlite://pdf/{pdf_id}", data
+    ))
+    assert embed.resource.mimeType == "application/pdf"
+    assert embed.resource.meta["size"] == len(data)
+
+    # as_image returns an MCP Image object
+    img = svc.as_image(b"x", "image/png")
+    assert isinstance(img, MCPImage)
+
+
+def test_build_page_payload_validates_fields():
+    # missing required id
+    with pytest.raises(ValueError):
+        build_page_payload({"pdf_id": 1, "page_number": 1})
+    # non-int page number
+    with pytest.raises(ValueError):
+        build_page_payload({"id": 1, "pdf_id": 1, "page_number": "x"})
+
+
+# helpers
+import asyncio
+
+
+def asyncio_run(awaitable):
+    return asyncio.get_event_loop().run_until_complete(awaitable)

tests/test_mcp_resources_unit.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import pytest
+
+from pdf2sqlite.mcp_server.tools import _pdf_summary_block
+from pdf2sqlite.mcp_server.uri import PdfResource
+
+
+def test_pdf_summary_block_for_full_pdf():
+    block = _pdf_summary_block(
+        "pdf2sqlite://pdf/1",
+        1234,
+        PdfResource(1),
+    )
+    assert "PDF 1" in block.text
+    assert "Bytes: 1234" in block.text
+
+
+def test_pdf_summary_block_for_single_page():
+    block = _pdf_summary_block(
+        "pdf2sqlite://pdf/2/page/5",
+        77,
+        PdfResource(2, 5),
+    )
+    assert "PDF 2 page 5" in block.text
+    assert "Bytes: 77" in block.text

tests/test_mcp_tools_unit.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import pytest
+
+from pdf2sqlite.mcp_server.uri import (
+    PdfResource,
+    FigureResource,
+    TableImageResource,
+    build_pdf_uri,
+    build_pdf_page_uri,
+    build_figure_uri,
+    build_table_image_uri,
+    parse_resource_uri,
+)
+
+
+def test_parse_pdf_uri_full_document():
+    uri = "pdf2sqlite://pdf/123"
+    desc = parse_resource_uri(uri)
+    assert isinstance(desc, PdfResource)
+    assert desc.pdf_id == 123
+    assert desc.page_number is None
+    assert build_pdf_uri(123) == uri
+
+
+def test_parse_pdf_uri_single_page():
+    uri = "pdf2sqlite://pdf/456/page/7"
+    desc = parse_resource_uri(uri)
+    assert isinstance(desc, PdfResource)
+    assert desc.pdf_id == 456
+    assert desc.page_number == 7
+    assert build_pdf_page_uri(456, 7) == uri
+
+
+def test_parse_figure_uri():
+    uri = "pdf2sqlite://figure/999"
+    desc = parse_resource_uri(uri)
+    assert isinstance(desc, FigureResource)
+    assert desc.figure_id == 999
+    assert build_figure_uri(999) == uri
+
+
+def test_parse_table_image_uri():
+    uri = "pdf2sqlite://table-image/42"
+    desc = parse_resource_uri(uri)
+    assert isinstance(desc, TableImageResource)
+    assert desc.table_id == 42
+    assert build_table_image_uri(42) == uri
+
+
+def test_parse_uri_rejects_bad_scheme():
+    with pytest.raises(ValueError):
+        parse_resource_uri("http://pdf/1")
+
+
+def test_parse_uri_rejects_missing_target():
+    with pytest.raises(ValueError):
+        parse_resource_uri("pdf2sqlite://")
+
+
+def test_parse_uri_rejects_malformed_pdf_paths():
+    with pytest.raises(ValueError):
+        parse_resource_uri("pdf2sqlite://pdf/")
+    with pytest.raises(ValueError):
+        parse_resource_uri("pdf2sqlite://pdf/123/page/")
+    with pytest.raises(ValueError):
+        parse_resource_uri("pdf2sqlite://pdf/123/x/1")
+
+
+def test_parse_uri_rejects_non_int_identifiers():
+    with pytest.raises(ValueError):
+        parse_resource_uri("pdf2sqlite://pdf/abc")
+    with pytest.raises(ValueError):
+        parse_resource_uri("pdf2sqlite://pdf/1/page/two")
+    with pytest.raises(ValueError):
+        parse_resource_uri("pdf2sqlite://figure/notanint")
+    with pytest.raises(ValueError):
+        parse_resource_uri("pdf2sqlite://table-image/notanint")