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

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

2 

3import os 

4import tempfile 

5from agentos.security.sandbox import SandboxExecutor, SandboxConfig, SandboxMode 

6 

7 

8class TestProcessSandbox: 

9 def test_basic_python_execution(self): 

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

11 sb = SandboxExecutor(config) 

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

13 assert result.success 

14 assert result.exit_code == 0 

15 assert "hello from sandbox" in result.stdout 

16 sb._sandbox.cleanup() 

17 

18 def test_basic_bash_execution(self): 

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

20 sb = SandboxExecutor(config) 

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

22 assert result.success 

23 assert "bash test" in result.stdout 

24 sb._sandbox.cleanup() 

25 

26 def test_code_with_error(self): 

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

28 sb = SandboxExecutor(config) 

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

30 assert not result.success 

31 assert result.exit_code != 0 

32 sb._sandbox.cleanup() 

33 

34 def test_timeout(self): 

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

36 sb = SandboxExecutor(config) 

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

38 assert not result.success 

39 assert result.exit_code == -1 

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

41 sb._sandbox.cleanup() 

42 

43 def test_stdout_truncation(self): 

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

45 sb = SandboxExecutor(config) 

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

47 assert result.truncated 

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

49 sb._sandbox.cleanup() 

50 

51 def test_input_files(self): 

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

53 sb = SandboxExecutor(config) 

54 

55 # create a temp input file 

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

57 f.write("input data") 

58 input_path = f.name 

59 

60 result = sb._sandbox.execute_code( 

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

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

63 ) 

64 assert result.success 

65 assert "input data" in result.stdout 

66 

67 os.unlink(input_path) 

68 sb._sandbox.cleanup() 

69 

70 def test_output_collection(self): 

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

72 sb = SandboxExecutor(config) 

73 

74 result = sb._sandbox.execute_code( 

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

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

77 ) 

78 assert result.success 

79 

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

81 assert "output.txt" in files 

82 assert "data.json" in files 

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

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

85 

86 # verify content 

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

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

89 sb._sandbox.cleanup() 

90 

91 def test_execute_command(self): 

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

93 sb = SandboxExecutor(config) 

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

95 assert result.success 

96 assert "cmd_test" in result.stdout 

97 sb._sandbox.cleanup() 

98 

99 def test_context_manager(self): 

100 config = SandboxConfig(mode=SandboxMode.PROCESS) 

101 sb = SandboxExecutor(config) 

102 result = None 

103 with sb: 

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

105 assert result is not None 

106 assert result.success 

107 assert "2" in result.stdout 

108 

109 

110class TestSandboxConfig: 

111 def test_defaults(self): 

112 config = SandboxConfig() 

113 assert config.mode == SandboxMode.PROCESS 

114 assert config.memory_limit_mb == 256 

115 assert config.timeout_seconds == 30 

116 

117 def test_custom(self): 

118 config = SandboxConfig( 

119 mode=SandboxMode.DOCKER, 

120 memory_limit_mb=512, 

121 timeout_seconds=10, 

122 network_enabled=True, 

123 ) 

124 assert config.mode == SandboxMode.DOCKER 

125 assert config.memory_limit_mb == 512 

126 assert config.timeout_seconds == 10 

127 assert config.network_enabled 

128 

129 

130class TestDockerFallback: 

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

132 

133 def test_docker_fallback_when_no_docker(self): 

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

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

136 executor = SandboxExecutor(config) 

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

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

139 assert result.success 

140 executor._sandbox.cleanup()