diff --git a/test/test_streamlayerformat.py b/test/test_streamlayerformat.py new file mode 100644 index 000000000..056afa4ce --- /dev/null +++ b/test/test_streamlayerformat.py @@ -0,0 +1,143 @@ +import unittest +import tarfile + +from StringIO import StringIO +from util.streamlayerformat import StreamLayerMerger, AUFS_WHITEOUT + +class TestStreamLayerMerger(unittest.TestCase): + def create_layer(self, **kwargs): + output = StringIO() + with tarfile.open(fileobj=output, mode='w:gz') as tar: + for filename in kwargs: + current_filename = filename + current_contents = kwargs[filename] + + if current_contents is None: + # This is a deleted file. + current_filename = AUFS_WHITEOUT + current_filename + current_contents = '' + + info = tarfile.TarInfo(name=current_filename) + info.size = len(current_contents) + tar.addfile(info, fileobj=StringIO(current_contents)) + + return output.getvalue() + + def squash_layers(self, layers): + def get_layers(): + return [StringIO(layer) for layer in layers] + + merger = StreamLayerMerger(get_layers) + merged_data = ''.join(merger.get_generator()) + return merged_data + + def assertHasFile(self, squashed, filename, contents): + with tarfile.open(fileobj=StringIO(squashed), mode='r:*') as tar: + member = tar.getmember(filename) + self.assertEquals(contents, '\n'.join(tar.extractfile(member).readlines())) + + def assertDoesNotHaveFile(self, squashed, filename): + with tarfile.open(fileobj=StringIO(squashed), mode='r:*') as tar: + try: + member = tar.getmember(filename) + self.fail('Filename %s found' % filename) + except: + pass + + def test_single_layer(self): + tar_layer = self.create_layer( + some_file = 'foo', + another_file = 'bar', + third_file = 'meh') + + squashed = self.squash_layers([tar_layer]) + + self.assertHasFile(squashed, 'some_file', 'foo') + self.assertHasFile(squashed, 'another_file', 'bar') + self.assertHasFile(squashed, 'third_file', 'meh') + + def test_multiple_layers(self): + second_layer = self.create_layer( + some_file = 'foo', + another_file = 'bar', + third_file = 'meh') + + first_layer = self.create_layer( + top_file = 'top') + + squashed = self.squash_layers([first_layer, second_layer]) + + self.assertHasFile(squashed, 'some_file', 'foo') + self.assertHasFile(squashed, 'another_file', 'bar') + self.assertHasFile(squashed, 'third_file', 'meh') + self.assertHasFile(squashed, 'top_file', 'top') + + def test_multiple_layers_overwrite(self): + second_layer = self.create_layer( + some_file = 'foo', + another_file = 'bar', + third_file = 'meh') + + first_layer = self.create_layer( + another_file = 'top') + + squashed = self.squash_layers([first_layer, second_layer]) + + self.assertHasFile(squashed, 'some_file', 'foo') + self.assertHasFile(squashed, 'third_file', 'meh') + self.assertHasFile(squashed, 'another_file', 'top') + + def test_deleted_file(self): + second_layer = self.create_layer( + some_file = 'foo', + another_file = 'bar', + third_file = 'meh') + + first_layer = self.create_layer( + another_file = None) + + squashed = self.squash_layers([first_layer, second_layer]) + + self.assertHasFile(squashed, 'some_file', 'foo') + self.assertHasFile(squashed, 'third_file', 'meh') + self.assertDoesNotHaveFile(squashed, 'another_file') + + def test_deleted_readded_file(self): + third_layer = self.create_layer( + another_file = 'bar') + + second_layer = self.create_layer( + some_file = 'foo', + another_file = None, + third_file = 'meh') + + first_layer = self.create_layer( + another_file = 'newagain') + + squashed = self.squash_layers([first_layer, second_layer, third_layer]) + + self.assertHasFile(squashed, 'some_file', 'foo') + self.assertHasFile(squashed, 'third_file', 'meh') + self.assertHasFile(squashed, 'another_file', 'newagain') + + def test_deleted_in_lower_layer(self): + third_layer = self.create_layer( + another_file = 'bar') + + second_layer = self.create_layer( + some_file = 'foo', + another_file = None, + third_file = 'meh') + + first_layer = self.create_layer( + top_file = 'top') + + squashed = self.squash_layers([first_layer, second_layer, third_layer]) + + self.assertHasFile(squashed, 'some_file', 'foo') + self.assertHasFile(squashed, 'third_file', 'meh') + self.assertHasFile(squashed, 'top_file', 'top') + self.assertDoesNotHaveFile(squashed, 'another_file') + +if __name__ == '__main__': + unittest.main()