docker
/
docker-py
mirror of https://github.com/docker/docker-py.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Properly close files and sockets in unit tests

This commit is contained in:
Roman Mohr 2015-02-28 20:55:45 +01:00
parent 39ee5981b4
commit a86bfffd89
1 changed files with 30 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
tests/test.py
View File

@ -2115,11 +2115,10 @@ class DockerClientTest(Cleanup, unittest.TestCase):
folder = tempfile.mkdtemp() folder = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, folder) self.addCleanup(shutil.rmtree, folder)
dockercfg_path = os.path.join(folder, '.dockercfg') dockercfg_path = os.path.join(folder, '.dockercfg')
f = open(dockercfg_path, 'w') with open(dockercfg_path, 'w') as f:
auth_ = base64.b64encode(b'sakuya:izayoi').decode('ascii') auth_ = base64.b64encode(b'sakuya:izayoi').decode('ascii')
f.write('auth = {0}\n'.format(auth_)) f.write('auth = {0}\n'.format(auth_))
f.write('email = sakuya@scarlet.net') f.write('email = sakuya@scarlet.net')
f.close()
cfg = docker.auth.load_config(dockercfg_path) cfg = docker.auth.load_config(dockercfg_path)
self.assertTrue(docker.auth.INDEX_URL in cfg) self.assertTrue(docker.auth.INDEX_URL in cfg)
self.assertNotEqual(cfg[docker.auth.INDEX_URL], None) self.assertNotEqual(cfg[docker.auth.INDEX_URL], None)
@ -2175,18 +2174,18 @@ class DockerClientTest(Cleanup, unittest.TestCase):
(['test/foo', 'a.txt'], ['bar', 'bar/a.txt', 'bar/b.py', (['test/foo', 'a.txt'], ['bar', 'bar/a.txt', 'bar/b.py',
'bar/other.png', 'test']), 'bar/other.png', 'test']),
): ):
archive = docker.utils.tar(base, exclude=exclude) with docker.utils.tar(base, exclude=exclude) as archive:
tar = tarfile.open(fileobj=archive) tar = tarfile.open(fileobj=archive)
self.assertEqual(sorted(tar.getnames()), names) self.assertEqual(sorted(tar.getnames()), names)
def test_tar_with_empty_directory(self): def test_tar_with_empty_directory(self):
base = tempfile.mkdtemp() base = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base) self.addCleanup(shutil.rmtree, base)
for d in ['foo', 'bar']: for d in ['foo', 'bar']:
os.makedirs(os.path.join(base, d)) os.makedirs(os.path.join(base, d))
archive = docker.utils.tar(base) with docker.utils.tar(base) as archive:
tar = tarfile.open(fileobj=archive) tar = tarfile.open(fileobj=archive)
self.assertEqual(sorted(tar.getnames()), ['bar', 'foo']) self.assertEqual(sorted(tar.getnames()), ['bar', 'foo'])
def test_tar_with_file_symlinks(self): def test_tar_with_file_symlinks(self):
base = tempfile.mkdtemp() base = tempfile.mkdtemp()
@ -2195,9 +2194,9 @@ class DockerClientTest(Cleanup, unittest.TestCase):
f.write("content") f.write("content")
os.makedirs(os.path.join(base, 'bar')) os.makedirs(os.path.join(base, 'bar'))
os.symlink('../foo', os.path.join(base, 'bar/foo')) os.symlink('../foo', os.path.join(base, 'bar/foo'))
archive = docker.utils.tar(base) with docker.utils.tar(base) as archive:
tar = tarfile.open(fileobj=archive) tar = tarfile.open(fileobj=archive)
self.assertEqual(sorted(tar.getnames()), ['bar', 'bar/foo', 'foo']) self.assertEqual(sorted(tar.getnames()), ['bar', 'bar/foo', 'foo'])
def test_tar_with_directory_symlinks(self): def test_tar_with_directory_symlinks(self):
base = tempfile.mkdtemp() base = tempfile.mkdtemp()
@ -2205,9 +2204,9 @@ class DockerClientTest(Cleanup, unittest.TestCase):
for d in ['foo', 'bar']: for d in ['foo', 'bar']:
os.makedirs(os.path.join(base, d)) os.makedirs(os.path.join(base, d))
os.symlink('../foo', os.path.join(base, 'bar/foo')) os.symlink('../foo', os.path.join(base, 'bar/foo'))
archive = docker.utils.tar(base) with docker.utils.tar(base) as archive:
tar = tarfile.open(fileobj=archive) tar = tarfile.open(fileobj=archive)
self.assertEqual(sorted(tar.getnames()), ['bar', 'bar/foo', 'foo']) self.assertEqual(sorted(tar.getnames()), ['bar', 'bar/foo', 'foo'])
class StreamTest(Cleanup, unittest.TestCase): class StreamTest(Cleanup, unittest.TestCase):
@ -2293,21 +2292,21 @@ class StreamTest(Cleanup, unittest.TestCase):
b'\r\n' b'\r\n'
) + b'\r\n'.join(lines) ) + b'\r\n'.join(lines)
client = docker.Client(base_url="http+unix://" + self.socket_file) with docker.Client(base_url="http+unix://" + self.socket_file) \
for i in range(5): as client:
try: for i in range(5):
stream = client.build( try:
path=self.build_context, stream = client.build(
stream=True path=self.build_context,
) stream=True
break )
except requests.ConnectionError as e: break
if i == 4: except requests.ConnectionError as e:
raise e if i == 4:
raise e
self.assertEqual(list(stream), [
str(i).encode() for i in range(50)])
self.assertEqual(list(stream), [
str(i).encode() for i in range(50)])
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()