test: add unit tests for RAG, document_reader, signal_loader, CLI, models, DOCX reports

Also fix chunk_text() infinite loop when chunk_overlap >= chunk_size (clamp overlap),
and skip parquet test when pyarrow/fastparquet is not installed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: LGDiMaggio

src/rag.py

@@ -69,6 +69,8 @@ def chunk_text(
       - ``index`` : sequential chunk number
       - ``start`` / ``end`` : character offsets in the original text
     """
+    if chunk_overlap >= chunk_size:
+        chunk_overlap = max(0, chunk_size - 1)
     chunks: list[dict[str, Any]] = []
     start = 0
     idx = 0

tests/test_cli_transport.py

@@ -0,0 +1,155 @@
+"""
+Tests for CLI argument parsing and transport configuration.
+
+Covers:
+- argparse defaults (transport=stdio, host=127.0.0.1, port=8000)
+- Explicit flag values (--transport sse, --host, --port)
+- Short flag aliases (-t, -p)
+- Environment variable fallbacks (MCP_TRANSPORT, MCP_HOST, MCP_PORT)
+- Invalid transport choice → error
+- Invalid port (non-numeric) → error
+- CLI args override env vars
+"""
+
+import argparse
+import os
+import pytest
+from unittest.mock import patch, MagicMock
+
+
+def _build_parser(env_transport="stdio", env_host="127.0.0.1", env_port="8000"):
+    """Replicate the argparse setup from main() for isolated testing."""
+    parser = argparse.ArgumentParser(description="Predictive Maintenance MCP Server")
+    parser.add_argument(
+        "--transport", "-t",
+        choices=["stdio", "sse", "streamable-http"],
+        default=env_transport,
+        help="Transport protocol",
+    )
+    parser.add_argument(
+        "--host",
+        default=env_host,
+        help="Bind address for SSE/HTTP",
+    )
+    parser.add_argument(
+        "--port", "-p",
+        type=int,
+        default=int(env_port),
+        help="Port for SSE/HTTP transport",
+    )
+    return parser
+
+
+# ── Default values ─────────────────────────────────────────────────────────
+
+class TestCLIDefaults:
+
+    def test_default_transport_stdio(self):
+        args = _build_parser().parse_args([])
+        assert args.transport == "stdio"
+
+    def test_default_host(self):
+        args = _build_parser().parse_args([])
+        assert args.host == "127.0.0.1"
+
+    def test_default_port(self):
+        args = _build_parser().parse_args([])
+        assert args.port == 8000
+
+
+# ── Explicit flags ─────────────────────────────────────────────────────────
+
+class TestCLIExplicitFlags:
+
+    def test_transport_sse(self):
+        args = _build_parser().parse_args(["--transport", "sse"])
+        assert args.transport == "sse"
+
+    def test_transport_streamable_http(self):
+        args = _build_parser().parse_args(["--transport", "streamable-http"])
+        assert args.transport == "streamable-http"
+
+    def test_host_custom(self):
+        args = _build_parser().parse_args(["--host", "0.0.0.0"])
+        assert args.host == "0.0.0.0"
+
+    def test_port_custom(self):
+        args = _build_parser().parse_args(["--port", "9090"])
+        assert args.port == 9090
+
+    def test_short_flag_transport(self):
+        args = _build_parser().parse_args(["-t", "sse"])
+        assert args.transport == "sse"
+
+    def test_short_flag_port(self):
+        args = _build_parser().parse_args(["-p", "7777"])
+        assert args.port == 7777
+
+    def test_all_flags_combined(self):
+        args = _build_parser().parse_args([
+            "-t", "sse", "--host", "10.0.0.1", "-p", "3000"
+        ])
+        assert args.transport == "sse"
+        assert args.host == "10.0.0.1"
+        assert args.port == 3000
+
+
+# ── Invalid arguments ──────────────────────────────────────────────────────
+
+class TestCLIInvalidArgs:
+
+    def test_invalid_transport_rejected(self):
+        with pytest.raises(SystemExit):
+            _build_parser().parse_args(["--transport", "websocket"])
+
+    def test_non_numeric_port_rejected(self):
+        with pytest.raises(SystemExit):
+            _build_parser().parse_args(["--port", "abc"])
+
+
+# ── Environment variable fallbacks ─────────────────────────────────────────
+
+class TestCLIEnvVars:
+
+    def test_mcp_transport_env(self, monkeypatch):
+        monkeypatch.setenv("MCP_TRANSPORT", "sse")
+        env_t = os.environ.get("MCP_TRANSPORT", "stdio")
+        args = _build_parser(env_transport=env_t).parse_args([])
+        assert args.transport == "sse"
+
+    def test_mcp_host_env(self, monkeypatch):
+        monkeypatch.setenv("MCP_HOST", "192.168.1.100")
+        env_h = os.environ.get("MCP_HOST", "127.0.0.1")
+        args = _build_parser(env_host=env_h).parse_args([])
+        assert args.host == "192.168.1.100"
+
+    def test_mcp_port_env(self, monkeypatch):
+        monkeypatch.setenv("MCP_PORT", "5555")
+        env_p = os.environ.get("MCP_PORT", "8000")
+        args = _build_parser(env_port=env_p).parse_args([])
+        assert args.port == 5555
+
+    def test_cli_overrides_env(self, monkeypatch):
+        monkeypatch.setenv("MCP_TRANSPORT", "sse")
+        monkeypatch.setenv("MCP_PORT", "5555")
+        env_t = os.environ.get("MCP_TRANSPORT", "stdio")
+        env_p = os.environ.get("MCP_PORT", "8000")
+        args = _build_parser(env_transport=env_t, env_port=env_p).parse_args(
+            ["--transport", "streamable-http", "--port", "9999"]
+        )
+        assert args.transport == "streamable-http"
+        assert args.port == 9999
+
+
+# ── main() integration (smoke test) ───────────────────────────────────────
+
+def test_main_help_exits_cleanly():
+    """Verify --help works without crashing (exercises the real main parser)."""
+    import sys
+    from unittest.mock import patch as _patch
+
+    with _patch.object(sys, 'argv', ['test', '--help']):
+        with pytest.raises(SystemExit) as exc_info:
+            from predictive_maintenance_mcp.machinery_diagnostics_server import main
+            main()
+        assert exc_info.value.code == 0

tests/test_document_reader.py

@@ -0,0 +1,163 @@
+"""
+Tests for the Document Reader module.
+
+Covers:
+- Bearing characteristic frequency calculations (BPFO, BPFI, BSF, FTF)
+- PDF text extraction (basic, missing file, OCR fallback)
+- Edge cases: zero contact angle, high contact angle, zero RPM
+"""
+
+import math
+import pytest
+from pathlib import Path
+
+from predictive_maintenance_mcp.document_reader import (
+    calculate_bearing_frequencies,
+    extract_text_from_pdf,
+    HAS_PDF,
+    HAS_OCR,
+)
+
+
+# ── Bearing frequency calculations ─────────────────────────────────────────
+
+class TestBearingFrequencies:
+    """Tests against known bearing geometry values."""
+
+    def test_standard_deep_groove_6205(self):
+        """SKF 6205: Z=9, Bd=7.94mm, Pd=34.55mm, alpha=0, RPM=1797."""
+        freqs = calculate_bearing_frequencies(
+            num_balls=9,
+            ball_diameter_mm=7.94,
+            pitch_diameter_mm=34.55,
+            contact_angle_deg=0.0,
+            shaft_speed_rpm=1797,
+        )
+        shaft_freq = 1797 / 60.0  # 29.95 Hz
+
+        assert "BPFO" in freqs
+        assert "BPFI" in freqs
+        assert "BSF" in freqs
+        assert "FTF" in freqs
+        assert "shaft_freq_hz" in freqs
+        assert abs(freqs["shaft_freq_hz"] - shaft_freq) < 0.01
+
+        # BPFO ~ (Z/2)*fs*(1 - Bd/Pd) = 4.5 * 29.95 * (1-0.2298) ≈ 103.8
+        assert 80 < freqs["BPFO"] < 110
+        # BPFI > BPFO always (inner race spins faster relative to rolling elements)
+        assert freqs["BPFI"] > freqs["BPFO"]
+        # FTF < shaft frequency
+        assert freqs["FTF"] < shaft_freq
+
+    def test_all_frequencies_positive(self):
+        freqs = calculate_bearing_frequencies(
+            num_balls=8,
+            ball_diameter_mm=12.0,
+            pitch_diameter_mm=50.0,
+            contact_angle_deg=0.0,
+            shaft_speed_rpm=3000,
+        )
+        assert all(freqs[k] > 0 for k in ("BPFO", "BPFI", "BSF", "FTF"))
+
+    def test_angular_contact_bearing(self):
+        """Non-zero contact angle (angular contact bearing)."""
+        freqs_zero = calculate_bearing_frequencies(
+            num_balls=16, ball_diameter_mm=12.7,
+            pitch_diameter_mm=71.5, contact_angle_deg=0.0,
+            shaft_speed_rpm=2000,
+        )
+        freqs_15 = calculate_bearing_frequencies(
+            num_balls=16, ball_diameter_mm=12.7,
+            pitch_diameter_mm=71.5, contact_angle_deg=15.0,
+            shaft_speed_rpm=2000,
+        )
+        # cos(15°) < cos(0°), so d_ratio*cos(alpha) is smaller
+        # → BPFO should increase, BPFI should decrease compared to alpha=0
+        # Actually: BPFO = (Z/2)*fs*(1 - d_ratio*cos(a)) → larger with smaller cos
+        assert freqs_15["BPFO"] > freqs_zero["BPFO"]
+        assert freqs_15["BPFI"] < freqs_zero["BPFI"]
+
+    def test_high_contact_angle(self):
+        """45° contact angle (thrust bearing)."""
+        freqs = calculate_bearing_frequencies(
+            num_balls=12, ball_diameter_mm=10.0,
+            pitch_diameter_mm=60.0, contact_angle_deg=45.0,
+            shaft_speed_rpm=1500,
+        )
+        assert all(freqs[k] > 0 for k in ("BPFO", "BPFI", "BSF", "FTF"))
+
+    def test_frequency_ratios(self):
+        """BPFI/BPFO ratio should be > 1 for typical bearings."""
+        freqs = calculate_bearing_frequencies(
+            num_balls=8, ball_diameter_mm=10.0,
+            pitch_diameter_mm=50.0, shaft_speed_rpm=1500,
+        )
+        assert freqs["BPFI"] / freqs["BPFO"] > 1.0
+
+    def test_input_parameters_returned(self):
+        """Verify input parameters are echoed in result dict."""
+        freqs = calculate_bearing_frequencies(
+            num_balls=9, ball_diameter_mm=7.94,
+            pitch_diameter_mm=34.55, contact_angle_deg=0.0,
+            shaft_speed_rpm=1797,
+        )
+        params = freqs.get("input_parameters", {})
+        assert params.get("num_balls") == 9
+        assert params.get("shaft_speed_rpm") == 1797
+
+    def test_different_speeds(self):
+        """Frequencies should scale linearly with RPM."""
+        freqs_1000 = calculate_bearing_frequencies(
+            num_balls=8, ball_diameter_mm=10.0,
+            pitch_diameter_mm=50.0, shaft_speed_rpm=1000,
+        )
+        freqs_2000 = calculate_bearing_frequencies(
+            num_balls=8, ball_diameter_mm=10.0,
+            pitch_diameter_mm=50.0, shaft_speed_rpm=2000,
+        )
+        # Double RPM → double all frequencies
+        assert abs(freqs_2000["BPFO"] / freqs_1000["BPFO"] - 2.0) < 0.01
+        assert abs(freqs_2000["BPFI"] / freqs_1000["BPFI"] - 2.0) < 0.01
+        assert abs(freqs_2000["BSF"] / freqs_1000["BSF"] - 2.0) < 0.01
+
+
+# ── PDF text extraction ────────────────────────────────────────────────────
+
+class TestExtractTextFromPdf:
+
+    @pytest.mark.skipif(not HAS_PDF, reason="pypdf not installed")
+    def test_missing_file_raises(self):
+        with pytest.raises(FileNotFoundError):
+            extract_text_from_pdf(Path("nonexistent_file.pdf"))
+
+    @pytest.mark.skipif(not HAS_PDF, reason="pypdf not installed")
+    def test_real_catalog_if_available(self):
+        """Extract text from a real bearing catalog in the repo."""
+        catalogs_dir = Path(__file__).parent.parent / "resources" / "bearing_catalogs"
+        pdfs = list(catalogs_dir.glob("*.pdf")) if catalogs_dir.exists() else []
+        if not pdfs:
+            pytest.skip("No PDF catalogs found in resources/")
+
+        text = extract_text_from_pdf(pdfs[0], max_pages=3)
+        assert isinstance(text, str)
+        assert len(text) > 50  # Should have some content
+
+    @pytest.mark.skipif(not HAS_PDF, reason="pypdf not installed")
+    def test_max_pages_limits_extraction(self):
+        """max_pages should limit processing."""
+        catalogs_dir = Path(__file__).parent.parent / "resources" / "bearing_catalogs"
+        pdfs = list(catalogs_dir.glob("*.pdf")) if catalogs_dir.exists() else []
+        if not pdfs:
+            pytest.skip("No PDF catalogs found in resources/")
+
+        text_1 = extract_text_from_pdf(pdfs[0], max_pages=1)
+        text_3 = extract_text_from_pdf(pdfs[0], max_pages=3)
+        # More pages should produce more (or equal) text
+        assert len(text_3) >= len(text_1)
+
+    def test_no_pypdf_raises_import_error(self):
+        """When pypdf not installed, should raise ImportError."""
+        if HAS_PDF:
+            pytest.skip("pypdf IS installed — can't test missing path")
+        with pytest.raises(ImportError, match="pypdf"):
+            extract_text_from_pdf(Path("test.pdf"))