Karim shoair
71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
import pytest
|
|
import pytest_httpbin
|
|
from unittest.mock import Mock, patch
|
|
|
|
from scrapling.core.ai import ScraplingMCPServer, ResponseModel
|
|
|
|
|
|
@pytest_httpbin.use_class_based_httpbin
|
|
class TestMCPServer:
|
|
"""Test MCP server functionality"""
|
|
|
|
@pytest.fixture(scope="class")
|
|
def test_url(self, httpbin):
|
|
return f"{httpbin.url}/html"
|
|
|
|
@pytest.fixture
|
|
def server(self):
|
|
return ScraplingMCPServer()
|
|
|
|
def test_server_creation(self, server):
|
|
"""Test server instance creation"""
|
|
assert server._server is not None
|
|
assert server._server.name == "Scrapling"
|
|
|
|
def test_get_tool(self, server, test_url):
|
|
"""Test the get tool method"""
|
|
result = server.get(url=test_url, extraction_type="markdown")
|
|
assert isinstance(result, ResponseModel)
|
|
assert result.status == 200
|
|
assert result.url == test_url
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_get_tool(self, server, test_url):
|
|
"""Test the bulk_get tool method"""
|
|
results = await server.bulk_get(urls=(test_url, test_url), extraction_type="html")
|
|
|
|
assert len(results) == 2
|
|
assert all(isinstance(r, ResponseModel) for r in results)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_tool(self, server, test_url):
|
|
"""Test the fetch tool method"""
|
|
result = await server.fetch(url=test_url, headless=True)
|
|
assert isinstance(result, ResponseModel)
|
|
assert result.status == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_fetch_tool(self, server, test_url):
|
|
"""Test the bulk_fetch tool method"""
|
|
result = await server.bulk_fetch(urls=(test_url, test_url), headless=True)
|
|
assert all(isinstance(r, ResponseModel) for r in result)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stealthy_fetch_tool(self, server, test_url):
|
|
"""Test the stealthy_fetch tool method"""
|
|
result = await server.stealthy_fetch(url=test_url, headless=True)
|
|
assert isinstance(result, ResponseModel)
|
|
assert result.status == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_stealthy_fetch_tool(self, server, test_url):
|
|
"""Test the bulk_stealthy_fetch tool method"""
|
|
result = await server.bulk_stealthy_fetch(urls=(test_url, test_url), headless=True)
|
|
assert all(isinstance(r, ResponseModel) for r in result)
|
|
|
|
def test_serve_method(self, server):
|
|
"""Test the serve method"""
|
|
with patch.object(server._server, 'run') as mock_run:
|
|
server.serve()
|
|
mock_run.assert_called_once_with(transport="stdio")
|