File size: 10,516 Bytes
a89f25d
 
 
fa11621
a89f25d
 
 
 
fa11621
a89f25d
 
 
fa11621
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a89f25d
 
fa11621
 
 
 
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
a89f25d
 
fa11621
 
 
 
a89f25d
fa11621
 
 
 
 
 
a89f25d
fa11621
 
 
 
a89f25d
 
fa11621
 
 
c2fe03b
 
a89f25d
 
fa11621
 
 
 
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
a89f25d
 
fa11621
 
 
 
 
 
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a89f25d
 
fa11621
 
 
 
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a89f25d
 
fa11621
 
 
 
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a89f25d
 
fa11621
 
 
 
 
 
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a89f25d
 
fa11621
 
 
 
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a89f25d
 
fa11621
 
 
 
a89f25d
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a89f25d
fa11621
a89f25d
fa11621
 
a89f25d
fa11621
 
a89f25d
 
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
f9b2131
fa11621
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
f9b2131
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
import pytest
from fastapi.testclient import TestClient
from app import app
import time


@pytest.fixture
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