(self)
| 101 | assert exec_log == b'root\n' |
| 102 | |
| 103 | def test_exec_command_streaming(self): |
| 104 | container = self.client.create_container(TEST_IMG, 'cat', |
| 105 | detach=True, stdin_open=True) |
| 106 | id = container['Id'] |
| 107 | self.tmp_containers.append(id) |
| 108 | self.client.start(id) |
| 109 | |
| 110 | exec_id = self.client.exec_create(id, ['echo', 'hello\nworld']) |
| 111 | assert 'Id' in exec_id |
| 112 | |
| 113 | res = b'' |
| 114 | for chunk in self.client.exec_start(exec_id, stream=True): |
| 115 | res += chunk |
| 116 | assert res == b'hello\nworld\n' |
| 117 | |
| 118 | def test_exec_start_socket(self): |
| 119 | container = self.client.create_container(TEST_IMG, 'cat', |
nothing calls this directly
no test coverage detected