quay/test/test_changes.py

import unittest
import tarfile

from StringIO import StringIO
from util.registry.streamlayerformat import AUFS_WHITEOUT
from util.registry.changes import files_and_dirs_from_tar

class TestChanges(unittest.TestCase):
  def create_layer(self, **kwargs):
    output = StringIO()
    with tarfile.open(fileobj=output, mode='w:gz') as tar:
      for current_contents in kwargs:
        current_filename = kwargs[current_contents]

        if current_contents == '_':
          # This is a deleted file.
          if current_filename.endswith('/'):
            current_filename = current_filename[:-1]

          parts = current_filename.split('/')
          if len(parts) > 1:
            current_filename = '/'.join(parts[:-1]) + '/' + AUFS_WHITEOUT + parts[-1]
          else:
            current_filename = AUFS_WHITEOUT + parts[-1]

          current_contents = ''

        info = tarfile.TarInfo(name=current_filename)
        info.size = len(current_contents)
        tar.addfile(info, fileobj=StringIO(current_contents))

    return output.getvalue()


  def test_single_layer(self):
    tar_layer = self.create_layer(
      foo = 'some_file',
      bar = 'another_file',
      meh = 'third_file')

    deleted_prefixes = set()
    result = list(files_and_dirs_from_tar(StringIO(tar_layer), deleted_prefixes))

    self.assertIn('/some_file', result)
    self.assertIn('/another_file', result)
    self.assertIn('/third_file', result)

    self.assertFalse(deleted_prefixes)


  def test_delete_file(self):
    tar_layer = self.create_layer(
      foo = 'some_file',
      _ = 'another_file',
      meh = 'third_file')

    deleted_prefixes = set()
    result = list(files_and_dirs_from_tar(StringIO(tar_layer), deleted_prefixes))

    self.assertIn('/some_file', result)
    self.assertIn('/third_file', result)

    self.assertIn('another_file', deleted_prefixes)

  def test_pax_tarfile_error(self):
    # If this fails it will raise an exception
    with open('test/data/sample/layers/paxheader.tar') as tarstream:
      result = list(files_and_dirs_from_tar(tarstream, set()))

    self.assertIn('/file', result)

if __name__ == '__main__':
  unittest.main()