Finish file/dir tests
This commit is contained in:
@@ -6,7 +6,7 @@ import sys
|
||||
import uuid
|
||||
from os import PathLike
|
||||
from pathlib import Path, PurePath
|
||||
from typing import IO, Dict, List, Optional, Sequence, Union
|
||||
from typing import Any, IO, Dict, List, Optional, Sequence, Union
|
||||
|
||||
|
||||
class DockerContainer:
|
||||
@@ -77,16 +77,17 @@ class DockerContainer:
|
||||
# a container is running and the host filesystem is
|
||||
# mounted. https://github.com/moby/moby/issues/38995
|
||||
# Use `docker exec` instead.
|
||||
quote = lambda p: shlex.quote(str(p))
|
||||
if from_path.is_dir():
|
||||
self.call(['mkdir', '-p', to_path])
|
||||
subprocess.run(
|
||||
f'tar cf - . | docker exec -i {self.name} tar -xC {to_path} -f -',
|
||||
f'tar cf - . | docker exec -i {self.name} tar -xC {quote(to_path)} -f -',
|
||||
shell=True,
|
||||
check=True,
|
||||
cwd=from_path)
|
||||
else:
|
||||
subprocess.run(
|
||||
f'cat {from_path} | docker exec -i {self.name} sh -c "cat > {to_path}"',
|
||||
f'cat {quote(from_path)} | docker exec -i {self.name} sh -c "cat > {quote(to_path)}"',
|
||||
shell=True,
|
||||
check=True)
|
||||
|
||||
@@ -94,8 +95,9 @@ class DockerContainer:
|
||||
# note: we assume from_path is a dir
|
||||
to_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
quote = lambda p: shlex.quote(str(p))
|
||||
subprocess.run(
|
||||
f'docker exec -i {self.name} tar -cC {from_path} -f - . | tar -xf -',
|
||||
f'docker exec -i {self.name} tar -cC {quote(from_path)} -f - . | tar -xf -',
|
||||
shell=True,
|
||||
check=True,
|
||||
cwd=to_path
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
from pathlib import Path, PurePath
|
||||
import platform
|
||||
import random
|
||||
import shutil
|
||||
import subprocess
|
||||
import textwrap
|
||||
from uuid import uuid4
|
||||
@@ -104,8 +106,7 @@ def test_binary_output():
|
||||
def test_file_operations(tmp_path: Path):
|
||||
with DockerContainer(DEFAULT_IMAGE) as container:
|
||||
# test copying a file in
|
||||
test_binary_data = uuid4().bytes + uuid4().bytes + uuid4().bytes + uuid4().bytes
|
||||
|
||||
test_binary_data = bytes(random.randrange(256) for _ in range(1000))
|
||||
original_test_file = tmp_path / 'test.dat'
|
||||
original_test_file.write_bytes(test_binary_data)
|
||||
|
||||
@@ -116,11 +117,32 @@ def test_file_operations(tmp_path: Path):
|
||||
output = container.call(['cat', dst_file], capture_output=True)
|
||||
assert test_binary_data == bytes(output, encoding='utf8', errors='surrogateescape')
|
||||
|
||||
@pytest.mark.docker
|
||||
def test_dir_operations(tmp_path: Path):
|
||||
with DockerContainer(DEFAULT_IMAGE) as container:
|
||||
test_binary_data = bytes(random.randrange(256) for _ in range(1000))
|
||||
original_test_file = tmp_path / 'test.dat'
|
||||
original_test_file.write_bytes(test_binary_data)
|
||||
|
||||
# test copying a dir in
|
||||
test_dir = tmp_path / 'test_dir'
|
||||
new_test_file = tmp_path / 'test-new.dat'
|
||||
container.copy_out(dst_file, new_test_file)
|
||||
test_dir.mkdir()
|
||||
test_file = test_dir / 'test.dat'
|
||||
shutil.copyfile(original_test_file, test_file)
|
||||
|
||||
assert original_test_file.read_bytes() == new_test_file.read_bytes()
|
||||
dst_dir = PurePath('/tmp/test_dir')
|
||||
dst_file = dst_dir / 'test.dat'
|
||||
container.copy_into(test_dir, dst_dir)
|
||||
|
||||
output = container.call(['cat', dst_file], capture_output=True)
|
||||
assert test_binary_data == bytes(output, encoding='utf8', errors='surrogateescape')
|
||||
|
||||
# test glob
|
||||
assert container.glob(dst_dir / '*.dat') == [dst_file]
|
||||
|
||||
# test copy dir out
|
||||
new_test_dir = tmp_path / 'test_dir_new'
|
||||
container.copy_out(dst_dir, new_test_dir)
|
||||
|
||||
assert test_binary_data == (new_test_dir / 'test.dat').read_bytes()
|
||||
|
||||
assert container.glob(PurePath('/tmp/*.dat')) == [dst_file]
|
||||
|
||||
Reference in New Issue
Block a user