gecko/python/mozbuild/mozpack/test/test_files.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

from mozpack.files import (
    Dest,
    File,
    GeneratedFile,
    DeflatedFile,
    ManifestFile,
    XPTFile,
    MinifiedProperties,
    FileFinder,
    JarFinder,
)
from mozpack.mozjar import (
    JarReader,
    JarWriter,
)
from mozpack.chrome.manifest import (
    ManifestContent,
    ManifestResource,
    ManifestLocale,
    ManifestOverride,
)
import unittest
import mozunit
import os
import shutil
import random
import string
import mozpack.path
from mozpack.copier import ensure_parent_dir
from tempfile import mkdtemp
from io import BytesIO
from xpt import Typelib


class TestWithTmpDir(unittest.TestCase):
    def setUp(self):
        self.tmpdir = mkdtemp()

    def tearDown(self):
        shutil.rmtree(self.tmpdir)

    def tmppath(self, relpath):
        return os.path.normpath(os.path.join(self.tmpdir, relpath))


class MockDest(BytesIO, Dest):
    def __init__(self):
        BytesIO.__init__(self)
        self.mode = None

    def read(self, length=-1):
        if self.mode != 'r':
            self.seek(0)
            self.mode = 'r'
        return BytesIO.read(self, length)

    def write(self, data):
        if self.mode != 'w':
            self.seek(0)
            self.truncate(0)
            self.mode = 'w'
        return BytesIO.write(self, data)

    def exists(self):
        return True

    def close(self):
        if self.mode:
            self.mode = None


class DestNoWrite(Dest):
    def write(self, data):
        raise RuntimeError


class TestDest(TestWithTmpDir):
    def test_dest(self):
        dest = Dest(self.tmppath('dest'))
        self.assertFalse(dest.exists())
        dest.write('foo')
        self.assertTrue(dest.exists())
        dest.write('foo')
        self.assertEqual(dest.read(4), 'foof')
        self.assertEqual(dest.read(), 'oo')
        self.assertEqual(dest.read(), '')
        dest.write('bar')
        self.assertEqual(dest.read(4), 'bar')
        dest.close()
        self.assertEqual(dest.read(), 'bar')
        dest.write('foo')
        dest.close()
        dest.write('qux')
        self.assertEqual(dest.read(), 'qux')

rand = ''.join(random.choice(string.letters) for i in xrange(131597))
samples = [
    '',
    'test',
    'fooo',
    'same',
    'same',
    'Different and longer',
    rand,
    rand,
    rand[:-1] + '_',
    'test'
]


class TestFile(TestWithTmpDir):
    def test_file(self):
        '''
        Check that File.copy yields the proper content in the destination file
        in all situations that trigger different code paths:
        - different content
        - different content of the same size
        - same content
        - long content
        '''
        src = self.tmppath('src')
        dest = self.tmppath('dest')

        for content in samples:
            with open(src, 'wb') as tmp:
                tmp.write(content)
            # Ensure the destination file, when it exists, is older than the
            # source
            if os.path.exists(dest):
                time = os.path.getmtime(src) - 1
                os.utime(dest, (time, time))
            f = File(src)
            f.copy(dest)
            self.assertEqual(content, open(dest, 'rb').read())
            self.assertEqual(content, f.open().read())
            self.assertEqual(content, f.open().read())

    def test_file_dest(self):
        '''
        Similar to test_file, but for a destination object instead of
        a destination file. This ensures the destination object is being
        used properly by File.copy, ensuring that other subclasses of Dest
        will work.
        '''
        src = self.tmppath('src')
        dest = MockDest()

        for content in samples:
            with open(src, 'wb') as tmp:
                tmp.write(content)
            f = File(src)
            f.copy(dest)
            self.assertEqual(content, dest.getvalue())

    def test_file_open(self):
        '''
        Test whether File.open returns an appropriately reset file object.
        '''
        src = self.tmppath('src')
        content = ''.join(samples)
        with open(src, 'wb') as tmp:
            tmp.write(content)

        f = File(src)
        self.assertEqual(content[:42], f.open().read(42))
        self.assertEqual(content, f.open().read())

    def test_file_no_write(self):
        '''
        Test various conditions where File.copy is expected not to write
        in the destination file.
        '''
        src = self.tmppath('src')
        dest = self.tmppath('dest')

        with open(src, 'wb') as tmp:
            tmp.write('test')

        # Initial copy
        f = File(src)
        f.copy(dest)

        # Ensure subsequent copies won't trigger writes
        f.copy(DestNoWrite(dest))
        self.assertEqual('test', open(dest, 'rb').read())

        # When the source file is newer, but with the same content, no copy
        # should occur
        time = os.path.getmtime(src) - 1
        os.utime(dest, (time, time))
        f.copy(DestNoWrite(dest))
        self.assertEqual('test', open(dest, 'rb').read())

        # When the source file is older than the destination file, even with
        # different content, no copy should occur.
        with open(src, 'wb') as tmp:
            tmp.write('fooo')
        time = os.path.getmtime(dest) - 1
        os.utime(src, (time, time))
        f.copy(DestNoWrite(dest))
        self.assertEqual('test', open(dest, 'rb').read())

        # Double check that under conditions where a copy occurs, we would get
        # an exception.
        time = os.path.getmtime(src) - 1
        os.utime(dest, (time, time))
        self.assertRaises(RuntimeError, f.copy, DestNoWrite(dest))

        # skip_if_older=False is expected to force a copy in this situation.
        f.copy(dest, skip_if_older=False)
        self.assertEqual('fooo', open(dest, 'rb').read())


class TestGeneratedFile(TestWithTmpDir):
    def test_generated_file(self):
        '''
        Check that GeneratedFile.copy yields the proper content in the
        destination file in all situations that trigger different code paths
        (see TestFile.test_file)
        '''
        dest = self.tmppath('dest')

        for content in samples:
            f = GeneratedFile(content)
            f.copy(dest)
            self.assertEqual(content, open(dest, 'rb').read())

    def test_generated_file_open(self):
        '''
        Test whether GeneratedFile.open returns an appropriately reset file
        object.
        '''
        content = ''.join(samples)
        f = GeneratedFile(content)
        self.assertEqual(content[:42], f.open().read(42))
        self.assertEqual(content, f.open().read())

    def test_generated_file_no_write(self):
        '''
        Test various conditions where GeneratedFile.copy is expected not to
        write in the destination file.
        '''
        dest = self.tmppath('dest')

        # Initial copy
        f = GeneratedFile('test')
        f.copy(dest)

        # Ensure subsequent copies won't trigger writes
        f.copy(DestNoWrite(dest))
        self.assertEqual('test', open(dest, 'rb').read())

        # When using a new instance with the same content, no copy should occur
        f = GeneratedFile('test')
        f.copy(DestNoWrite(dest))
        self.assertEqual('test', open(dest, 'rb').read())

        # Double check that under conditions where a copy occurs, we would get
        # an exception.
        f = GeneratedFile('fooo')
        self.assertRaises(RuntimeError, f.copy, DestNoWrite(dest))


class TestDeflatedFile(TestWithTmpDir):
    def test_deflated_file(self):
        '''
        Check that DeflatedFile.copy yields the proper content in the
        destination file in all situations that trigger different code paths
        (see TestFile.test_file)
        '''
        src = self.tmppath('src.jar')
        dest = self.tmppath('dest')

        contents = {}
        with JarWriter(src) as jar:
            for content in samples:
                name = ''.join(random.choice(string.letters)
                               for i in xrange(8))
                jar.add(name, content, compress=True)
                contents[name] = content

        for j in JarReader(src):
            f = DeflatedFile(j)
            f.copy(dest)
            self.assertEqual(contents[j.filename], open(dest, 'rb').read())

    def test_deflated_file_open(self):
        '''
        Test whether DeflatedFile.open returns an appropriately reset file
        object.
        '''
        src = self.tmppath('src.jar')
        content = ''.join(samples)
        with JarWriter(src) as jar:
            jar.add('content', content)

        f = DeflatedFile(JarReader(src)['content'])
        self.assertEqual(content[:42], f.open().read(42))
        self.assertEqual(content, f.open().read())

    def test_deflated_file_no_write(self):
        '''
        Test various conditions where DeflatedFile.copy is expected not to
        write in the destination file.
        '''
        src = self.tmppath('src.jar')
        dest = self.tmppath('dest')

        with JarWriter(src) as jar:
            jar.add('test', 'test')
            jar.add('test2', 'test')
            jar.add('fooo', 'fooo')

        jar = JarReader(src)
        # Initial copy
        f = DeflatedFile(jar['test'])
        f.copy(dest)

        # Ensure subsequent copies won't trigger writes
        f.copy(DestNoWrite(dest))
        self.assertEqual('test', open(dest, 'rb').read())

        # When using a different file with the same content, no copy should
        # occur
        f = DeflatedFile(jar['test2'])
        f.copy(DestNoWrite(dest))
        self.assertEqual('test', open(dest, 'rb').read())

        # Double check that under conditions where a copy occurs, we would get
        # an exception.
        f = DeflatedFile(jar['fooo'])
        self.assertRaises(RuntimeError, f.copy, DestNoWrite(dest))


class TestManifestFile(TestWithTmpDir):
    def test_manifest_file(self):
        f = ManifestFile('chrome')
        f.add(ManifestContent('chrome', 'global', 'toolkit/content/global/'))
        f.add(ManifestResource('chrome', 'gre-resources', 'toolkit/res/'))
        f.add(ManifestResource('chrome/pdfjs', 'pdfjs', './'))
        f.add(ManifestContent('chrome/pdfjs', 'pdfjs', 'pdfjs'))
        f.add(ManifestLocale('chrome', 'browser', 'en-US',
                             'en-US/locale/browser/'))

        f.copy(self.tmppath('chrome.manifest'))
        self.assertEqual(open(self.tmppath('chrome.manifest')).readlines(), [
            'content global toolkit/content/global/\n',
            'resource gre-resources toolkit/res/\n',
            'resource pdfjs pdfjs/\n',
            'content pdfjs pdfjs/pdfjs\n',
            'locale browser en-US en-US/locale/browser/\n',
        ])

        self.assertRaises(
            ValueError,
            f.remove,
            ManifestContent('', 'global', 'toolkit/content/global/')
        )
        self.assertRaises(
            ValueError,
            f.remove,
            ManifestOverride('chrome', 'chrome://global/locale/netError.dtd',
                             'chrome://browser/locale/netError.dtd')
        )

        f.remove(ManifestContent('chrome', 'global',
                                 'toolkit/content/global/'))
        self.assertRaises(
            ValueError,
            f.remove,
            ManifestContent('chrome', 'global', 'toolkit/content/global/')
        )

        f.copy(self.tmppath('chrome.manifest'))
        content = open(self.tmppath('chrome.manifest')).read()
        self.assertEqual(content[:42], f.open().read(42))
        self.assertEqual(content, f.open().read())

# Compiled typelib for the following IDL:
#     interface foo;
#     [uuid(5f70da76-519c-4858-b71e-e3c92333e2d6)]
#     interface bar {
#         void bar(in foo f);
#     };
bar_xpt = GeneratedFile(
    b'\x58\x50\x43\x4F\x4D\x0A\x54\x79\x70\x65\x4C\x69\x62\x0D\x0A\x1A' +
    b'\x01\x02\x00\x02\x00\x00\x00\x7B\x00\x00\x00\x24\x00\x00\x00\x5C' +
    b'\x80\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' +
    b'\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x5F' +
    b'\x70\xDA\x76\x51\x9C\x48\x58\xB7\x1E\xE3\xC9\x23\x33\xE2\xD6\x00' +
    b'\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\x0D\x00\x66\x6F\x6F\x00' +
    b'\x62\x61\x72\x00\x62\x61\x72\x00\x00\x00\x00\x01\x00\x00\x00\x00' +
    b'\x09\x01\x80\x92\x00\x01\x80\x06\x00\x00\x00'
)

# Compiled typelib for the following IDL:
#     [uuid(3271bebc-927e-4bef-935e-44e0aaf3c1e5)]
#     interface foo {
#         void foo();
#     };
foo_xpt = GeneratedFile(
    b'\x58\x50\x43\x4F\x4D\x0A\x54\x79\x70\x65\x4C\x69\x62\x0D\x0A\x1A' +
    b'\x01\x02\x00\x01\x00\x00\x00\x57\x00\x00\x00\x24\x00\x00\x00\x40' +
    b'\x80\x00\x00\x32\x71\xBE\xBC\x92\x7E\x4B\xEF\x93\x5E\x44\xE0\xAA' +
    b'\xF3\xC1\xE5\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x09\x00' +
    b'\x66\x6F\x6F\x00\x66\x6F\x6F\x00\x00\x00\x00\x01\x00\x00\x00\x00' +
    b'\x05\x00\x80\x06\x00\x00\x00'
)

# Compiled typelib for the following IDL:
#     [uuid(7057f2aa-fdc2-4559-abde-08d939f7e80d)]
#     interface foo {
#         void foo();
#     };
foo2_xpt = GeneratedFile(
    b'\x58\x50\x43\x4F\x4D\x0A\x54\x79\x70\x65\x4C\x69\x62\x0D\x0A\x1A' +
    b'\x01\x02\x00\x01\x00\x00\x00\x57\x00\x00\x00\x24\x00\x00\x00\x40' +
    b'\x80\x00\x00\x70\x57\xF2\xAA\xFD\xC2\x45\x59\xAB\xDE\x08\xD9\x39' +
    b'\xF7\xE8\x0D\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x09\x00' +
    b'\x66\x6F\x6F\x00\x66\x6F\x6F\x00\x00\x00\x00\x01\x00\x00\x00\x00' +
    b'\x05\x00\x80\x06\x00\x00\x00'
)


def read_interfaces(file):
    return dict((i.name, i) for i in Typelib.read(file).interfaces)


class TestXPTFile(TestWithTmpDir):
    def test_xpt_file(self):
        x = XPTFile()
        x.add(foo_xpt)
        x.add(bar_xpt)
        x.copy(self.tmppath('interfaces.xpt'))

        foo = read_interfaces(foo_xpt.open())
        foo2 = read_interfaces(foo2_xpt.open())
        bar = read_interfaces(bar_xpt.open())
        linked = read_interfaces(self.tmppath('interfaces.xpt'))
        self.assertEqual(foo['foo'], linked['foo'])
        self.assertEqual(bar['bar'], linked['bar'])

        x.remove(foo_xpt)
        x.copy(self.tmppath('interfaces2.xpt'))
        linked = read_interfaces(self.tmppath('interfaces2.xpt'))
        self.assertEqual(bar['foo'], linked['foo'])
        self.assertEqual(bar['bar'], linked['bar'])

        x.add(foo_xpt)
        x.copy(DestNoWrite(self.tmppath('interfaces.xpt')))
        linked = read_interfaces(self.tmppath('interfaces.xpt'))
        self.assertEqual(foo['foo'], linked['foo'])
        self.assertEqual(bar['bar'], linked['bar'])

        x = XPTFile()
        x.add(foo2_xpt)
        x.add(bar_xpt)
        x.copy(self.tmppath('interfaces.xpt'))
        linked = read_interfaces(self.tmppath('interfaces.xpt'))
        self.assertEqual(foo2['foo'], linked['foo'])
        self.assertEqual(bar['bar'], linked['bar'])

        x = XPTFile()
        x.add(foo_xpt)
        x.add(foo2_xpt)
        x.add(bar_xpt)
        from xpt import DataError
        self.assertRaises(DataError, x.copy, self.tmppath('interfaces.xpt'))


class TestMinifiedProperties(TestWithTmpDir):
    def test_minified_properties(self):
        propLines = [
            '# Comments are removed',
            'foo = bar',
            '',
            '# Another comment',
        ]
        prop = GeneratedFile('\n'.join(propLines))
        self.assertEqual(MinifiedProperties(prop).open().readlines(),
                         ['foo = bar\n', '\n'])
        open(self.tmppath('prop'), 'wb').write('\n'.join(propLines))
        MinifiedProperties(File(self.tmppath('prop'))) \
            .copy(self.tmppath('prop2'))
        self.assertEqual(open(self.tmppath('prop2')).readlines(),
                         ['foo = bar\n', '\n'])


class MatchTestTemplate(object):
    def prepare_match_test(self, with_dotfiles=False):
        self.add('bar')
        self.add('foo/bar')
        self.add('foo/baz')
        self.add('foo/qux/1')
        self.add('foo/qux/bar')
        self.add('foo/qux/2/test')
        self.add('foo/qux/2/test2')
        if with_dotfiles:
            self.add('foo/.foo')
            self.add('foo/.bar/foo')

    def do_match_test(self):
        self.do_check('', [
            'bar', 'foo/bar', 'foo/baz', 'foo/qux/1', 'foo/qux/bar',
            'foo/qux/2/test', 'foo/qux/2/test2'
        ])
        self.do_check('*', [
            'bar', 'foo/bar', 'foo/baz', 'foo/qux/1', 'foo/qux/bar',
            'foo/qux/2/test', 'foo/qux/2/test2'
        ])
        self.do_check('foo/qux', [
            'foo/qux/1', 'foo/qux/bar', 'foo/qux/2/test', 'foo/qux/2/test2'
        ])
        self.do_check('foo/b*', ['foo/bar', 'foo/baz'])
        self.do_check('baz', [])
        self.do_check('foo/foo', [])
        self.do_check('foo/*ar', ['foo/bar'])
        self.do_check('*ar', ['bar'])
        self.do_check('*/bar', ['foo/bar'])
        self.do_check('foo/*ux', [
            'foo/qux/1', 'foo/qux/bar', 'foo/qux/2/test', 'foo/qux/2/test2'
        ])
        self.do_check('foo/q*ux', [
            'foo/qux/1', 'foo/qux/bar', 'foo/qux/2/test', 'foo/qux/2/test2'
        ])
        self.do_check('foo/*/2/test*', ['foo/qux/2/test', 'foo/qux/2/test2'])
        self.do_check('**/bar', ['bar', 'foo/bar', 'foo/qux/bar'])
        self.do_check('foo/**/test', ['foo/qux/2/test'])
        self.do_check('foo/**', [
            'foo/bar', 'foo/baz', 'foo/qux/1', 'foo/qux/bar',
            'foo/qux/2/test', 'foo/qux/2/test2'
        ])
        self.do_check('**/2/test*', ['foo/qux/2/test', 'foo/qux/2/test2'])
        self.do_check('**/foo', [
            'foo/bar', 'foo/baz', 'foo/qux/1', 'foo/qux/bar',
            'foo/qux/2/test', 'foo/qux/2/test2'
        ])
        self.do_check('**/barbaz', [])
        self.do_check('f**/bar', ['foo/bar'])

    def do_finder_test(self, finder):
        self.assertTrue(finder.contains('foo/.foo'))
        self.assertTrue(finder.contains('foo/.bar'))
        self.assertTrue('foo/.foo' in [f for f, c in
                                       finder.find('foo/.foo')])
        self.assertTrue('foo/.bar/foo' in [f for f, c in
                                           finder.find('foo/.bar')])
        self.assertEqual(sorted([f for f, c in finder.find('foo/.*')]),
                         ['foo/.bar/foo', 'foo/.foo'])
        for pattern in ['foo', '**', '**/*', '**/foo', 'foo/*']:
            self.assertFalse('foo/.foo' in [f for f, c in
                                            finder.find(pattern)])
            self.assertFalse('foo/.bar/foo' in [f for f, c in
                                                finder.find(pattern)])
            self.assertEqual(sorted([f for f, c in finder.find(pattern)]),
                             sorted([f for f, c in finder
                                     if mozpack.path.match(f, pattern)]))


def do_check(test, finder, pattern, result):
    if result:
        test.assertTrue(finder.contains(pattern))
    else:
        test.assertFalse(finder.contains(pattern))
    test.assertEqual(sorted(list(f for f, c in finder.find(pattern))),
                     sorted(result))


class TestFileFinder(MatchTestTemplate, TestWithTmpDir):
    def add(self, path):
        ensure_parent_dir(self.tmppath(path))
        open(self.tmppath(path), 'wb').write(path)

    def do_check(self, pattern, result):
        do_check(self, self.finder, pattern, result)

    def test_file_finder(self):
        self.prepare_match_test(with_dotfiles=True)
        self.finder = FileFinder(self.tmpdir)
        self.do_match_test()
        self.do_finder_test(self.finder)


class TestJarFinder(MatchTestTemplate, TestWithTmpDir):
    def add(self, path):
        self.jar.add(path, path, compress=True)

    def do_check(self, pattern, result):
        do_check(self, self.finder, pattern, result)

    def test_jar_finder(self):
        self.jar = JarWriter(file=self.tmppath('test.jar'))
        self.prepare_match_test()
        self.jar.finish()
        reader = JarReader(file=self.tmppath('test.jar'))
        self.finder = JarFinder(self.tmppath('test.jar'), reader)
        self.do_match_test()


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