(self)
| 116 | assert res == b'hello\nworld\n' |
| 117 | |
| 118 | def test_exec_start_socket(self): |
| 119 | container = self.client.create_container(TEST_IMG, 'cat', |
| 120 | detach=True, stdin_open=True) |
| 121 | container_id = container['Id'] |
| 122 | self.client.start(container_id) |
| 123 | self.tmp_containers.append(container_id) |
| 124 | |
| 125 | line = 'yay, interactive exec!' |
| 126 | # `echo` appends CRLF, `printf` doesn't |
| 127 | exec_id = self.client.exec_create( |
| 128 | container_id, ['printf', line], tty=True) |
| 129 | assert 'Id' in exec_id |
| 130 | |
| 131 | socket = self.client.exec_start(exec_id, socket=True) |
| 132 | self.addCleanup(socket.close) |
| 133 | |
| 134 | (stream, next_size) = next_frame_header(socket) |
| 135 | assert stream == 1 # stdout (0 = stdin, 1 = stdout, 2 = stderr) |
| 136 | assert next_size == len(line) |
| 137 | data = read_exactly(socket, next_size) |
| 138 | assert data.decode('utf-8') == line |
| 139 | |
| 140 | def test_exec_start_detached(self): |
| 141 | container = self.client.create_container(TEST_IMG, 'cat', |
nothing calls this directly
no test coverage detected