Test FastAPI Functionality

write a test.py to test the functionality of this: from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel import requests import pytesseract from bs4 import BeautifulSoup from PIL import Image import os from typing import Dict, Any, Optional, List from uuid import uuid4 from playwright.async_api import async_playwright, Browser app = FastAPI() OLLAMA_API_URL = "http://localhost:11434/api/generate" class OllamaRequest(BaseModel): prompt: str model: str = "mistral" session_id: Optional[str] = None class WebPageRequest(BaseModel): url: str session_id: Optional[str] = None class BrowseAction(BaseModel): action_type: str selector: str text: Optional[str] = None class BrowseRequest(BaseModel): url: str actions: List[BrowseAction] session_id: Optional[str] = None class ContextManager: def __init__(self): self.contexts: Dict[str, Dict[str, Any]] = {} def create_context(self, session_id: str) -> str: if not session_id: session_id = str(uuid4()) self.contexts[session_id] = {} return session_id def update_context(self, session_id: str, key: str, value: Any): if session_id in self.contexts: self.contexts[session_id][key] = value def get_context(self, session_id: str, key: str) -> Any: return self.contexts.get(session_id, {}).get(key) def reset_context(self, session_id: str): if session_id in self.contexts: del self.contexts[session_id] context_manager = ContextManager() def get_session(session_id: Optional[str] = None): if not session_id: session_id = str(uuid4()) if session_id not in context_manager.contexts: context_manager.create_context(session_id) return session_id @app.on_event("startup") async def startup_event(): try: playwright = await async_playwright().start() app.state.playwright = playwright app.state.browser = await playwright.chromium.launch() except Exception as e: print(f"Error starting Playwright: {str(e)}") raise @app.on_event("shutdown") async def shutdown_event(): await app.state.browser.close() await app.state.playwright.stop() @app.post("/mcp/ollama") async def ollama_generate(request: OllamaRequest, session_id: str = Depends(get_session)): previous_text = context_manager.get_context(session_id, "previous_text") payload = { "model": request.model, "prompt": f"{request.prompt} {previous_text or ''}", "stream": False } try: response = requests.post(OLLAMA_API_URL, json=payload) response.raise_for_status() result = response.json().get("response", "") context_manager.update_context(session_id, "previous_text", result) return {"response": result, "session_id": session_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/mcp/web_search") async def web_search(query: str, session_id: str = Depends(get_session)): try: search_url = f"https://www.google.com/search?q={query}" response = requests.get(search_url, timeout=5) response.raise_for_status() soup = BeautifulSoup(response.text, 'html.parser') results = [] for g in soup.find_all('div', class_='g'): if anchors := g.find_all('a'): link = anchors[0]['href'] if title := g.find('h3'): results.append({"title": title.text, "link": link}) context_manager.update_context(session_id, "search_results", results) return {"results": results, "session_id": session_id} except requests.RequestException as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/mcp/fetch_webpage") async def fetch_webpage(request: WebPageRequest, session_id: str = Depends(get_session)): try: response = requests.get(request.url) response.raise_for_status() soup = BeautifulSoup(response.text, 'html.parser') content = soup.get_text(separator='\n', strip=True) context_manager.update_context(session_id, "webpage_content", content) return {"content": content, "session_id": session_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/mcp/browse") async def browse_page(request: BrowseRequest, session_id: str = Depends(get_session)): context = None try: browser = app.state.browser context = await browser.new_context() page = await context.new_page() await page.goto(request.url) for action in request.actions: if action.action_type == "click": await page.click(action.selector) elif action.action_type == "type": await page.fill(action.selector, action.text or "") elif action.action_type == "navigate": await page.goto(action.selector) elif action.action_type == "wait": await page.wait_for_timeout(int(action.selector)) content = await page.content() text_content = await page.evaluate("() => document.body.innerText") context_manager.update_context(session_id, "browsed_content", { "html": content, "text": text_content }) return {"content": text_content, "session_id": session_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) finally: if context: await context.close() @app.post("/mcp/read_file") async def read_file(file_path: str, session_id: str = Depends(get_session)): try: if not os.path.exists(file_path): raise FileNotFoundError("File not found") with open(file_path, 'r') as f: content = f.read() context_manager.update_context(session_id, "file_content", content) return {"content": content, "session_id": session_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/mcp/ocr_image") async def ocr_image(image_path: str, session_id: str = Depends(get_session)): try: image = Image.open(image_path) text = pytesseract.image_to_string(image) context_manager.update_context(session_id, "ocr_text", text) return {"text": text, "session_id": session_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.delete("/mcp/context/{session_id}") async def clear_context(session_id: str): context_manager.reset_context(session_id) return {"status": "context cleared"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

To test the functionality of the FastAPI app you've provided, we can create a `test.py` script using `pytest` and `httpx` to send test requests to your FastAPI endpoints. Below is an example `test.py` script for testing the functionality of the FastAPI app: ```python import pytest from fastapi.testclient import TestClient from main import app # Assuming your FastAPI app is defined in 'main.py' client = TestClient(app) @pytest.fixture def setup_context(): """Create a session context before tests.""" response = client.post("/mcp/ollama", json={"prompt": "Test"}) assert response.status_code == 200 return response.json()["session_id"] def test_ollama_generate(setup_context): session_id = setup_context response = client.post("/mcp/ollama", json={"prompt": "This is a test prompt."}, params={"session_id": session_id}) assert response.status_code == 200 data = response.json() assert "response" in data assert "session_id" in data assert data["session_id"] == session_id def test_web_search(setup_context): session_id = setup_context response = client.post("/mcp/web_search", json={"query": "FastAPI"}, params={"session_id": session_id}) assert response.status_code == 200 data = response.json() assert "results" in data assert "session_id" in data assert data["session_id"] == session_id def test_fetch_webpage(setup_context): session_id = setup_context response = client.post("/mcp/fetch_webpage", json={"url": "https://www.example.com"}, params={"session_id": session_id}) assert response.status_code == 200 data = response.json() assert "content" in data assert "session_id" in data assert data["session_id"] == session_id def test_browse_page(setup_context): session_id = setup_context actions = [ {"action_type": "click", "selector": "button#submit"}, {"action_type": "type", "selector": "input#search", "text": "Hello"} ] response = client.post("/mcp/browse", json={"url": "https://www.example.com", "actions": actions}, params={"session_id": session_id}) assert response.status_code == 200 data = response.json() assert "content" in data assert "session_id" in data assert data["session_id"] == session_id def test_read_file(setup_context): session_id = setup_context response = client.post("/mcp/read_file", json={"file_path": "test_file.txt"}, params={"session_id": session_id}) assert response.status_code == 200 data = response.json() assert "content" in data assert "session_id" in data assert data["session_id"] == session_id def test_ocr_image(setup_context): session_id = setup_context response = client.post("/mcp/ocr_image", json={"image_path": "test_image.png"}, params={"session_id": session_id}) assert response.status_code == 200 data = response.json() assert "text" in data assert "session_id" in data assert data["session_id"] == session_id def test_clear_context(setup_context): session_id = setup_context response = client.delete(f"/mcp/context/{session_id}") assert response.status_code == 200 data = response.json() assert "status" in data assert data["status"] == "context cleared" if __name__ == "__main__": pytest.main() ``` ### Explanation: - **`TestClient`**: This is used to make synchronous HTTP requests to the FastAPI app for testing purposes. - **`setup_context` fixture**: This fixture runs a pre-test request to initialize a session and get the `session_id`. - **Test functions**: Each test corresponds to a route of the FastAPI app. It makes a request to that endpoint and checks the response status code and the presence of required fields. - **`assert` statements**: These ensure the response contains the correct data and meets expectations, such as correct session IDs, content, and status codes. ### Running Tests: To run the tests, ensure that `pytest` and `httpx` are installed and then run the command: ```bash pytest test.py ``` This will execute all the test cases and show results in the terminal.

write a curl command to test the browser use for danielkliewer.com

To test the browser usage for `danielkliewer.com`, particularly the `browse` functionality from your FastAPI application (which uses Playwright to simulate browser interactions), we can use the following `curl` command to simulate browsing actions. For this example, we'll assume that we want to test browsing the homepage of `danielkliewer.com` and performing an action like clicking a button or typing text. Here’s how you could craft the `curl` command: ### Example `curl` Command: ```bash curl -X 'POST' \ 'http://localhost:8000/mcp/browse' \ -H 'Content-Type: application/json' \ -d '{ "url": "https://danielkliewer.com", "actions": [ { "action_type": "click", "selector": "button#subscribe" }, { "action_type": "type", "selector": "input#email", "text": "test@example.com" } ] }' ``` ### Explanation: - **`-X 'POST'`**: Specifies the HTTP request method (POST). - **URL**: The endpoint you're testing is `/mcp/browse`, which uses Playwright for browser interaction. The FastAPI app is running on `localhost:8000` in this example. - **Headers**: The `-H 'Content-Type: application/json'` sets the request header to indicate that you're sending JSON data. - **`-d` (Data)**: The body of the request includes: - `url`: The website URL (`https://danielkliewer.com`) to visit. - `actions`: A list of actions to perform on the page. In this case: - **Clicking** a button with `id="subscribe"`. - **Typing** the email `test@example.com` into an input field with `id="email"`. This `curl` command will send the request to your FastAPI app, where Playwright will simulate browsing the page and performing the specified actions. ### Expected Response: The server should respond with the page's content or any text content extracted after performing the actions. You can then verify if the actions were successfully carried out. If any errors occur, you should see an HTTP error message explaining the issue.