Download file from docker container
(self, container_file_path: str, host_file_path: str)
| 539 | return f"{self.tmp_dir}/{file_name}" |
| 540 | |
| 541 | def copy_from(self, container_file_path: str, host_file_path: str): |
| 542 | """Download file from docker container""" |
| 543 | tar_bytes_gen, _ = self._container.get_archive(container_file_path) |
| 544 | |
| 545 | # convert to bytes |
| 546 | tar_bytes = b"" |
| 547 | for chunk in tar_bytes_gen: |
| 548 | tar_bytes += chunk |
| 549 | |
| 550 | tar = tarfile.open(fileobj=io.BytesIO(initial_bytes=tar_bytes)) |
| 551 | assert len(tar.getmembers()) == 1 |
| 552 | tar_element_reader = tar.extractfile(tar.getmembers()[0]) |
| 553 | with open(host_file_path, "wb") as host_file: |
| 554 | for chunk in tar_element_reader: |
| 555 | host_file.write(chunk) |
| 556 | |
| 557 | def close(self): |
| 558 | """Close docker container session""" |
no test coverage detected