def test_import_with_gitignore(self):
     # Make sure our git imports don't get confused by .gitignore files.
     new_content = {"fee/fi": "fo fum", ".gitignore": "fee/"}
     new_tree = self.cache.import_tree(shared.create_dir(new_content))
     export_dir = shared.create_dir()
     self.cache.export_tree(new_tree, export_dir)
     assert_contents(export_dir, new_content)

 def test_import_with_gitignore(self):
     # Make sure our git imports don't get confused by .gitignore files.
     new_content = {'fee/fi': 'fo fum', '.gitignore': 'fee/'}
     new_tree = yield from self.cache.import_tree(create_dir(new_content))
     export_dir = create_dir()
     yield from self.cache.export_tree(new_tree, export_dir)
     assert_contents(export_dir, new_content)

 def do_integration_test(self, args, expected, *, cwd=None,
                         **peru_cmd_kwargs):
     if not cwd:
         cwd = self.test_dir
     run_peru_command(args, cwd, **peru_cmd_kwargs)
     assert_contents(self.test_dir, expected,
                     excludes=[DEFAULT_PERU_FILE_NAME, '.peru'])

 def do_plugin_test(self, type, plugin_fields, expected_content, *,
                    fetch_dir=None):
     fetch_dir = fetch_dir or shared.create_dir()
     output = test_plugin_fetch(
         self.plugin_context, type, plugin_fields, fetch_dir)
     assert_contents(fetch_dir, expected_content)
     return output

    def test_reup_all(self):
        yaml_with_imports = dedent('''\
            imports:
                foo: ./
                bar: ./

            git module foo:
                url: {}
                rev: {}

            git module bar:
                url: {}
                reup: otherbranch
            ''').format(self.foo_dir, self.foo_master, self.bar_dir)
        test_dir = shared.create_dir({'peru.yaml': yaml_with_imports})
        expected = dedent('''\
            imports:
                foo: ./
                bar: ./

            git module foo:
                url: {}
                rev: {}

            git module bar:
                url: {}
                reup: otherbranch
                rev: {}
            ''').format(self.foo_dir, self.foo_master, self.bar_dir,
                        self.bar_otherbranch)
        run_peru_command(['reup'], test_dir)
        # This time we finally pull in barfile.
        assert_contents(test_dir,
                        {'peru.yaml': expected, 'a': 'b', 'barfile': 'new'},
                        excludes=['.peru'])

 def test_executable(self):
     exe = yield from rule.make_files_executable(self.cache, self.content_tree, ["b/*"])
     new_content_dir = shared.create_dir()
     yield from self.cache.export_tree(exe, new_content_dir)
     shared.assert_contents(new_content_dir, self.content)
     shared.assert_not_executable(os.path.join(new_content_dir, "a"))
     shared.assert_executable(os.path.join(new_content_dir, "b/c"))

 async def test_executable(self):
     exe = await rule.make_files_executable(self.cache, self.content_tree,
                                            ['b/*'])
     new_content_dir = shared.create_dir()
     await self.cache.export_tree(exe, new_content_dir)
     shared.assert_contents(new_content_dir, self.content)
     shared.assert_not_executable(os.path.join(new_content_dir, 'a'))
     shared.assert_executable(os.path.join(new_content_dir, 'b/c'))

 def do_plugin_test(self, type, plugin_fields, expected_content, *,
                    hide_stderr=False):
     fetch_dir = shared.create_dir()
     output = plugin_fetch(
         self.plugin_context, type, plugin_fields, fetch_dir,
         capture_output=True, stderr_to_stdout=hide_stderr)
     assert_contents(fetch_dir, expected_content)
     return output

    def test_merge_from_map(self):
        imports = {'foo': ('path1',), 'bar': ('path2',)}
        target_trees = {'foo': self.content_tree, 'bar': self.content_tree}

        merged_tree = merge_imports_tree(self.cache, imports, target_trees)

        merged_dir = create_dir()
        self.cache.export_tree(merged_tree, merged_dir)
        expected_content = {'path1/a': 'a', 'path2/a': 'a'}
        assert_contents(merged_dir, expected_content)

 def test_unpack_windows_zip(self):
     '''This zip was packed on Windows, so it doesn't include any file
     permissions. This checks that our executable-flag-restoring code
     doesn't barf when the flag isn't there.'''
     test_dir = shared.create_dir()
     archive = shared.test_resources / 'from_windows.zip'
     curl_plugin.extract_zip(str(archive), test_dir)
     shared.assert_contents(test_dir, {'windows_test/test.txt': 'Notepad!'})
     txt_file = join(test_dir, 'windows_test/test.txt')
     shared.assert_not_executable(txt_file)

 async def test_export_force_with_preexisting_files(self):
     # Create a working tree with a conflicting file.
     dirty_content = {'a': 'junk'}
     export_dir = create_dir(dirty_content)
     # Export should fail by default.
     with self.assertRaises(peru.cache.DirtyWorkingCopyError):
         await self.cache.export_tree(self.content_tree, export_dir)
     assert_contents(export_dir, dirty_content)
     # But it should suceed with the force flag.
     await self.cache.export_tree(self.content_tree, export_dir, force=True)
     assert_contents(export_dir, self.content)

 def test_merge_with_deep_prefix(self):
     '''This test was inspired by a bug on Windows where we would give git a
     backslash-separated merge prefix, even though git demands forward slash
     as a path separator.'''
     content = {'file': 'stuff'}
     content_dir = create_dir(content)
     tree = yield from self.cache.import_tree(content_dir)
     prefixed_tree = yield from self.cache.merge_trees(None, tree, 'a/b/')
     export_dir = create_dir()
     yield from self.cache.export_tree(prefixed_tree, export_dir)
     assert_contents(export_dir, {'a/b/file': 'stuff'})

 def test_import_with_files(self):
     all_content = {'foo': '',
                    'bar': '',
                    'baz/bing': ''}
     test_dir = create_dir(all_content)
     tree = self.cache.import_tree(test_dir, picks=['foo', 'baz'])
     expected_content = {'foo': '',
                         'baz/bing': ''}
     out_dir = create_dir()
     self.cache.export_tree(tree, out_dir)
     assert_contents(out_dir, expected_content)

 def test_missing_files_in_previous_tree(self):
     '''Export should allow missing files, and it should recreate them.'''
     export_dir = create_dir()
     # Nothing in content_tree exists yet, so this export should be the same
     # as if previous_tree wasn't specified.
     yield from self.cache.export_tree(
         self.content_tree, export_dir, previous_tree=self.content_tree)
     assert_contents(export_dir, self.content)
     # Make sure the same applies with just a single missing file.
     os.remove(os.path.join(export_dir, 'a'))
     yield from self.cache.export_tree(
         self.content_tree, export_dir, previous_tree=self.content_tree)
     assert_contents(export_dir, self.content)

    def test_single_reup(self):
        expected = dedent('''\
            git module foo:
                url: {}
                rev: {}

            git module bar:
                url: {}
                reup: otherbranch
            ''').format(self.foo_dir, self.foo_master, self.bar_dir)
        run_peru_command(['reup', 'foo'], self.test_dir)
        assert_contents(self.test_dir, {'peru.yaml': expected},
                        excludes=['.peru'])

 def test_export_force_with_changed_files(self):
     export_dir = create_dir()
     self.cache.export_tree(self.content_tree, export_dir)
     # If we dirty a file, a resync should fail.
     with open(os.path.join(export_dir, 'a'), 'w') as f:
         f.write('dirty')
     with self.assertRaises(peru.cache.DirtyWorkingCopyError):
         self.cache.export_tree(self.content_tree, export_dir,
                                previous_tree=self.content_tree)
     # But it should succeed with the --force flag.
     self.cache.export_tree(self.content_tree, export_dir, force=True,
                            previous_tree=self.content_tree)
     assert_contents(export_dir, self.content)

    def test_merge_trees(self):
        merged_tree = yield from self.cache.merge_trees(
            self.content_tree, self.content_tree, 'subdir')
        expected_content = dict(self.content)
        for path, content in self.content.items():
            expected_content[os.path.join('subdir', path)] = content
        export_dir = create_dir()
        yield from self.cache.export_tree(merged_tree, export_dir)
        assert_contents(export_dir, expected_content)

        with self.assertRaises(peru.cache.MergeConflictError):
            # subdir/ is already populated, so this merge should throw.
            yield from self.cache.merge_trees(
                merged_tree, self.content_tree, 'subdir')

    def test_merge_from_list(self):
        # This represents a list of key-value pairs in YAML, for example:
        #     imports:
        #         - foo: path1
        #         - foo: path2
        imports_list = [{'foo': 'path1'}, {'foo': 'path2'}]
        imports = build_imports(imports_list)
        target_trees = {'foo': self.content_tree}

        merged_tree = merge_imports_tree(self.cache, imports, target_trees)

        merged_dir = create_dir()
        self.cache.export_tree(merged_tree, merged_dir)
        expected_content = {'path1/a': 'a', 'path2/a': 'a'}
        assert_contents(merged_dir, expected_content)

    def test_merge_from_multimap(self):
        # This represents a list of key-value pairs in YAML, for example:
        #     imports:
        #         foo:
        #           - path1
        #           - path2
        imports = {'foo': ('path1', 'path2')}
        target_trees = {'foo': self.content_tree}

        merged_tree = merge_imports_tree(self.cache, imports, target_trees)

        merged_dir = create_dir()
        self.cache.export_tree(merged_tree, merged_dir)
        expected_content = {'path1/a': 'a', 'path2/a': 'a'}
        assert_contents(merged_dir, expected_content)

    def test_sync_from_subdir(self):
        peru_yaml = dedent('''\
            # Use a relative module path, to make sure it gets resolved
            # relative to the project root and not the dir where peru was
            # called.
            cp module relative_foo:
                path: {}

            imports:
                relative_foo: subdir
            '''.format(os.path.relpath(self.module_dir, start=self.test_dir)))
        shared.write_files(self.test_dir, {'peru.yaml': peru_yaml})
        subdir = os.path.join(self.test_dir, 'a', 'b')
        peru.compat.makedirs(subdir)
        run_peru_command(['sync'], subdir)
        self.assertTrue(os.path.isdir(os.path.join(self.test_dir, '.peru')),
                        msg=".peru dir didn't end up in the right place")
        assert_contents(os.path.join(self.test_dir, 'subdir'), {'foo': 'bar'})