import io import uuid from e2b.sandbox.filesystem.filesystem import FileType, WriteInfo, WriteEntry def test_write_text_file(sandbox, debug): filename = "test_write.txt" content = "This is a test file." info = sandbox.files.write(filename, content) assert info.path == f"/home/user/{filename}" assert info.type == FileType.FILE exists = sandbox.files.exists(filename) assert exists read_content = sandbox.files.read(filename) assert read_content == content if debug: sandbox.files.remove(filename) def test_write_binary_file(sandbox, debug): filename = "test_write.bin" text = "This is a test binary file." # equivalent to `open("path/to/local/file", "rb")` content = io.BytesIO(text.encode("utf-8")) info = sandbox.files.write(filename, content) assert info.path == f"/home/user/{filename}" exists = sandbox.files.exists(filename) assert exists read_content = sandbox.files.read(filename) assert read_content == text if debug: sandbox.files.remove(filename) def test_write_multiple_files(sandbox, debug): num_test_files = 10 # Attempt to write with empty files array empty_info = sandbox.files.write_files([]) assert isinstance(empty_info, list) assert len(empty_info) == 0 # Attempt to write with no files assert sandbox.files.write_files([]) == [] # Attempt to write with one file in array one_file_path = "one_test_file.txt" info = sandbox.files.write_files( [WriteEntry(path=one_file_path, data="This is a test file.")] ) assert isinstance(info, list) assert len(info) == 1 info = info[0] assert isinstance(info, WriteInfo) assert info.path == "/home/user/one_test_file.txt" exists = sandbox.files.exists(info.path) assert exists read_content = sandbox.files.read(info.path) assert read_content == "This is a test file." # Attempt to write with multiple files in array files = [] for i in range(num_test_files): path = f"test_write_{i}.txt" content = f"This is a test file {i}." files.append(WriteEntry(path=path, data=content)) infos = sandbox.files.write_files(files) assert isinstance(infos, list) assert len(infos) == len(files) for i, info in enumerate(infos): assert isinstance(info, WriteInfo) assert info.path == f"/home/user/test_write_{i}.txt" assert info.type == FileType.FILE exists = sandbox.files.exists(path) assert exists read_content = sandbox.files.read(info.path) assert read_content == files[i]["data"] if debug: sandbox.files.remove(one_file_path) for i in range(num_test_files): sandbox.files.remove(f"test_write_{i}.txt") def test_overwrite_file(sandbox, debug): filename = "test_overwrite.txt" initial_content = "Initial content." new_content = "New content." sandbox.files.write(filename, initial_content) sandbox.files.write(filename, new_content) read_content = sandbox.files.read(filename) assert read_content == new_content if debug: sandbox.files.remove(filename) def test_write_to_non_existing_directory(sandbox, debug): filename = "non_existing_dir/test_write.txt" content = "This should succeed too." sandbox.files.write(filename, content) exists = sandbox.files.exists(filename) assert exists read_content = sandbox.files.read(filename) assert read_content == content if debug: sandbox.files.remove(filename) def test_write_with_secured_envd(sandbox_factory): filename = f"non_existing_dir_{uuid.uuid4()}/test_write.txt" content = "This should succeed too." sbx = sandbox_factory(timeout=30, secure=True) 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 def test_write_files_with_different_data_types(sandbox, debug): text_data = "Text string data" bytes_data = b"Bytes data" bytes_io_data = io.BytesIO(b"BytesIO data") string_io_data = io.StringIO("StringIO data") files = [ WriteEntry(path="writefiles_text.txt", data=text_data), WriteEntry(path="writefiles_bytes.bin", data=bytes_data), WriteEntry(path="writefiles_bytesio.bin", data=bytes_io_data), WriteEntry(path="writefiles_stringio.txt", data=string_io_data), ] infos = sandbox.files.write_files(files) assert len(infos) == 4 text_content = sandbox.files.read("writefiles_text.txt") assert text_content == text_data bytes_content = sandbox.files.read("writefiles_bytes.bin") assert bytes_content == "Bytes data" bytes_io_content = sandbox.files.read("writefiles_bytesio.bin") assert bytes_io_content == "BytesIO data" string_io_content = sandbox.files.read("writefiles_stringio.txt") assert string_io_content == "StringIO data" if debug: for file in files: sandbox.files.remove(file["path"]) def test_write_io_with_octet_stream(sandbox, debug): filename = "test_write_octet_io.bin" text = "Streamed octet-stream upload. " * 10_000 content = io.BytesIO(text.encode("utf-8")) info = sandbox.files.write(filename, content, use_octet_stream=True) assert info.path == f"/home/user/{filename}" read_content = sandbox.files.read(filename) assert read_content == text if debug: sandbox.files.remove(filename) def test_write_text_io_with_octet_stream(sandbox, debug): filename = "test_write_octet_text_io.txt" text = "Streamed text octet-stream upload." sandbox.files.write(filename, io.StringIO(text), use_octet_stream=True) read_content = sandbox.files.read(filename) assert read_content == text if debug: sandbox.files.remove(filename) def test_write_io_with_octet_stream_and_gzip(sandbox, debug): filename = "test_write_octet_io_gzip.bin" text = "Streamed gzipped octet-stream upload. " * 10_000 content = io.BytesIO(text.encode("utf-8")) sandbox.files.write(filename, content, use_octet_stream=True, gzip=True) read_content = sandbox.files.read(filename) assert read_content == text if debug: sandbox.files.remove(filename)