tests for secured envd access

Author: sitole

packages/python-sdk/e2b/connection_config.py

@@ -42,7 +42,7 @@ def __init__(
         self.debug = debug or ConnectionConfig._debug()
         self.api_key = api_key or ConnectionConfig._api_key()
         self.access_token = access_token or ConnectionConfig._access_token()
-        self.headers = headers
+        self.headers = headers or {}
 
         self.request_timeout = ConnectionConfig._get_request_timeout(
             REQUEST_TIMEOUT,
@@ -75,6 +75,8 @@ def _get_request_timeout(
     def get_request_timeout(self, request_timeout: Optional[float] = None):
         return self._get_request_timeout(self.request_timeout, request_timeout)
 
+    def get_request_base_headers(self):
+        return self.headers or {}
 
 Username = Literal["root", "user"]
 """

packages/python-sdk/tests/async/sandbox_async/files/test_watch.py

@@ -112,3 +112,11 @@ async def test_watch_file(async_sandbox: AsyncSandbox):
 
     with pytest.raises(SandboxException):
         await async_sandbox.files.watch_dir(filename, on_event=lambda e: None)
+
+async def test_watch_file_with_secured_envd(template):
+    sbx = await AsyncSandbox.create(template, timeout=30, secure=True)
+    try:
+        await sbx.files.watch_dir("/home/user/", on_event=lambda e: None)
+        await sbx.files.write("test_watch.txt", "This file will be watched.")
+    finally:
+        await sbx.kill()

packages/python-sdk/tests/async/sandbox_async/files/test_write.py

@@ -120,3 +120,24 @@ async def test_write_to_non_existing_directory(async_sandbox: AsyncSandbox):
 
     read_content = await async_sandbox.files.read(filename)
     assert read_content == content
+
+async def test_write_with_secured_envd(template):
+    filename = "non_existing_dir/test_write.txt"
+    content = "This should succeed too."
+
+    sbx = await AsyncSandbox.create(template, timeout=30, secure=True)
+    try:
+        assert await sbx.is_running()
+        assert sbx._envd_version is not None
+        assert sbx._envd_access_token is not None
+
+        await sbx.files.write(filename, content)
+
+        exists = await sbx.files.exists(filename)
+        assert exists
+
+        read_content = await sbx.files.read(filename)
+        assert read_content == content
+
+    finally:
+        await sbx.kill()

packages/python-sdk/tests/async/sandbox_async/test_secure.py

@@ -0,0 +1,12 @@
+import pytest
+
+from e2b import AsyncSandbox
+
+async def test_start_secured(template):
+    sbx = await AsyncSandbox.create(template, timeout=5, secure=True)
+    try:
+        assert await sbx.is_running()
+        assert sbx._envd_version is not None
+        assert sbx._envd_access_token is not None
+    finally:
+        await sbx.kill()

packages/python-sdk/tests/sync/sandbox_sync/files/test_watch.py

@@ -109,3 +109,11 @@ def test_watch_file(sandbox: Sandbox):
 
     with pytest.raises(SandboxException):
         sandbox.files.watch_dir(filename)
+
+def test_watch_file_with_secured_envd(template):
+    sbx = Sandbox(template, timeout=30, secrue=True)
+    try:
+        sbx.files.watch_dir("/home/user/")
+        sbx.files.write("test_watch.txt", "This file will be watched.")
+    finally:
+        sbx.kill()

packages/python-sdk/tests/sync/sandbox_sync/files/test_write.py

@@ -1,6 +1,7 @@
 import io
 
 from e2b.sandbox.filesystem.filesystem import EntryInfo
+from e2b.sandbox_sync.main import Sandbox
 
 
 def test_write_text_file(sandbox):
@@ -119,3 +120,25 @@ def test_write_to_non_existing_directory(sandbox):
 
     read_content = sandbox.files.read(filename)
     assert read_content == content
+
+def test_write_with_secured_envd(template):
+    filename = "non_existing_dir/test_write.txt"
+    content = "This should succeed too."
+
+    sbx = Sandbox(template, timeout=30, secrue=True)
+    try:
+        assert sbx.is_running()
+        assert sbx._envd_version is not None
+        assert sbx._envd_access_token is not None
+
+        sbx.files.write(filename, content)
+
+        exists = sbx.files.exists(filename)
+        assert exists
+
+        read_content = sbx.files.read(filename)
+        assert read_content == content
+
+    finally:
+        sbx.kill()
+

packages/python-sdk/tests/sync/sandbox_sync/test_secure.py

@@ -0,0 +1,13 @@
+import pytest
+
+from e2b import Sandbox
+
+
+def test_start_secured(template):
+    sbx = Sandbox(template, timeout=5, secrue=True)
+    try:
+        assert sbx.is_running()
+        assert sbx._envd_version is not None
+        assert sbx._envd_access_token is not None
+    finally:
+        sbx.kill()