Add test mode to suppress browser and file operations

Author: rich-iannone

pointblank_mcp_server/pointblank_server.py

@@ -1,6 +1,8 @@
 import json
 import logging
 import math
+import os
+import sys
 import uuid
 import webbrowser
 from contextlib import asynccontextmanager
@@ -21,6 +23,13 @@
 
 import pointblank as pb
 
+# Detect if we're running in a test environment
+TESTING_MODE = (
+    "pytest" in sys.modules
+    or os.environ.get("PYTEST_CURRENT_TEST") is not None
+    or os.environ.get("POINTBLANK_TESTING") == "true"
+)
+
 # Try to import Pandas, but make it optional
 try:
     import pandas as pd
@@ -113,6 +122,14 @@ def _save_dataframe_to_csv(df: Any, output_path: Path) -> None:
             raise TypeError(f"Unsupported DataFrame type '{type(df).__name__}' for CSV export.")
 
 
+def _open_browser_conditionally(url: str) -> None:
+    """Open browser only if not in testing mode."""
+    if not TESTING_MODE:
+        webbrowser.open(url)
+    else:
+        logger.debug(f"Browser opening suppressed in testing mode for: {url}")
+
+
 def _load_dataframe_from_path(input_path: str, backend: str = "auto") -> Any:
     """Load DataFrame from file using specified backend or auto-detect."""
     p_path = Path(input_path)
@@ -307,6 +324,16 @@ def _generate_validation_report_html(validator: pb.Validate, validator_id: str)
     Returns the file path.
     """
     try:
+        # Generate timestamped filename
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        filename = f"pointblank_validation_report_{validator_id}_{timestamp}.html"
+        file_path = Path.cwd() / filename
+
+        # Skip file generation during testing
+        if TESTING_MODE:
+            logger.debug(f"Skipping HTML file generation during testing: {filename}")
+            return str(file_path.resolve())  # Return path but don't create file
+
         # Get the validation report as a GT table
         gt_report = validator.get_tabular_report()
 
@@ -329,11 +356,6 @@ def _generate_validation_report_html(validator: pb.Validate, validator_id: str)
         if isinstance(html_content, bytes):
             html_content = html_content.decode("utf-8", errors="replace")
 
-        # Generate timestamped filename
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-        filename = f"pointblank_validation_report_{validator_id}_{timestamp}.html"
-        file_path = Path.cwd() / filename
-
         # Save HTML file with explicit UTF-8 encoding
         with open(file_path, "w", encoding="utf-8", newline="") as f:
             f.write(html_content)
@@ -1858,7 +1880,7 @@ async def interrogate_validator(
         # Generate HTML report table and open in browser
         try:
             html_report_path = _generate_validation_report_html(validator, validator_id)
-            webbrowser.open(f"file://{html_report_path}")
+            _open_browser_conditionally(f"file://{html_report_path}")
             await ctx.report_progress(
                 50, 100, f"Validation report opened in browser: {html_report_path}"
             )
@@ -2142,15 +2164,19 @@ async def preview_table(
             )
             html_path = Path.cwd() / html_filename
 
-            with open(html_path, "w", encoding="utf-8") as f:
-                f.write(full_html)
+            # Skip file generation during testing
+            if TESTING_MODE:
+                browser_msg = f"HTML preview generated (file creation skipped during testing)\n\nFile location: {html_path}"
+            else:
+                with open(html_path, "w", encoding="utf-8") as f:
+                    f.write(full_html)
 
-            # Open in default browser
-            try:
-                webbrowser.open(f"file://{html_path}")
-                browser_msg = f"HTML preview saved and opened in default browser!\n\nFile location: {html_path}"
-            except Exception as browser_error:
-                browser_msg = f"HTML preview saved to: {html_path}\n\n📖 Could not open browser automatically: {str(browser_error)}\nPlease open the file manually in your browser."
+                # Open in default browser
+                try:
+                    _open_browser_conditionally(f"file://{html_path}")
+                    browser_msg = f"HTML preview saved and opened in default browser!\n\nFile location: {html_path}"
+                except Exception as browser_error:
+                    browser_msg = f"HTML preview saved to: {html_path}\n\n📖 Could not open browser automatically: {str(browser_error)}\nPlease open the file manually in your browser."
 
         except Exception as e:
             browser_msg = f"Error saving HTML file: {str(e)}"
@@ -2228,15 +2254,19 @@ async def missing_values_table(
             html_filename = f"pointblank_missing_values_{dataframe_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"
             html_path = Path.cwd() / html_filename
 
-            with open(html_path, "w", encoding="utf-8") as f:
-                f.write(full_html)
+            # Skip file generation during testing
+            if TESTING_MODE:
+                browser_msg = f"HTML missing values analysis generated (file creation skipped during testing)\n\nFile location: {html_path}"
+            else:
+                with open(html_path, "w", encoding="utf-8") as f:
+                    f.write(full_html)
 
-            # Open in default browser
-            try:
-                webbrowser.open(f"file://{html_path}")
-                browser_msg = f"HTML missing values analysis saved and opened in default browser!\n\nFile location: {html_path}"
-            except Exception as browser_error:
-                browser_msg = f"HTML analysis saved to: {html_path}\n\n📖 Could not open browser automatically: {str(browser_error)}\nPlease open the file manually in your browser."
+                # Open in default browser
+                try:
+                    _open_browser_conditionally(f"file://{html_path}")
+                    browser_msg = f"HTML missing values analysis saved and opened in default browser!\n\nFile location: {html_path}"
+                except Exception as browser_error:
+                    browser_msg = f"HTML analysis saved to: {html_path}\n\n📖 Could not open browser automatically: {str(browser_error)}\nPlease open the file manually in your browser."
 
         except Exception as e:
             browser_msg = f"Error saving HTML file: {str(e)}"
@@ -2315,15 +2345,19 @@ async def column_summary_table(
             html_filename = f"pointblank_column_summary_{dataframe_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"
             html_path = Path.cwd() / html_filename
 
-            with open(html_path, "w", encoding="utf-8") as f:
-                f.write(full_html)
+            # Skip file generation during testing
+            if TESTING_MODE:
+                browser_msg = f"HTML column summary generated (file creation skipped during testing)\n\nFile location: {html_path}"
+            else:
+                with open(html_path, "w", encoding="utf-8") as f:
+                    f.write(full_html)
 
-            # Open in default browser
-            try:
-                webbrowser.open(f"file://{html_path}")
-                browser_msg = f"HTML column summary saved and opened in default browser!\n\nFile location: {html_path}"
-            except Exception as browser_error:
-                browser_msg = f"HTML summary saved to: {html_path}\n\n📖 Could not open browser automatically: {str(browser_error)}\nPlease open the file manually in your browser."
+                # Open in default browser
+                try:
+                    _open_browser_conditionally(f"file://{html_path}")
+                    browser_msg = f"HTML column summary saved and opened in default browser!\n\nFile location: {html_path}"
+                except Exception as browser_error:
+                    browser_msg = f"HTML summary saved to: {html_path}\n\n📖 Could not open browser automatically: {str(browser_error)}\nPlease open the file manually in your browser."
 
         except Exception as e:
             browser_msg = f"Error saving HTML file: {str(e)}"

tests/test_mcp_server.py

@@ -3,7 +3,6 @@
 import pandas as pd
 import pytest
 
-# Import the fastmcp Client and your mcp application instance
 from fastmcp import Client
 from pointblank_mcp_server.pointblank_server import mcp
 
@@ -35,54 +34,50 @@ def csv_file(tmp_path_factory, sample_df) -> str:
     return str(file_path)
 
 
-# --- Test Functions ---
-
-
 @pytest.mark.asyncio
 async def test_list_available_tools(mcp_server):
     """Tests if the MCP is up and correctly lists its registered tools."""
-    print("\n=== Testing Tool Listing ===")
+
     async with Client(mcp_server) as client:
         tools = await client.list_tools()
         tool_names = [tool.name for tool in tools]
 
-        print(f"✅ Found {len(tool_names)} tools.")
         assert "load_dataframe" in tool_names
         assert "create_validator" in tool_names
         assert "add_validation_step" in tool_names
         assert "interrogate_validator" in tool_names
         assert "get_validation_step_output" in tool_names
-        print("✅ All expected tools are registered.")
 
 
 @pytest.mark.asyncio
 async def test_full_validation_workflow(mcp_server, csv_file: str, tmp_path: Path):
-    """
-    Tests a complete user workflow using in-memory calls to the MCP server.
-    """
-    print("\n=== Testing Full Validation Workflow ===")
+    """Tests a complete user workflow using in-memory calls to the MCP server."""
 
     # The client connects directly to the mcp object in memory
     async with Client(mcp_server) as client:
         # 1. Load a DataFrame from the CSV file
-        print("🔧 Step 1: Calling 'load_dataframe'...")
+
         result = await client.call_tool("load_dataframe", {"input_path": csv_file})
+
         assert not result.is_error
+
         df_info = result.data
+
         assert tuple(df_info.shape) == (5, 4)
+
         df_id = df_info.df_id
-        print(f"✅ DataFrame loaded with ID: {df_id}")
 
         # 2. Create a validator for the loaded DataFrame
-        print("\n🔧 Step 2: Calling 'create_validator'...")
+
         result = await client.call_tool("create_validator", {"df_id": df_id})
+
         assert not result.is_error
+
         validator_info = result.data
         validator_id = validator_info.validator_id
-        print(f"✅ Validator created with ID: {validator_id}")
 
         # 3. Add validation steps
-        print("\n🔧 Step 3: Calling 'add_validation_step'...")
+
         # This is STEP 1
         step1_params = {"columns": "id"}
         result = await client.call_tool(
@@ -94,7 +89,6 @@ async def test_full_validation_workflow(mcp_server, csv_file: str, tmp_path: Pat
             },
         )
         assert not result.is_error
-        print("✅ Added passing step: 'id' column is not null")
 
         # This is STEP 2
         step2_params = {"columns": "age", "value": 10}
@@ -107,21 +101,41 @@ async def test_full_validation_workflow(mcp_server, csv_file: str, tmp_path: Pat
             },
         )
         assert not result.is_error
-        print("✅ Added failing step: 'age' column values < 10")
 
         # 4. Interrogate the validator
-        print("\n🔧 Step 4: Calling 'interrogate_validator'...")
+
         result = await client.call_tool("interrogate_validator", {"validator_id": validator_id})
+
         assert not result.is_error
+
         interrogate_result = result.data
         summary = interrogate_result["validation_summary"]
 
+        # Verify our new functionality - Python code generation
+        assert "python_code" in interrogate_result
+
+        python_code = interrogate_result["python_code"]
+
+        assert isinstance(python_code, str)
+        assert "import pointblank as pb" in python_code
+        assert "pb.Validate(df)" in python_code
+        assert ".col_vals_not_null(columns='id')" in python_code
+        assert ".col_vals_lt(columns='age', value=10)" in python_code
+        assert ".interrogate()" in python_code
+
+        # Verify instructions are provided
+        assert "instructions" in interrogate_result
+
+        instructions = interrogate_result["instructions"]
+
+        assert "html_report" in instructions
+        assert "python_code" in instructions
+
         # summary is 0-indexed, so summary[1] is the second step
         assert summary[1]["f_passed"] < 1.0
-        print("✅ Interrogation complete. Found expected failures.")
 
         # 5. Get the output for the specific failing step (step_index=2)
-        print("\n🔧 Step 5: Calling 'get_validation_step_output' for a specific failing step...")
+
         failed_step_path = str(tmp_path / "failed_step_output.csv")
         result = await client.call_tool(
             "get_validation_step_output",
@@ -133,15 +147,18 @@ async def test_full_validation_workflow(mcp_server, csv_file: str, tmp_path: Pat
             },
         )
         assert not result.is_error
+
         output_info = result.data
+
         assert output_info.output_file is not None, f"No CSV was written: {output_info.message}"
         assert Path(output_info.output_file).exists()
+
         failed_df = pd.read_csv(output_info.output_file)
+
         assert len(failed_df) == 4  # Should contain the 4 rows that failed the age check
-        print(f"✅ Failing rows for step 2 correctly saved to '{output_info.output_file}'")
 
         # 6. Get the output for all passing data across the entire run
-        print("\n🔧 Step 6: Calling 'get_validation_step_output' for all passing data...")
+
         passed_run_path = str(tmp_path / "passed_run_output.csv")
         result = await client.call_tool(
             "get_validation_step_output",
@@ -152,11 +169,13 @@ async def test_full_validation_workflow(mcp_server, csv_file: str, tmp_path: Pat
             },
         )
         assert not result.is_error
+
         pass_info = result.data
+
         assert pass_info.output_file is not None, f"No CSV was written: {pass_info.message}"
         assert Path(pass_info.output_file).exists()
+
         pass_df = pd.read_csv(pass_info.output_file)
+
         assert len(pass_df) == 1  # Only the row with age=9 passed all validations
         assert pass_df.iloc[0]["age"] == 9
-        print(f"✅ All passing rows correctly saved to '{pass_info.output_file}'")
-        print("\n🎉 Full workflow test passed!")