Spaces:
Sleeping
Sleeping
| import pytest | |
| from fastapi.testclient import TestClient | |
| from app import app | |
| import time | |
| def client(): | |
| """Create test client""" | |
| return TestClient(app) | |
| # ========== Session Management Tests ========== | |
| def test_create_session(client): | |
| """Test creating a new session""" | |
| response = client.post("/sessions", json={ | |
| "metadata": {"test": "create_session"}, | |
| "timeout_minutes": 30 | |
| }) | |
| if response.status_code == 201: | |
| data = response.json() | |
| assert "session_id" in data | |
| assert data["status"] in ["creating", "ready"] | |
| assert data["timeout_minutes"] == 30 | |
| # Cleanup | |
| session_id = data["session_id"] | |
| client.delete(f"/sessions/{session_id}") | |
| def test_list_sessions(client): | |
| """Test listing all sessions""" | |
| # Create a session first | |
| create_response = client.post("/sessions", json={"metadata": {"test": "list"}}) | |
| if create_response.status_code == 201: | |
| # List sessions | |
| response = client.get("/sessions") | |
| assert response.status_code == 200 | |
| sessions = response.json() | |
| assert isinstance(sessions, list) | |
| # Cleanup | |
| session_id = create_response.json()["session_id"] | |
| client.delete(f"/sessions/{session_id}") | |
| def test_get_session(client): | |
| """Test getting session details""" | |
| # Create session | |
| create_response = client.post("/sessions", json={"metadata": {"test": "get"}}) | |
| if create_response.status_code == 201: | |
| session_id = create_response.json()["session_id"] | |
| # Get session details | |
| response = client.get(f"/sessions/{session_id}") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["session_id"] == session_id | |
| # Cleanup | |
| client.delete(f"/sessions/{session_id}") | |
| def test_get_nonexistent_session(client): | |
| """Test getting a session that doesn't exist""" | |
| response = client.get("/sessions/nonexistent-id") | |
| # Returns 503 if Docker is unavailable, 404 if Docker is available but session doesn't exist | |
| assert response.status_code in [404, 503] | |
| def test_destroy_session(client): | |
| """Test destroying a session""" | |
| # Create session | |
| create_response = client.post("/sessions", json={"metadata": {"test": "destroy"}}) | |
| if create_response.status_code == 201: | |
| session_id = create_response.json()["session_id"] | |
| # Destroy session | |
| response = client.delete(f"/sessions/{session_id}") | |
| assert response.status_code == 200 | |
| # Verify it's gone | |
| get_response = client.get(f"/sessions/{session_id}") | |
| assert get_response.status_code == 404 | |
| # ========== File Operations Tests ========== | |
| def test_upload_file(client): | |
| """Test uploading a file to session""" | |
| # Create session | |
| create_response = client.post("/sessions", json={"metadata": {"test": "upload"}}) | |
| if create_response.status_code == 201: | |
| session_id = create_response.json()["session_id"] | |
| time.sleep(2) # Wait for container to be ready | |
| # Upload file | |
| file_content = b"print('Hello from test file')" | |
| response = client.post( | |
| f"/sessions/{session_id}/files", | |
| files={"file": ("test.py", file_content, "text/x-python")} | |
| ) | |
| if response.status_code == 200: | |
| data = response.json() | |
| assert data["filename"] == "test.py" | |
| assert data["size"] == len(file_content) | |
| # Cleanup | |
| client.delete(f"/sessions/{session_id}") | |
| def test_list_files(client): | |
| """Test listing files in session""" | |
| # Create session | |
| create_response = client.post("/sessions", json={"metadata": {"test": "list_files"}}) | |
| if create_response.status_code == 201: | |
| session_id = create_response.json()["session_id"] | |
| time.sleep(2) | |
| # Upload a file first | |
| client.post( | |
| f"/sessions/{session_id}/files", | |
| files={"file": ("test.txt", b"test content", "text/plain")} | |
| ) | |
| # List files | |
| response = client.get(f"/sessions/{session_id}/files") | |
| assert response.status_code == 200 | |
| files = response.json() | |
| assert isinstance(files, list) | |
| # Cleanup | |
| client.delete(f"/sessions/{session_id}") | |
| def test_download_file(client): | |
| """Test downloading a file from session""" | |
| # Create session | |
| create_response = client.post("/sessions", json={"metadata": {"test": "download"}}) | |
| if create_response.status_code == 201: | |
| session_id = create_response.json()["session_id"] | |
| time.sleep(2) | |
| # Upload file | |
| file_content = b"download test content" | |
| upload_response = client.post( | |
| f"/sessions/{session_id}/files", | |
| files={"file": ("download.txt", file_content, "text/plain")} | |
| ) | |
| if upload_response.status_code == 200: | |
| # Download file | |
| response = client.get(f"/sessions/{session_id}/files/download.txt") | |
| assert response.status_code == 200 | |
| assert response.content == file_content | |
| # Cleanup | |
| client.delete(f"/sessions/{session_id}") | |
| # ========== Execute in Session Tests ========== | |
| def test_execute_in_session(client): | |
| """Test executing code in a session""" | |
| # Create session | |
| create_response = client.post("/sessions", json={"metadata": {"test": "execute"}}) | |
| if create_response.status_code == 201: | |
| session_id = create_response.json()["session_id"] | |
| time.sleep(2) | |
| # Execute code | |
| response = client.post( | |
| f"/sessions/{session_id}/execute", | |
| json={ | |
| "code": "print('Hello from session')", | |
| "language": "python", | |
| "working_dir": "/workspace" | |
| } | |
| ) | |
| if response.status_code == 200: | |
| data = response.json() | |
| assert "Hello from session" in data["stdout"] | |
| assert data["exit_code"] == 0 | |
| # Cleanup | |
| client.delete(f"/sessions/{session_id}") | |
| def test_execute_file_in_session(client): | |
| """Test executing an uploaded file""" | |
| # Create session | |
| create_response = client.post("/sessions", json={"metadata": {"test": "execute_file"}}) | |
| if create_response.status_code == 201: | |
| session_id = create_response.json()["session_id"] | |
| time.sleep(2) | |
| # Upload Python file | |
| file_content = b"print('Executed from file')\nprint(2 + 2)" | |
| upload_response = client.post( | |
| f"/sessions/{session_id}/files", | |
| files={"file": ("script.py", file_content, "text/x-python")} | |
| ) | |
| if upload_response.status_code == 200: | |
| # Execute file | |
| response = client.post( | |
| f"/sessions/{session_id}/execute-file", | |
| json={ | |
| "filepath": "/workspace/script.py", | |
| "language": "python", | |
| "args": [] | |
| } | |
| ) | |
| if response.status_code == 200: | |
| data = response.json() | |
| assert "Executed from file" in data["stdout"] | |
| assert "4" in data["stdout"] | |
| assert data["exit_code"] == 0 | |
| # Cleanup | |
| client.delete(f"/sessions/{session_id}") | |
| def test_persistent_state_across_executions(client): | |
| """Test that session maintains state across multiple executions""" | |
| # Create session | |
| create_response = client.post("/sessions", json={"metadata": {"test": "persistent"}}) | |
| if create_response.status_code == 201: | |
| session_id = create_response.json()["session_id"] | |
| time.sleep(2) | |
| # First execution: create file | |
| response1 = client.post( | |
| f"/sessions/{session_id}/execute", | |
| json={ | |
| "code": "with open('/workspace/data.txt', 'w') as f: f.write('persistent')", | |
| "language": "python" | |
| } | |
| ) | |
| if response1.status_code == 200: | |
| # Second execution: read file (should still exist) | |
| response2 = client.post( | |
| f"/sessions/{session_id}/execute", | |
| json={ | |
| "code": "with open('/workspace/data.txt', 'r') as f: print(f.read())", | |
| "language": "python" | |
| } | |
| ) | |
| if response2.status_code == 200: | |
| data = response2.json() | |
| assert "persistent" in data["stdout"] | |
| # Cleanup | |
| client.delete(f"/sessions/{session_id}") | |
| # ========== Backward Compatibility Tests ========== | |
| def test_stateless_execute_still_works(client): | |
| """Test that original /execute endpoint still works""" | |
| response = client.post("/execute", json={ | |
| "code": "print('Stateless execution')", | |
| "language": "python" | |
| }) | |
| if response.status_code == 200: | |
| data = response.json() | |
| assert "Stateless execution" in data["stdout"] | |
| assert data["exit_code"] == 0 | |
| # ========== API Root and Health Tests ========== | |
| def test_root_endpoint(client): | |
| """Test root endpoint""" | |
| response = client.get("/") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["name"] == "isolated-sandbox" | |
| assert data["version"] == "2.0.0" | |
| assert "endpoints" in data | |
| def test_health_endpoint(client): | |
| """Test health check""" | |
| response = client.get("/health") | |
| if response.status_code == 200: | |
| data = response.json() | |
| assert data["status"] == "healthy" | |
| assert "docker" in data | |
| def test_languages_endpoint(client): | |
| """Test languages listing""" | |
| response = client.get("/languages") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "languages" in data | |
| assert len(data["languages"]) > 0 | |
| def test_ui_rules_endpoint(client): | |
| """Test UI rules documentation endpoint""" | |
| response = client.get("/ui-rules") | |
| assert response.status_code == 200 | |
| assert response.headers["content-type"].startswith("text/html") | |
| content = response.text | |
| assert "UI/UX Design Rules" in content | |
| assert "isolated-sandbox" in content | |
| assert "MUST" in content | |
| assert "SHOULD" in content | |
| assert "NEVER" in content | |