"""
Mock HTTP server for consistent testing without external dependencies.
This server provides:
- Static HTML pages with predictable content
- JSON/XML/CSV endpoints
- Rate limiting simulation
- Error condition simulation
- Dynamic content generation
"""
import json
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread
from typing import Dict, Optional
from urllib.parse import parse_qs, urlparse
class MockHTTPRequestHandler(BaseHTTPRequestHandler):
"""Request handler for the mock HTTP server."""
# Track request count for rate limiting simulation
request_count: Dict[str, int] = {}
def log_message(self, format, *args):
"""Suppress default logging."""
pass
def do_GET(self):
"""Handle GET requests."""
parsed_path = urlparse(self.path)
path = parsed_path.path
query_params = parse_qs(parsed_path.query)
# Route requests
if path == "/":
self._serve_home()
elif path == "/products":
self._serve_products()
elif path == "/projects":
self._serve_projects()
elif path == "/api/data.json":
self._serve_json_data()
elif path == "/api/data.xml":
self._serve_xml_data()
elif path == "/api/data.csv":
self._serve_csv_data()
elif path == "/slow":
self._serve_slow_response()
elif path == "/error/404":
self._serve_404()
elif path == "/error/500":
self._serve_500()
elif path == "/rate-limited":
self._serve_rate_limited()
elif path == "/dynamic":
self._serve_dynamic_content()
elif path == "/pagination":
self._serve_pagination(query_params)
else:
self._serve_404()
def _serve_home(self):
"""Serve home page."""
html = """
Mock Test Website
Welcome to Mock Test Website
This is a test website for ScrapeGraphAI testing.
"""
self._send_html_response(html)
def _serve_products(self):
"""Serve products page."""
html = """
Products
Our Products
Product Alpha
High-quality product for testing
$99.99
In Stock
Product Beta
Another great product
$149.99
Limited Stock
Product Gamma
Premium product option
$199.99
Out of Stock
"""
self._send_html_response(html)
def _serve_projects(self):
"""Serve projects page."""
html = """
Projects
Our Projects
Project Alpha
A comprehensive web scraping solution
GitHub
Project Beta
AI-powered data extraction tool
GitHub
"""
self._send_html_response(html)
def _serve_json_data(self):
"""Serve JSON endpoint."""
data = {
"company": "Test Company",
"description": "A mock company for testing",
"employees": [
{"name": "Alice", "role": "Engineer", "department": "Engineering"},
{"name": "Bob", "role": "Designer", "department": "Design"},
{"name": "Charlie", "role": "Manager", "department": "Operations"},
],
"founded": "2020",
"location": "San Francisco",
}
self._send_json_response(data)
def _serve_xml_data(self):
"""Serve XML endpoint."""
xml = """
Test Company
A mock company for testing
Alice
Engineer
Bob
Designer
"""
self._send_xml_response(xml)
def _serve_csv_data(self):
"""Serve CSV endpoint."""
csv = """name,role,department
Alice,Engineer,Engineering
Bob,Designer,Design
Charlie,Manager,Operations"""
self._send_csv_response(csv)
def _serve_slow_response(self):
"""Simulate a slow response."""
time.sleep(2) # 2 second delay
self._send_html_response("Slow Response
")
def _serve_404(self):
"""Serve 404 error."""
self.send_response(404)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(b"404 Not Found
")
def _serve_500(self):
"""Serve 500 error."""
self.send_response(500)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(b"500 Internal Server Error
")
def _serve_rate_limited(self):
"""Simulate rate limiting."""
client_ip = self.client_address[0]
self.request_count[client_ip] = self.request_count.get(client_ip, 0) + 1
if self.request_count[client_ip] > 5:
self.send_response(429)
self.send_header("Content-type", "text/html")
self.send_header("Retry-After", "60")
self.end_headers()
self.wfile.write(b"429 Too Many Requests
")
else:
self._send_html_response("Rate Limited Endpoint
")
def _serve_dynamic_content(self):
"""Serve dynamically generated content."""
timestamp = int(time.time())
html = f"""
Dynamic Content
Dynamic Content
Generated at: {timestamp}
Random value: {timestamp % 1000}
"""
self._send_html_response(html)
def _serve_pagination(self, query_params):
"""Serve paginated content."""
page = int(query_params.get("page", ["1"])[0])
per_page = 10
total_items = 50
items = []
start = (page - 1) * per_page
end = min(start + per_page, total_items)
for i in range(start, end):
items.append(f'Item {i + 1}')
next_page = page + 1 if end < total_items else None
prev_page = page - 1 if page > 1 else None
html = f"""
Pagination - Page {page}
Paginated Content - Page {page}
"""
self._send_html_response(html)
def _send_html_response(self, html: str, status: int = 200):
"""Send HTML response."""
self.send_response(status)
self.send_header("Content-type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(html.encode("utf-8"))
def _send_json_response(self, data: dict, status: int = 200):
"""Send JSON response."""
self.send_response(status)
self.send_header("Content-type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(data).encode("utf-8"))
def _send_xml_response(self, xml: str, status: int = 200):
"""Send XML response."""
self.send_response(status)
self.send_header("Content-type", "application/xml")
self.end_headers()
self.wfile.write(xml.encode("utf-8"))
def _send_csv_response(self, csv: str, status: int = 200):
"""Send CSV response."""
self.send_response(status)
self.send_header("Content-type", "text/csv")
self.end_headers()
self.wfile.write(csv.encode("utf-8"))
class MockHTTPServer:
"""Mock HTTP server for testing."""
def __init__(self, host: str = "localhost", port: int = 8888):
self.host = host
self.port = port
self.server: Optional[HTTPServer] = None
self.thread: Optional[Thread] = None
def start(self):
"""Start the mock server in a background thread."""
self.server = HTTPServer((self.host, self.port), MockHTTPRequestHandler)
self.thread = Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
time.sleep(0.1) # Give server time to start
def stop(self):
"""Stop the mock server."""
if self.server:
self.server.shutdown()
self.server.server_close()
if self.thread:
self.thread.join(timeout=1)
def get_url(self, path: str = "") -> str:
"""Get full URL for a given path."""
return f"http://{self.host}:{self.port}{path}"
def __enter__(self):
"""Context manager entry."""
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
self.stop()