Coverage for agentos/tests/test_sandbox_executor.py: 0%

1"""测试 sandbox_executor — 进程级和 Docker 沙箱执行。""" 

2 

3import os 

4import tempfile 

5import pytest 

6from agentos.security.sandbox import SandboxExecutor, SandboxConfig, SandboxMode, SandboxResult 

7 

8 

9class TestProcessSandbox: 

10 def test_basic_python_execution(self): 

11 config = SandboxConfig(mode=SandboxMode.PROCESS, timeout_seconds=5) 

12 sb = SandboxExecutor(config) 

13 result = sb._sandbox.execute_code("print('hello from sandbox')") 

14 assert result.success 

15 assert result.exit_code == 0 

16 assert "hello from sandbox" in result.stdout 

17 sb._sandbox.cleanup() 

18 

19 def test_basic_bash_execution(self): 

20 config = SandboxConfig(mode=SandboxMode.PROCESS, timeout_seconds=5) 

21 sb = SandboxExecutor(config) 

22 result = sb._sandbox.execute_code("echo 'bash test'", language="bash") 

23 assert result.success 

24 assert "bash test" in result.stdout 

25 sb._sandbox.cleanup() 

26 

27 def test_code_with_error(self): 

28 config = SandboxConfig(mode=SandboxMode.PROCESS, timeout_seconds=5) 

29 sb = SandboxExecutor(config) 

30 result = sb._sandbox.execute_code("raise ValueError('test error')") 

31 assert not result.success 

32 assert result.exit_code != 0 

33 sb._sandbox.cleanup() 

34 

35 def test_timeout(self): 

36 config = SandboxConfig(mode=SandboxMode.PROCESS, timeout_seconds=0.2) 

37 sb = SandboxExecutor(config) 

38 result = sb._sandbox.execute_code("import time; time.sleep(10)") 

39 assert not result.success 

40 assert result.exit_code == -1 

41 assert "Timeout" in (result.error or "") 

42 sb._sandbox.cleanup() 

43 

44 def test_stdout_truncation(self): 

45 config = SandboxConfig(mode=SandboxMode.PROCESS, max_output_bytes=50, timeout_seconds=5) 

46 sb = SandboxExecutor(config) 

47 result = sb._sandbox.execute_code("print('A' * 200)") 

48 assert result.truncated 

49 assert len(result.stdout) <= 50 + len("\n... [stdout truncated]") 

50 sb._sandbox.cleanup() 

51 

52 def test_input_files(self): 

53 config = SandboxConfig(mode=SandboxMode.PROCESS, timeout_seconds=5) 

54 sb = SandboxExecutor(config) 

55 

56 # create a temp input file 

57 with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: 

58 f.write("input data") 

59 input_path = f.name 

60 

61 result = sb._sandbox.execute_code( 

62 "with open('input.txt') as f: print(f.read())", 

63 input_files={"input.txt": input_path}, 

64 ) 

65 assert result.success 

66 assert "input data" in result.stdout 

67 

68 os.unlink(input_path) 

69 sb._sandbox.cleanup() 

70 

71 def test_output_collection(self): 

72 config = SandboxConfig(mode=SandboxMode.PROCESS, timeout_seconds=5) 

73 sb = SandboxExecutor(config) 

74 

75 result = sb._sandbox.execute_code( 

76 "with open('output.txt', 'w') as f: f.write('generated'); " 

77 "import json; json.dump({'a': 1}, open('data.json', 'w'))" 

78 ) 

79 assert result.success 

80 

81 files = sb._sandbox.collect_output_files([".txt", ".json"]) 

82 assert "output.txt" in files 

83 assert "data.json" in files 

84 assert os.path.exists(files["output.txt"]) 

85 assert os.path.exists(files["data.json"]) 

86 

87 # verify content 

88 with open(files["output.txt"]) as f: 

89 assert f.read() == "generated" 

90 sb._sandbox.cleanup() 

91 

92 def test_execute_command(self): 

93 config = SandboxConfig(mode=SandboxMode.PROCESS, timeout_seconds=5) 

94 sb = SandboxExecutor(config) 

95 result = sb._sandbox.execute_command("echo cmd_test && ls") 

96 assert result.success 

97 assert "cmd_test" in result.stdout 

98 sb._sandbox.cleanup() 

99 

100 def test_context_manager(self): 

101 config = SandboxConfig(mode=SandboxMode.PROCESS) 

102 sb = SandboxExecutor(config) 

103 result = None 

104 with sb: 

105 result = sb._sandbox.execute_code("x = 1 + 1; print(x)") 

106 assert result is not None 

107 assert result.success 

108 assert "2" in result.stdout 

109 

110 

111class TestSandboxConfig: 

112 def test_defaults(self): 

113 config = SandboxConfig() 

114 assert config.mode == SandboxMode.PROCESS 

115 assert config.memory_limit_mb == 256 

116 assert config.timeout_seconds == 30 

117 

118 def test_custom(self): 

119 config = SandboxConfig( 

120 mode=SandboxMode.DOCKER, 

121 memory_limit_mb=512, 

122 timeout_seconds=10, 

123 network_enabled=True, 

124 ) 

125 assert config.mode == SandboxMode.DOCKER 

126 assert config.memory_limit_mb == 512 

127 assert config.timeout_seconds == 10 

128 assert config.network_enabled 

129 

130 

131class TestDockerFallback: 

132 """Docker 不可用时自动降级到 Process 模式。""" 

133 

134 def test_docker_fallback_when_no_docker(self): 

135 # 该环境可能没有 Docker，应自动降级 

136 config = SandboxConfig(mode=SandboxMode.DOCKER, timeout_seconds=3) 

137 executor = SandboxExecutor(config) 

138 # 内部已将 mode 调整或降级 

139 result = executor._sandbox.execute_code("print('fallback works')") 

140 assert result.success 

141 executor._sandbox.cleanup()