from unittest.mock import ( Mock, patch, sentinel, ) import contextlib import datetime import os import pytest import re import shutil import subprocess import tempfile import pkg_resources import pygit2 import tenacity from gitubuntu.git_repository import HeadInfoItem from gitubuntu.patch_state import PatchState import gitubuntu.repo_builder as repo_builder from gitubuntu.repo_builder import Placeholder import gitubuntu.repo_comparator as repo_comparator import gitubuntu.source_builder as source_builder import gitubuntu.spec as spec import gitubuntu.version import gitubuntu.importer as target from gitubuntu.test_fixtures import pygit2_repo, repo @pytest.mark.parametrize('head_versions,expected', [ ( { 'prefix/ubuntu/trusty': ('a', '1.0-1'), }, { 'prefix/ubuntu/trusty-devel': 'a', 'prefix/ubuntu/devel': 'a', } ), ( { 'prefix/ubuntu/trusty': ('t', '1.0-1'), 'prefix/ubuntu/xenial': ('x', '1.0-2'), 'prefix/ubuntu/yakkety': ('y', '1.0-3'), }, { 'prefix/ubuntu/trusty-devel': 't', 'prefix/ubuntu/xenial-devel': 'x', 'prefix/ubuntu/yakkety-devel': 'y', 'prefix/ubuntu/devel': 'y', } ), ( { 'prefix/ubuntu/trusty': ('t', '1.0-1'), 'prefix/ubuntu/trusty-security': ('ts', '1.0-1.2'), 'prefix/ubuntu/trusty-proposed': ('tp', '1.0-1.1'), }, { 'prefix/ubuntu/trusty-devel': 'ts', 'prefix/ubuntu/devel': 'ts', } ), ]) def test_devel_branch_updates(head_versions, expected): result = target._devel_branch_updates( ref_prefix='prefix/', head_info={ head: HeadInfoItem( commit_id=commit_id, version=version, commit_time=None ) for head, (commit_id, version) in head_versions.items() }, series_name_list=['yakkety', 'xenial', 'trusty'], ) assert set(result) == set(expected.items()) def test_get_import_tag_msg(): assert target.get_import_tag_msg() == 'git-ubuntu import' @pytest.mark.parametrize( 'patch_state, expected', [ ( PatchState.UNAPPLIED, b'1-1 (patches unapplied)\n\nImported using git-ubuntu import.', ), ( PatchState.APPLIED, b'1-1 (patches applied)\n\nImported using git-ubuntu import.', ), ], ) def test_get_import_commit_msg( repo, patch_state, expected, ): """Test that get_import_commit_msg is generally correct This is the general parameterised test for the common case uses of target.get_import_commit_msg. This test uses a default source_builder.SourceSpec() to construct a test source tree to provide to get_import_commit_msg, so defaults from SourceSpec() will be used such as a package version string of '1-1'. :param GitUbuntuRepository repo: fixture providing a temporary GitUbuntuRepository instance to use :param PatchState patch_state: whether the commit is for an unapplied or applied import, as passed through to the function under test :param bytes expected: the expected return value from get_import_commit_msg """ publish_spec = source_builder.SourceSpec() publish_source = source_builder.Source(publish_spec) publish_oid = repo_builder.SourceTree(publish_source).write(repo.raw_repo) publish_tree_hash = str( repo.raw_repo.get(publish_oid).peel(pygit2.Tree).id ) assert target.get_import_commit_msg( repo, publish_spec.version, patch_state, ) == expected @pytest.mark.parametrize( 'input_repo, patch_state, expected', [ ( repo_builder.Repo(), PatchState.UNAPPLIED, [], ), ( repo_builder.Repo( tags={'importer/import/1-1': repo_builder.Commit()}, ), PatchState.UNAPPLIED, ['refs/tags/importer/import/1-1'], ), ( repo_builder.Repo( commits=[ repo_builder.Commit(name='import'), repo_builder.Commit(name='reimport1'), ], tags={ 'importer/import/1-1': repo_builder.Commit(), 'importer/reimport/import/1-1/0': repo_builder.Commit(), 'importer/reimport/import/1-1/1': repo_builder.Commit(), }, ), PatchState.UNAPPLIED, [ 'refs/tags/importer/reimport/import/1-1/0', 'refs/tags/importer/reimport/import/1-1/1', ], ), ( repo_builder.Repo(), PatchState.APPLIED, [], ), ( repo_builder.Repo( tags={'importer/applied/1-1': repo_builder.Commit()}, ), PatchState.APPLIED, ['refs/tags/importer/applied/1-1'], ), ( repo_builder.Repo( commits=[ repo_builder.Commit(name='applied'), repo_builder.Commit(name='reimport1'), ], tags={ 'importer/applied/1-1': repo_builder.Commit(), 'importer/reimport/applied/1-1/0': repo_builder.Commit(), 'importer/reimport/applied/1-1/1': repo_builder.Commit(), }, ), PatchState.APPLIED, [ 'refs/tags/importer/reimport/applied/1-1/0', 'refs/tags/importer/reimport/applied/1-1/1', ], ), ], ) def test_get_existing_import_tags(repo, patch_state, input_repo, expected): """Test that get_existing_import_tags is generally correct This is the general parameterised test for the common case uses of target.get_existing_import_tags. :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository :param PatchState patch_state: passed through to get_existing_import_tags :param repo_builder.Repo input_repo: input repository data :param list(str) expected: the names of the references that are expected to be returned, in order. """ input_repo.write(repo.raw_repo) assert [ ref.name for ref in target.get_existing_import_tags(repo, '1-1', 'importer', patch_state) ] == expected def test_get_existing_import_tags_ordering(repo): """Test that get_existing_import_tags returns results in the correct order To maintain hash stability, the spec defines that multiple changelog parents must appear in the order that they were published. For this to work, get_existing_import_tags must return the tags in the correct order even if the underlying git repository tags appear in an arbitrary order. :param GitUbuntuRepository repo: fixture of a temporary repository to use """ # Construct a synthetic git repository containing tags repo_builder.Repo( tags={ 'importer/import/1-1': repo_builder.Commit(), 'importer/reimport/import/1-1/0': repo_builder.Commit(), 'importer/reimport/import/1-1/1': repo_builder.Commit(), } ).write(repo.raw_repo) mock_get_all_reimport_tags_patch = patch.object( repo, 'get_all_reimport_tags', autospec=True, ) with mock_get_all_reimport_tags_patch as mock_get_all_reimport_tags: # Intentionally arrange for repo.get_all_reimport_tags to return the # expected result but in reverse order mock_get_all_reimport_tags.return_value = [ repo.raw_repo.lookup_reference('refs/tags/%s' % name) for name in [ 'importer/reimport/import/1-1/1', 'importer/reimport/import/1-1/0', ] ] # Call function under test tags = target.get_existing_import_tags(repo, '1-1', 'importer') # Check that our mock was called as expected mock_get_all_reimport_tags.assert_called_once_with( '1-1', 'importer', PatchState.UNAPPLIED, ) # Result tags should be in numeric order assert [ref.name for ref in tags] == [ 'refs/tags/importer/reimport/import/1-1/0', 'refs/tags/importer/reimport/import/1-1/1', ] @pytest.mark.parametrize( [ 'input_repo', 'validation_repo_delta', 'validation_repo_expected_identical_refs', ], [ ( repo_builder.Repo(), { 'add_commits': [ repo_builder.Commit(name='import'), ], 'update_tags': { 'importer/import/1-1': Placeholder('import'), }, }, [ 'refs/tags/importer/import/1-1', ], ), ( repo_builder.Repo( commits=[repo_builder.Commit(name='import')], tags={'importer/import/1-1': Placeholder('import')}, ), { 'add_commits': [ repo_builder.Commit(name='reimport'), ], 'update_tags': { 'importer/reimport/import/1-1/0': Placeholder('import'), 'importer/reimport/import/1-1/1': Placeholder('reimport'), }, }, [ 'refs/tags/importer/import/1-1', 'refs/tags/importer/reimport/import/1-1/0', 'refs/tags/importer/reimport/import/1-1/1', ], ), ( repo_builder.Repo( commits=[ repo_builder.Commit(name='import'), repo_builder.Commit(name='reimport1'), ], tags={ 'importer/import/1-1': Placeholder('import'), 'importer/reimport/import/1-1/0': Placeholder('import'), 'importer/reimport/import/1-1/1': Placeholder('reimport1'), }, ), { 'add_commits': [ repo_builder.Commit(name='reimport2'), ], 'update_tags': { 'importer/reimport/import/1-1/2': Placeholder('reimport2'), }, }, [ 'refs/tags/importer/import/1-1', 'refs/tags/importer/reimport/import/1-1/0', 'refs/tags/importer/reimport/import/1-1/1', 'refs/tags/importer/reimport/import/1-1/2', ], ), ], ) def test_create_import_tag( repo, input_repo, validation_repo_delta, validation_repo_expected_identical_refs, ): """ Unit test that create_import_tag creates the correct import tag :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository :param repo_builder.Repo input_repo: input repository data :param dict validation_repo_delta: how to transform the input repository into a "validation repository", expressed as a dict to provide as **kwargs to gitubuntu.repo_builder.Repo.copy() against the input repository. The validation repository is then used for the purposes of comparison against the output repository. :param list(str) validation_repo_expected_identical_refs: refs that must be identical between the validation repository and the output repository """ publish_commit = repo.raw_repo.get( repo_builder.Commit().write(repo.raw_repo) ).peel(pygit2.Commit) input_repo.write(repo.raw_repo) target.create_import_tag(repo, publish_commit, '1-1', 'importer') validation_repo = input_repo.copy(**validation_repo_delta) assert repo_comparator.equals( repoA=repo.raw_repo, repoB=validation_repo, test_refs=validation_repo_expected_identical_refs, ) def test_create_import_tag_hash_stability_on_first_import(repo): """Created import tags should be hash stable on first import :param GitUbuntuRepository repo: fixture providing a temporary GitUbuntuRepository instance to use """ publish_commit = repo.raw_repo.get( repo_builder.Commit( author=pygit2.Signature( 'Hash stability test author', 'newauthor@example.com', 1, 2, ), committer=pygit2.Signature( 'Hash stability test committer', 'newcommitter@example.com', 3, 4, ), ).write(repo.raw_repo) ).peel(pygit2.Commit) target.create_import_tag(repo, publish_commit, '1-1', 'importer') import_ref = repo.raw_repo.lookup_reference( 'refs/tags/importer/import/1-1', ) import_tag = import_ref.peel(pygit2.Tag) assert import_tag.tagger.name == spec.SYNTHESIZED_COMMITTER_NAME assert import_tag.tagger.email == spec.SYNTHESIZED_COMMITTER_EMAIL assert import_tag.tagger.time == 3 assert import_tag.tagger.offset == 4 def test_create_import_tag_hash_stability_on_reimport(repo): """Created import tags should be hash stable on reimport This includes both the /0 duplicate reimport tag of the original import tag as well as the /1 reimport tag being created. :param GitUbuntuRepository repo: fixture providing a temporary GitUbuntuRepository instance to use """ repo_builder.Repo( commits=[ repo_builder.Commit( name='root', author=pygit2.Signature( 'Hash stability test author', 'author@example.com', 1, 2, ), committer=pygit2.Signature( 'Hash stability test committer', 'committer@example.com', 3, 4, ), ), ], tags={ 'importer/import/1-1': Placeholder('root'), }, tagger=pygit2.Signature( 'Hash stability test name', 'stability@example.com', 5, 6, ), ).write(repo.raw_repo) publish_commit = repo.raw_repo.get( repo_builder.Commit( author=pygit2.Signature( 'Hash stability test new commit author', 'newauthor@example.com', 7, 8, ), committer=pygit2.Signature( 'Hash stability test new commit committer', 'newcommitter@example.com', 9, 10, ), ).write(repo.raw_repo) ).peel(pygit2.Commit) target.create_import_tag(repo, publish_commit, '1-1', 'importer') reimport_0_ref = repo.raw_repo.lookup_reference( 'refs/tags/importer/reimport/import/1-1/0', ) reimport_0_tag = reimport_0_ref.peel(pygit2.Tag) assert reimport_0_tag.tagger.name == 'Hash stability test name' assert reimport_0_tag.tagger.email == 'stability@example.com' assert reimport_0_tag.tagger.time == 5 assert reimport_0_tag.tagger.offset == 6 reimport_1_ref = repo.raw_repo.lookup_reference( 'refs/tags/importer/reimport/import/1-1/1', ) reimport_1_tag = reimport_1_ref.peel(pygit2.Tag) assert reimport_1_tag.tagger.name == spec.SYNTHESIZED_COMMITTER_NAME assert reimport_1_tag.tagger.email == spec.SYNTHESIZED_COMMITTER_EMAIL assert reimport_1_tag.tagger.time == 9 assert reimport_1_tag.tagger.offset == 10 @pytest.mark.parametrize( [ 'input_repo', 'validation_repo_delta', 'validation_repo_expected_identical_refs', ], [ ( repo_builder.Repo(), { 'add_commits': [repo_builder.Commit(name='import')], 'update_tags': {'importer/applied/1-1': Placeholder('import')}, }, [ 'refs/tags/importer/applied/1-1', ], ), ( repo_builder.Repo( commits=[repo_builder.Commit(name='import')], tags={'importer/applied/1-1': Placeholder('import')}, ), { 'add_commits': [repo_builder.Commit(name='reimport')], 'update_tags': { 'importer/reimport/applied/1-1/0': Placeholder('import'), 'importer/reimport/applied/1-1/1': Placeholder('reimport'), }, }, [ 'refs/tags/importer/applied/1-1', 'refs/tags/importer/reimport/applied/1-1/0', 'refs/tags/importer/reimport/applied/1-1/1', ], ), ( repo_builder.Repo( commits=[ repo_builder.Commit(name='import'), repo_builder.Commit(name='reimport1'), ], tags={ 'importer/applied/1-1': Placeholder('import'), 'importer/reimport/applied/1-1/0': Placeholder('import'), 'importer/reimport/applied/1-1/1': Placeholder('reimport1'), }, ), { 'add_commits': [ repo_builder.Commit(name='reimport2') ], 'update_tags': { 'importer/reimport/applied/1-1/2': Placeholder('reimport2'), }, }, [ 'refs/tags/importer/applied/1-1', 'refs/tags/importer/reimport/applied/1-1/0', 'refs/tags/importer/reimport/applied/1-1/1', 'refs/tags/importer/reimport/applied/1-1/2', ], ), ], ) def test_create_applied_tag( repo, input_repo, validation_repo_delta, validation_repo_expected_identical_refs, ): """ Unit test that create_applied_tag creates the correct import tag :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository :param repo_builder.Repo input_repo: input repository data :param dict validation_repo_delta: how to transform the input repository into a "validation repository", expressed as a dict to provide as **kwargs to gitubuntu.repo_builder.Repo.copy() against the input repository. The validation repository is then used for the purposes of comparison against the output repository. :param list(str) validation_repo_expected_identical_refs: refs that must be identical between the validation repository and the output repository """ publish_commit_str = str( repo.raw_repo.get( repo_builder.Commit().write(repo.raw_repo) ).peel(pygit2.Commit).id ) input_repo.write(repo.raw_repo) target.create_applied_tag(repo, publish_commit_str, '1-1', 'importer') validation_repo = input_repo.copy(**validation_repo_delta) assert repo_comparator.equals( repoA=repo.raw_repo, repoB=validation_repo, test_refs=validation_repo_expected_identical_refs, ) @pytest.mark.parametrize( [ 'input_repo', 'parent_overrides', 'changelog_versions', 'patch_state', 'expected_refs', ], [ ( repo_builder.Repo(), {}, ['1-2', '1-1',], PatchState.UNAPPLIED, [], ), ( repo_builder.Repo( commits=[repo_builder.Commit.from_spec(name='import')], tags={'importer/import/1-1': Placeholder('import')}, ), {}, ['1-2', '1-1'], PatchState.UNAPPLIED, ['refs/tags/importer/import/1-1'], ), ( repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='import'), repo_builder.Commit.from_spec(name='reimport', mutate=1), ], tags={ 'importer/import/1-1': Placeholder('import'), 'importer/reimport/import/1-1/0': Placeholder('import'), 'importer/reimport/import/1-1/1': Placeholder('reimport'), }, ), {}, ['1-2', '1-1'], PatchState.UNAPPLIED, [ 'refs/tags/importer/reimport/import/1-1/0', 'refs/tags/importer/reimport/import/1-1/1', ], ), ( repo_builder.Repo( commits=[repo_builder.Commit.from_spec(name='import')], tags={'importer/import/1-1': Placeholder('import')}, ), {}, ['1-3', '1-2', '1-1'], PatchState.UNAPPLIED, ['refs/tags/importer/import/1-1'], ), ( repo_builder.Repo(), {}, ['1-2', '1-1',], PatchState.APPLIED, [], ), ( repo_builder.Repo( commits=[repo_builder.Commit.from_spec(name='applied')], tags={'importer/applied/1-1': Placeholder('applied')}, ), {}, ['1-2', '1-1'], PatchState.APPLIED, ['refs/tags/importer/applied/1-1'], ), ( repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='import'), repo_builder.Commit.from_spec(name='reimport', mutate=1), ], tags={ 'importer/applied/1-1': Placeholder('import'), 'importer/reimport/applied/1-1/0': Placeholder('import'), 'importer/reimport/applied/1-1/1': Placeholder('reimport'), }, ), {}, ['1-2', '1-1'], PatchState.APPLIED, [ 'refs/tags/importer/reimport/applied/1-1/0', 'refs/tags/importer/reimport/applied/1-1/1', ], ), ( repo_builder.Repo( commits=[repo_builder.Commit.from_spec(name='applied')], tags={'importer/applied/1-1': Placeholder('applied')}, ), {}, ['1-3', '1-2', '1-1'], PatchState.APPLIED, ['refs/tags/importer/applied/1-1'], ), ], ) def test_get_changelog_parent_commits( repo, input_repo, parent_overrides, changelog_versions, patch_state, expected_refs, ): """Test that get_changelog_parent_commits is generally correct This is the general parameterised test for the common case uses of target.get_changelog_parent_commits. :param GitUbuntuRepository repo: fixture providing a temporary GitUbuntuRepository instance to use :param repo_builder.Repo input_repo: the input repository data to use that will be populated into @repo before @repo is passed through to get_changelog_parent_commits :param dict parent_overrides: passed through to get_changelog_parent_commits. :param PatchState patch_state: passed through to get_changelog_parent_commits :param list(str) expected_refs: the expected return value of get_changelog_parent_commits expressed using a list of reference names. Since get_changelog_parent_commits returns a list of commit hash strings, the reference names will need to be dereferenced before comparison; this way the test parameters don't need to be opaque hash strings. """ input_repo.write(repo.raw_repo) assert target.get_changelog_parent_commits( repo, 'importer', parent_overrides, changelog_versions, patch_state, ) == ([ str(repo.raw_repo.lookup_reference(expected_ref).peel(pygit2.Commit).id) for expected_ref in expected_refs ]) @patch('gitubuntu.git_repository.GitUbuntuRepository.add_base_remotes') def test_importer_main_cleanup_on_exception(add_base_remotes_mock): """Ensure that importer.main cleans up the temporary directory on errors If there are other files used/created by the importer outside of the repository, they are not tested for removal by this test. """ # leverage that the importer creates temporary directories local_tmpdir = tempfile.mkdtemp() try: assert not os.listdir(local_tmpdir) orig_mkdtemp = tempfile.mkdtemp def new_mkdtemp(suffix=None, prefix=None, dir=None): return orig_mkdtemp(suffix, prefix, dir=local_tmpdir) # patch here rather than as a decorator so we can use the original # function above with patch('tempfile.mkdtemp') as mkdtemp_mock: class MockError(Exception): pass mkdtemp_mock.side_effect = new_mkdtemp add_base_remotes_mock.side_effect = MockError() with pytest.raises(MockError): target.main(pkgname='placeholder', owner='placeholder') assert not os.listdir(local_tmpdir) finally: shutil.rmtree(local_tmpdir) @patch('gitubuntu.importer._main_with_repo') def test_importer_close_repository_on_exception(main_with_repo_mock): """Ensure that importer.main closes the repository in error cases Together with unit tests on the behaviour of GitUbuntuRepository.close(), this ensures that the repository directory is correctly cleaned up when required. This is effectively an open-box test as we rely on the following implementation details of importer.main: * That it calls _main_with_repo to do the actual work such that we can mock failure of its entire call with an exception. * That it will still function enough to correctly close the repository on _main_with_repo failure and when given no other arguments. If these implementation assumptions stop being true, this test will stop passing and will need reworking. """ class MockError(Exception): pass main_with_repo_mock.side_effect = MockError() repo = Mock() with pytest.raises(MockError): target.main(pkgname='placeholder', owner='placeholder', repo=repo) repo.close.assert_called() @pytest.mark.parametrize( [ 'input_repo', 'published_spec', 'expected_result', ], [ ( # Common case: upload tag has a changelog parent as an ancestor repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='import'), repo_builder.Commit.from_spec( name='upload', changelog_versions=['1-2', '1-1'], parents=[Placeholder('import')], ), ], tags={ 'importer/import/1-1': Placeholder('import'), 'importer/upload/1-2': Placeholder('upload'), }, ), {'changelog_versions': ['1-2', '1-1']}, True, ), ( # Upload tag is the first one, with no parents present repo_builder.Repo( commits=[ repo_builder.Commit.from_spec( name='upload', version='1-2', ), ], tags={ 'importer/upload/1-2': Placeholder('upload'), }, ), {'changelog_versions': ['1-2']}, True, ), ( # Upload tag mismatches published tree but is otherwise correct repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='import'), repo_builder.Commit.from_spec( name='upload', changelog_versions=['1-2', '1-1'], parents=[Placeholder('import')], mutate=True, ), ], tags={ 'importer/import/1-1': Placeholder('import'), 'importer/upload/1-2': Placeholder('upload'), }, ), {'changelog_versions': ['1-2', '1-1']}, target.RichHistoryTreeMismatch, ), ( # Upload tag doesn't have a changelog parent as an ancestor but is # otherwise correct repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='import'), repo_builder.Commit.from_spec( name='upload', changelog_versions=['1-2', '1-1'], ), ], tags={ 'importer/import/1-1': Placeholder('import'), 'importer/upload/1-2': Placeholder('upload'), }, ), {'changelog_versions': ['1-2', '1-1']}, target.RichHistoryHasNoChangelogParentAncestor, ), ], ) def test_validate_rich_history( repo, input_repo, published_spec, expected_result, ): """ General test for validate_rich_history(). This unit tests validate_rich_history() for various parameterized cases. Given an input repository and the specification of a Launchpad publication of a source package, we check that validate_rich_history() correctly accepts or rejects the rich history corresponding to the upload tag named 'importer/upload/1-2'. It is assumed that the package being imported is always of version '1-2' for all parameter sets. Since the target function requires rich history, the case of there not being rich history does not need to be tested here, as it wouldn't be called in this case. validate_rich_history() is generic for all sourced rich history, not just rich history sourced from an upload tag. But since it is independent of how the rich history commit arrived, it is easiest to use upload tags to test it; this results in coverage for all sources. :param GitUbuntuRepository repo: fixture providing a temporary GitUbuntuRepository instance to use :param repo_builder.Repo input_repo: input repository data :param dict published_spec: the package simulated being imported from the archive, specified as a dict to pass as **kwargs to repo_builder.Commit.from_spec() :param bool expected_result: the expected return value of, or exception raised by, the call to validate_rich_history() """ input_repo.write(repo.raw_repo) upload_tag = repo.raw_repo.lookup_reference( 'refs/tags/importer/upload/1-2', ) upload_tree = upload_tag.peel(pygit2.Tree) published_commit_builder = repo_builder.Commit.from_spec(**published_spec) published_commit_oid = published_commit_builder.write(repo.raw_repo) published_tree = repo.raw_repo.get(published_commit_oid).peel(pygit2.Tree) changelog_parents = target.get_unapplied_import_parents( repo=repo, version='1-2', namespace='importer', parent_overrides={}, unapplied_import_tree_hash=str(published_tree.id), ) try: result = target.validate_rich_history( repo=repo, rich_history_commit=upload_tag.peel(), changelog_parents=changelog_parents, import_tree=published_tree, ) except Exception as e: assert isinstance(e, expected_result) else: assert result == expected_result def test_add_changelog_note_to_commit(repo): """add_changelog_note_to_commit should add the expected note""" repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='1-1'), repo_builder.Commit.from_spec( name='1-2', changelog_versions=['1-1', '1-2'], parents=[Placeholder('1-1')], ), ], tags={'1-1': Placeholder('1-1'), '1-2': Placeholder('1-2')}, ).write(repo.raw_repo) parent_commit_ref = repo.raw_repo.lookup_reference('refs/tags/1-1') parent_commit = parent_commit_ref.peel(pygit2.Commit) child_commit_ref = repo.raw_repo.lookup_reference('refs/tags/1-2') child_commit = child_commit_ref.peel(pygit2.Commit) target.add_changelog_note_to_commit( repo=repo, namespace='importer', commit=child_commit, changelog_parents=[str(parent_commit.id)], ) note = repo.raw_repo.lookup_note( str(child_commit.id), 'refs/notes/importer/changelog', ) assert note.message == " * Test build from source_builder.\n" def test_add_changelog_note_to_commit_utf8(repo): """A changelog file with non-UTF8 should have such characters substituted :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository """ test_utf8_error_changelog_path = os.path.join( pkg_resources.resource_filename('gitubuntu', 'changelog_tests'), 'test_utf8_error', ) with open(test_utf8_error_changelog_path, 'rb') as f: utf8_changelog_blob = f.read() # We only need an example child commit with a debian/changelog file since # this is the only file accessed by add_changelog_note_to_commit(). # Further, the parent need only exist and can be empty. repo_builder.Repo( commits=[ repo_builder.Commit(name='parent'), repo_builder.Commit( tree=repo_builder.Tree({'debian': repo_builder.Tree( {'changelog': repo_builder.Blob(utf8_changelog_blob)} )}), name='child', ) ], tags={'parent': Placeholder('parent'), 'child': Placeholder('child')}, ).write(repo.raw_repo) parent_commit_ref = repo.raw_repo.lookup_reference('refs/tags/parent') parent_commit = parent_commit_ref.peel(pygit2.Commit) child_commit_ref = repo.raw_repo.lookup_reference('refs/tags/child') child_commit = child_commit_ref.peel(pygit2.Commit) target.add_changelog_note_to_commit( repo=repo, namespace='importer', commit=child_commit, changelog_parents=[str(parent_commit.id)], ) note = repo.raw_repo.lookup_note( str(child_commit.id), 'refs/notes/importer/changelog', ) assert "Nicol\N{REPLACEMENT CHARACTER}s" in note.message def test_double_changelog_note_add_does_not_fail(repo): """add_changelog_note_to_commit shouldn't fail if a note already exists""" repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='1-1'), repo_builder.Commit.from_spec( name='1-2', changelog_versions=['1-1', '1-2'], parents=[Placeholder('1-1')], ), ], tags={'1-1': Placeholder('1-1'), '1-2': Placeholder('1-2')}, ).write(repo.raw_repo) parent_commit_ref = repo.raw_repo.lookup_reference('refs/tags/1-1') parent_commit = parent_commit_ref.peel(pygit2.Commit) child_commit_ref = repo.raw_repo.lookup_reference('refs/tags/1-2') child_commit = child_commit_ref.peel(pygit2.Commit) # The first call should create the note (tested elsewhere) target.add_changelog_note_to_commit( repo=repo, namespace='importer', commit=child_commit, changelog_parents=[str(parent_commit.id)], ) # The second call should not fail even though the note already exists (as # created by the previous call) target.add_changelog_note_to_commit( repo=repo, namespace='importer', commit=child_commit, changelog_parents=[str(parent_commit.id)], ) def test_create_import_note(repo): """create_import_note() should create a note in the correct ref""" repo_builder.Repo( commits=[repo_builder.Commit(name='root')], tags={'root': repo_builder.Placeholder('root')}, ).write(repo.raw_repo) ref = repo.raw_repo.lookup_reference('refs/tags/root') commit = ref.peel(pygit2.Commit) target.create_import_note(repo, commit, 'importer') note = repo.raw_repo.lookup_note( str(commit.id), 'refs/notes/importer/importer', ) assert gitubuntu.version.VERSION in note.message def test_create_import_note_timestamp(repo): """create_import_note() should include the timestamp in the note""" repo_builder.Repo( commits=[repo_builder.Commit(name='root')], tags={'root': repo_builder.Placeholder('root')}, ).write(repo.raw_repo) ref = repo.raw_repo.lookup_reference('refs/tags/root') commit = ref.peel(pygit2.Commit) timestamp = datetime.datetime(2020, 3, 2, tzinfo=datetime.timezone.utc) target.create_import_note( repo=repo, commit=commit, namespace='importer', timestamp=timestamp, ) note = repo.raw_repo.lookup_note( str(commit.id), 'refs/notes/importer/importer', ) assert 'at 2020-03-02' in note.message def test_import_creates_import_note(repo): """When an import runs, the note should appear in the correct ref""" with source_builder.Source() as dsc_pathname: target.import_unapplied_dsc( repo=repo, version='1-1', namespace='importer', dist='ubuntu', dsc_pathname=dsc_pathname, head_name='ubuntu/focal', skip_orig=True, parent_overrides={}, ) ref = repo.raw_repo.lookup_reference('refs/tags/importer/import/1-1') commit = ref.peel(pygit2.Commit) note = repo.raw_repo.lookup_note( str(commit.id), 'refs/notes/importer/importer', ) assert gitubuntu.version.VERSION in note.message def patch_get_commit_environment(repo): """Patch a repository to use constant metadata :param GitUbuntuRepository repo: the repository whose behaviour will be patched. :rtype: contextmanager :returns: a context manager that, when entered, will cause the repository object to use constant metadata for created commits when metadata is looked up against a treeish. """ return patch.object( repo, 'get_commit_environment', return_value={ 'GIT_AUTHOR_NAME':'Test Builder', 'GIT_AUTHOR_EMAIL':'test@example.com', 'GIT_AUTHOR_DATE':'1970-01-01T00:00:00Z', 'GIT_COMMITTER_NAME':'Test Builder', 'GIT_COMMITTER_EMAIL':'test@example.com', 'GIT_COMMITTER_DATE':'1970-01-01T00:00:00Z', }, ) def MockSPI(dsc_path, version): """Construct a Mock SourcePackageInformation object In the following test cases we often need a Mock object with sufficient property information to enable an import. This function constructs such an object. :param str dsc_path: the path to a dsc file. This will be returned by the dsc_pathname attribute. :param str version: the package version string. This will be returned by the version attribute. :rtype: Mock :returns: a Mock object ready to use """ spi = Mock() spi.dsc_pathname = dsc_path spi.distribution_name = 'Ubuntu' spi.version = version spi.date_created = datetime.datetime( 1970, 1, 1, tzinfo=datetime.timezone.utc, ) head_name = Mock(name='head_name') head_name.return_value = 'importer/ubuntu/trusty' spi.head_name = head_name applied_head_name = Mock(name='applied_head_name') applied_head_name.return_value = 'importer/applied/ubuntu/trusty' spi.applied_head_name = applied_head_name spi.get_changes_file_url.return_value = None return spi @patch('gitubuntu.importer.get_import_tag_msg') @patch('gitubuntu.importer.get_import_commit_msg') def test_import_unapplied_spi_quilt_patches( get_import_commit_msg_mock, get_import_tag_msg_mock, repo, ): """Test that a package with quilt patches is imported with correct unapplied refs :param unittest.mock.Mock get_import_commit_msg_mock: mock of the function that determines the commit message to use for a given import :param unittest.mock.Mock get_import_tag_msg_mock: mock of the function that determines the tag message to use for a given import :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository """ # Match the repo_builder objects get_import_tag_msg_mock.return_value = 'Test tag' get_import_commit_msg_mock.return_value = b'Test commit' publish_spec = source_builder.SourceSpec(has_patches=True) input_repo = repo_builder.Repo() input_repo.write(repo.raw_repo) expected_result = repo_builder.Repo( commits=[ repo_builder.Commit( tree=repo_builder.SourceTree( source_builder.Source(publish_spec) ), name='publish' ), ], tags={'importer/import/1-1': Placeholder('publish')}, branches={'importer/ubuntu/trusty': Placeholder('publish')}, ) with source_builder.Source(publish_spec) as dsc_path: # import_unapplied_spi currently assumes it is called from the # repository directory (pristine-tar and other commands rely on # this) target.import_unapplied_spi( repo=repo, spi=MockSPI(dsc_path, publish_spec.version), namespace='importer', skip_orig=False, parent_overrides={}, ) assert repo_comparator.equals( repoA=repo.raw_repo, repoB=expected_result, test_refs=[ 'refs/heads/importer/ubuntu/trusty', 'refs/tags/importer/import/1-1', ], ) @pytest.mark.parametrize( [ 'input_repo', 'changelog_versions', 'validation_repo_delta', 'validation_repo_expected_identical_refs', ], [ pytest.param( repo_builder.Repo(), ['1-1'], { 'add_commits': [ repo_builder.Commit.from_spec(name='publish'), ], 'update_tags': {'importer/import/1-1': Placeholder('publish')}, }, [ 'refs/tags/importer/import/1-1', ] ), pytest.param( repo_builder.Repo( commits=[repo_builder.Commit.from_spec(name='import')], tags={'importer/import/1-1': Placeholder('import')}, ), ['1-2', '1-1'], { 'add_commits': [ repo_builder.Commit.from_spec( name='publish', parents=[Placeholder('import')], changelog_versions=['1-2', '1-1'], ), ], 'update_tags': {'importer/import/1-2': Placeholder('publish')}, }, [ 'refs/tags/importer/import/1-1', 'refs/tags/importer/import/1-2', ], ), pytest.param( repo_builder.Repo( commits=[repo_builder.Commit.from_spec(name='import')], tags={'importer/import/1-1': Placeholder('import')}, ), ['1-3', '1-2', '1-1'], { 'add_commits': [ repo_builder.Commit.from_spec( parents=[Placeholder('import')], name='publish', changelog_versions=['1-3', '1-2', '1-1'], ), ], 'update_tags': {'importer/import/1-3': Placeholder('publish')}, }, [ 'refs/tags/importer/import/1-1', 'refs/tags/importer/import/1-3', ], ), ( repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='import'), repo_builder.Commit.from_spec( name='reimport', mutate='Reimport tag contents', ), ], tags={ 'importer/import/1-1': Placeholder('import'), 'importer/reimport/import/1-1/0': Placeholder('import'), 'importer/reimport/import/1-1/1': Placeholder('reimport'), }, ), ['1-2', '1-1'], { 'add_commits': [ repo_builder.Commit.from_spec( parents=[ Placeholder('import'), Placeholder('reimport'), ], name='publish', changelog_versions=['1-2', '1-1'], ), ], 'update_tags': {'importer/import/1-2': Placeholder('publish')}, }, [ 'refs/tags/importer/import/1-1', 'refs/tags/importer/reimport/import/1-1/0', 'refs/tags/importer/reimport/import/1-1/1', 'refs/tags/importer/import/1-2', ], ), ] ) @patch('gitubuntu.importer.get_import_tag_msg') @patch('gitubuntu.importer.get_import_commit_msg') def test_import_unapplied_spi_parenting( get_import_commit_msg_mock, get_import_tag_msg_mock, repo, input_repo, changelog_versions, validation_repo_delta, validation_repo_expected_identical_refs, ): """Test that unapplied import commits have the correct parents :param unittest.mock.Mock get_import_commit_msg_mock: mock of the function that determines the commit message to use for a given import :param unittest.mock.Mock get_import_tag_msg_mock: mock of the function that determines the tag message to use for a given import :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository :param repo_builder.Repo input_repo: input repository data :param list(str) changelog_versions: the versions in the changelog of a fake package to test import :param dict validation_repo_delta: how to transform the input repository into a "validation repository", expressed as a dict to provide as **kwargs to gitubuntu.repo_builder.Repo.copy() against the input repository. The validation repository is then used for the purposes of comparison against the output repository. :param list(str) validation_repo_expected_identical_refs: refs that must be identical between the validation repository and the output repository Verify that if an import of a package is made into input_repo where the package being imported has the given changelog_versions, then the output repository has commits with the parents we expect. This is tested by comparing specific output references against the validation repository. """ # Match the repo_builder objects get_import_tag_msg_mock.return_value = 'Test tag' get_import_commit_msg_mock.return_value = b'Test commit' input_repo.write(repo.raw_repo) publish_spec = source_builder.SourceSpec( changelog_versions=changelog_versions, ) with source_builder.Source(publish_spec) as dsc_path: # import_unapplied_spi currently assumes it is called from the # repository directory (pristine-tar and other commands rely on # this) target.import_unapplied_spi( repo=repo, spi=MockSPI(dsc_path, publish_spec.version), namespace='importer', skip_orig=False, parent_overrides={}, ) # we would like to check the commit hashes, but we cannot # currently, as the commit messages are specific to the importer # and not codified validation_repo = input_repo.copy(**validation_repo_delta) assert repo_comparator.equals( repoA=repo.raw_repo, repoB=validation_repo, test_refs=validation_repo_expected_identical_refs, ) @patch('gitubuntu.importer.get_import_tag_msg') @patch('gitubuntu.importer.get_import_commit_msg') def test_import_unapplied_spi_parent_override( get_import_commit_msg_mock, get_import_tag_msg_mock, repo, ): """Test import_unapplied_spi() parent_override functionality Test that if parent_overrides is used in the import_unapplied_spi call then the resulting commit correctly uses the overridden parents specified. :param unittest.mock.Mock get_import_commit_msg_mock: mock of the function that determines the commit message to use for a given import :param unittest.mock.Mock get_import_tag_msg_mock: mock of the function that determines the tag message to use for a given import :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository :param repo_builder.Repo input_repo: input repository data """ # Match the repo_builder objects get_import_tag_msg_mock.return_value = 'Test tag' get_import_commit_msg_mock.return_value = b'Test commit' input_repo = repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='import1-1', version='1-1'), repo_builder.Commit.from_spec(name='import1-2', version='1-2'), ], tags={ 'importer/import/1-1': Placeholder('import1-1'), 'importer/import/1-2': Placeholder('import1-2'), }, ) input_repo.write(repo.raw_repo) publish_spec = source_builder.SourceSpec( changelog_versions=['2-1', '1-1'], ) with source_builder.Source(publish_spec) as dsc_path: # import_unapplied_spi currently assumes it is called from the # repository directory (pristine-tar and other commands rely on # this) target.import_unapplied_spi( repo=repo, spi=MockSPI(dsc_path, publish_spec.version), namespace='importer', skip_orig=False, parent_overrides={'2-1': {'changelog_parent': '1-2'}}, ) validation_repo = input_repo.copy( add_commits=[ repo_builder.Commit.from_spec( parents=[Placeholder('import1-2')], name='publish', changelog_versions=['2-1', '1-1'], ), ], update_tags={ 'importer/import/2-1': Placeholder('publish'), }, ) validation_repo_expected_identical_refs = [ 'refs/tags/importer/import/1-1', 'refs/tags/importer/import/1-2', 'refs/tags/importer/import/2-1', ] assert repo_comparator.equals( repoA=repo.raw_repo, repoB=validation_repo, test_refs=validation_repo_expected_identical_refs, ) def test_import_unapplied_spi_parent_override_failure(repo): """ Test override_parents ParentOverrideError raise When a parent override is specified but the specified version doesn't have an import tag, an exception should be raised. :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository """ repo_builder.Repo( commits=[repo_builder.Commit.from_spec(name='import1-1')], tags={'importer/import/1-1': Placeholder('import1-1')}, ).write(repo.raw_repo) with pytest.raises(target.ParentOverrideError): target.override_parents( parent_overrides={'2-1': {'changelog_parent': '1-2'}}, repo=repo, version='2-1', namespace='importer', ) @pytest.mark.parametrize( 'input_repo, expected_ancestor_commits, expected_parent_commits', [ # In general, these tests do not set applied commit parents in the # input repository since we have no mechanism to do that correctly, but # this doesn't matter for the purposes of these tests. # if only one import tag exists, then it is the parent pytest.param( repo_builder.Repo( commits=[ repo_builder.Commit.from_spec( name='unapplied1', has_patches=True, ), repo_builder.Commit.from_spec( parents=[Placeholder('unapplied1')], name='unapplied2', changelog_versions=['1-2', '1-1'], has_patches=True, ), repo_builder.Commit.from_spec( name='applied1', patches_applied=True, ), ], # no branches: technically not possible but branches are not # relevant to the test branches={}, tags={ 'importer/import/1-1': Placeholder('unapplied1'), 'importer/import/1-2': Placeholder('unapplied2'), 'importer/applied/1-1': Placeholder('applied1'), }, ), ['refs/tags/importer/import/1-2'], ['refs/tags/importer/applied/1-1'], ), # if multiple import tags exist, then do they all end up as parents? pytest.param( repo_builder.Repo( commits=[ repo_builder.Commit.from_spec( name='unapplied1', has_patches=True, ), repo_builder.Commit.from_spec( parents=[Placeholder('unapplied1')], name='unapplied2', changelog_versions=['1-2', '1-1'], has_patches=True, ), repo_builder.Commit.from_spec( parents=[Placeholder('unapplied1')], name='unapplied2reimport', changelog_versions=['1-2', '1-1'], has_patches=True, mutate='reimport tag', ), repo_builder.Commit.from_spec( name='applied1', patches_applied=True, ), ], # no branches: technically not possible but branches are not # relevant to the test branches={}, tags={ 'importer/import/1-1': Placeholder('unapplied1'), 'importer/import/1-2': Placeholder('unapplied2'), 'importer/reimport/import/1-2/0': Placeholder('unapplied2'), 'importer/reimport/import/1-2/1': Placeholder('unapplied2reimport'), 'importer/applied/1-1': Placeholder('applied1'), }, ), [ 'refs/tags/importer/reimport/import/1-2/0', 'refs/tags/importer/reimport/import/1-2/1', ], [ 'refs/tags/importer/applied/1-1', ], marks=pytest.mark.xfail(reason='LP: #1755247'), ), # do we correctly create a reimport tag because a different import # already exists? pytest.param( repo_builder.Repo( commits=[ repo_builder.Commit.from_spec( name='unapplied1', has_patches=True, ), repo_builder.Commit.from_spec( name='unapplied1_reimport', has_patches=True, mutate='reimport contents', ), repo_builder.Commit.from_spec( parents=[Placeholder('unapplied1')], name='unapplied2', changelog_versions=['1-2', '1-1'], has_patches=True, ), repo_builder.Commit.from_spec( name='applied1', patches_applied=True, ), repo_builder.Commit.from_spec( name='applied1_reimport', patches_applied=True, mutate='reimport contents', ), ], # no branches: technically not possible but branches are not # relevant to the test branches={}, tags={ 'importer/import/1-1': Placeholder('unapplied1'), 'importer/reimport/import/1-1/0': Placeholder('unapplied1'), 'importer/reimport/import/1-1/1': Placeholder('unapplied1_reimport'), 'importer/import/1-2': Placeholder('unapplied2'), 'importer/applied/1-1': Placeholder('applied1'), 'importer/reimport/applied/1-1/0': Placeholder('applied1'), 'importer/reimport/applied/1-1/1': Placeholder('applied1_reimport'), }, ), [ 'refs/tags/importer/reimport/import/1-2/0', 'refs/tags/importer/reimport/import/1-2/1', ], [ 'refs/tags/importer/reimport/applied/1-1/0', 'refs/tags/importer/reimport/applied/1-1/1', ], marks=pytest.mark.xfail(reason='LP: #1755247'), ), ], ) @patch('gitubuntu.importer.get_import_tag_msg') @patch('gitubuntu.importer.get_import_commit_msg') def test_import_applied_spi_parenting( get_import_commit_msg_mock, get_import_tag_msg_mock, repo, input_repo, expected_ancestor_commits, expected_parent_commits, ): """Test that applied import commits have the right parents :param unittest.mock.Mock get_import_commit_msg_mock: mock of the function that determines the commit message to use for a given import :param unittest.mock.Mock get_import_tag_msg_mock: mock of the function that determines the tag message to use for a given import :param repo gitubuntu.git_repository.GitUbuntuRepository: fixture to hold a temporary output repository :param repo_builder.Repo input_repo: input repository data :param list(str) expected_ancestor_commits: list of commit-ish strings that must be ancestors of the 'applied/1-2' tag following the applied import :param list(str) expected_parent_commits: list of commit-ish strings that must be parents of the 'applied/1-2' tag following the applied import. A fake package with version '1-2' that has a changelog parent of '1-1' is imported on top of the provided input_repo. The test fails if any of the expected_ancestor_commits or expected_parent_commits are not present. This test is ugly because we do not yet have a programmatic way to get the interstitial commits of the patch applications. """ # Match the repo_builder objects get_import_tag_msg_mock.return_value = 'Test tag' get_import_commit_msg_mock.return_value = b'Test commit' input_repo.write(repo.raw_repo) publish_spec = source_builder.SourceSpec( changelog_versions=['1-2', '1-1'], has_patches=True, ) with source_builder.Source(publish_spec) as dsc_path: target.import_applied_spi( repo=repo, spi=MockSPI(dsc_path, publish_spec.version), namespace='importer', allow_applied_failures=False, parent_overrides={}, ) applied_tag = repo.raw_repo.lookup_reference( 'refs/tags/importer/applied/1-2', ) applied_commit = applied_tag.peel(pygit2.Commit) applied_commit_parents = applied_commit.parents # convert refs to commits expected_ancestor_commit_hashes = [ str( repo.raw_repo.lookup_reference( expected_ancestor ).peel(pygit2.Commit).id ) for expected_ancestor in expected_ancestor_commits ] expected_parent_commit_hashes = [ str( repo.raw_repo.lookup_reference( expected_parent ).peel(pygit2.Commit).id ) for expected_parent in expected_parent_commits ] missing_ancestor_commit_hashes = [] for ancestor_commit_hash in expected_ancestor_commit_hashes: # We use merge_base as an ancestry test. If a merge base against an # expected ancestor isn't the expected ancestor itself, then it isn't # an ancestor. merge_base = repo.raw_repo.merge_base( repo.raw_repo.get(ancestor_commit_hash).id, applied_commit.id, ) if str( repo.raw_repo.get(merge_base).peel(pygit2.Commit).id ) != ancestor_commit_hash: missing_ancestor_commit_hashes.append(ancestor_commit_hash) missing_parent_commit_hashes = [] for parent_commit_hash in expected_parent_commit_hashes: for parent in applied_commit_parents: if str(parent.id) == parent_commit_hash: break else: missing_parent_commit_hashes.append(parent_commit_hash) assert ( not missing_ancestor_commit_hashes and not missing_parent_commit_hashes ) def test_parse_changelog_date_overrides_none(): """parse_changelog_date_overrides returns empty on path that is None """ result = frozenset(target.parse_changelog_date_overrides(None, None)) assert result == frozenset() def test_parse_changelog_date_overrides(tmpdir): """parse_changelog_date_overrides parses a standard file correctly :param py.path tmpdir: the pytest standard tmpdir fixture. """ test_file_path = tmpdir.join('test_changelog_date_override_file') test_file_path.write('''# a test comment # deliberately include some whitespace in the next line package1 1.1 package2 1.2 package1 1.3 ''') result = frozenset( target.parse_changelog_date_overrides(test_file_path, 'package1') ) assert result == frozenset({'1.1', '1.3'}) @pytest.mark.parametrize(['override', 'input_string', 'expected_result'], [ # Standard date that should parse and be used (False, 'Fri, 2 Feb 1971 12:34:56 +0100', (1971, 2, 2, 11, 34, 56, 60)), # Deliberately illegal date that cannot be parsed (True, 'Failday, 30 Feb 1971 99:99:99 +9999', (1972, 3, 3, 12, 45, 57, 0)), ]) def test_authorship_date(repo, override, input_string, expected_result): """Synthesized commit should use changelog or override when provided A synthesized commit should use the date of the changelog entry in the usual case, or commit_date when an override is requested. :param GitUbuntuRepository repo: fixture providing a temporary GitUbuntuRepository instance to use :param bool override: whether a changelog date override should be requested from import_unapplied_dsc() :param str input_string: the timestamp part of the changelog entry to use :param tuple expected_result: the expected author date of the synthesized commit, specified as six parameters to datetime.datetime() followed by the expected tz offset in minutes. """ spec = source_builder.SourceSpec(changelog_date=input_string) with source_builder.Source(spec) as dsc_pathname: target.import_unapplied_dsc( repo=repo, version='1-1', namespace='importer', dist='ubuntu', dsc_pathname=dsc_pathname, head_name='ubuntu/focal', skip_orig=True, parent_overrides={}, commit_date=datetime.datetime( 1972, 3, 3, 12, 45, 57, tzinfo=datetime.timezone.utc, ), changelog_date_overrides=( frozenset({'1-1'}) if override else frozenset() ), ) import_ref = repo.raw_repo.lookup_reference( 'refs/tags/importer/import/1-1', ) commit = import_ref.peel(pygit2.Commit) assert commit.author.time == int( datetime.datetime( *expected_result[:6], tzinfo=datetime.timezone.utc, ).timestamp() ) assert commit.author.offset == expected_result[6] @pytest.mark.parametrize(['test_input'], [ # Normal expected pattern ('https://git.launchpad.net/~first-last/ubuntu/+source/hello',), # With surrounding spaces (' https://git.launchpad.net/~first-last/ubuntu/+source/hello ',), # With query string ('https://git.launchpad.net/~first-last/ubuntu/+source/hello?a=b&c=d',), # External URLs are fine too (for the purposes of this function) ('https://github.com/foo/bar.git',), ]) def test_parse_rich_vcs_url_from_changes_good(test_input): """Valid URLs parse correctly""" result = target.parse_rich_vcs_url_from_changes({'Vcs-Git': test_input}) assert result == test_input.strip() def test_parse_rich_vcs_url_from_changes_with_branch(): """A VCS URL with a branch name specified parses correctly""" result = target.parse_rich_vcs_url_from_changes({ 'Vcs-Git': 'https://git.launchpad.net/~contributor/ubuntu/+source/hello' ' -b main-branch', }) assert result == ( 'https://git.launchpad.net/~contributor/ubuntu/+source/hello' ) @pytest.mark.parametrize(['test_input'], [ # HTTP, not HTTPS ('http://git.launchpad.net/~contributor/ubuntu/+source/hello',), # Attempted shell escapes using git-remote-ext(1) ('ext::sh',), ('ext::cat /etc/passwd',), # ssh type URLs aren't accepted (cannot authenticate) ('git@salsa.debian.org:sosreport-team/sosreport.git',), # For now, git: URLs aren't accepted (unencrypted with no validation) ('git://github.com/foo/bar.git',), # For now, %-escapes are not accepted, since they could be used to inject # arbitrary characters that we would need to validate are passed through # without interpretation ('http://git.launchpad.net/~contributor/ubuntu/+source/hell%6F',), # A '#' is probably an error such as in the following real-world example. # Maybe we could accept a '#' but we know of no such valid case and so it # isn't permitted for now, like most other special characters. ('https://salsa.debian.org/vdr-team/#PACKAGE#.git',), # Another real-world invalid case with special characters ('https://salsa.debian.org/ruby-team/<%=',), ]) def test_parse_rich_vcs_url_from_changes_bad(test_input): """Invalid URLs fail to validate""" with pytest.raises(target.RichHistoryInputValidationError): target.parse_rich_vcs_url_from_changes({'Vcs-Git': test_input}) @pytest.mark.parametrize(['test_input'], [ # Normal expected pattern ('refs/heads/feature-branch',), # With surrounding spaces (' refs/heads/feature-branch ',), # Some other valid refs ('refs/tags/v1.0',), # has a '.' ('refs/heads/merge-5.9.1+dfsg-4-kinetic',), # has a '+' ]) def test_parse_rich_vcs_ref_from_changes_good(test_input): """Valid refs parse correctly""" result = target.parse_rich_vcs_ref_from_changes( {'Vcs-Git-Ref': test_input}, ) assert result == test_input.strip() def test_parse_rich_vcs_ref_from_changes_multiple(): """Multiple refs parse as the first given ref only""" result = target.parse_rich_vcs_ref_from_changes( {'Vcs-Git-Ref': 'refs/heads/a refs/heads/b'}, ) assert result == 'refs/heads/a' @pytest.mark.parametrize(['test_input'], [ # Not starting with "refs/" ('foo',), # Attempt to be a forcing refspec ('+refs/heads/foo',), # Attempt to be a refspec specifying a local target ('refs/heads/foo:refs/heads/bar',), # Wildcard ('refs/tags/upload/*',), # git-check-ref-format(1) from Focal: rule 1 ('refs/tags/foo/.bar',), ('refs/tags/foo/bar.lock',), # git-check-ref-format(1) from Focal: rule 2 ('foo',), # git-check-ref-format(1) from Focal: rule 3 ('refs/tags/foo..bar',), # git-check-ref-format(1) from Focal: rule 4 ('refs/tags/foo\000',), ('refs/tags/foo\001',), ('refs/tags/foo\037',), ('refs/tags/foo\177',), # Trailing and leading whitespace are automatically trimmed, and we take # only the first whitespace-separated word and ignore the rest. So getting # a space into a parsed ref is impossible in our case and therefore we # cannot test it here. ('refs/tags/foo~',), ('refs/tags/foo^',), ('refs/tags/foo:',), # git-check-ref-format(1) from Focal: rule 5 ('refs/tags/foo?',), ('refs/tags/foo*',), ('refs/tags/foo[',), # git-check-ref-format(1) from Focal: rule 6 ('/refs/tags/foo',), ('refs/tags/foo//bar',), # git-check-ref-format(1) from Focal: rule 7 ('refs/tags/foo.',), # git-check-ref-format(1) from Focal: rule 8 ('refs/tags/${foo',), # git-check-ref-format(1) from Focal: rule 9 ('@',), # git-check-ref-format(1) from Focal: rule 10 ('refs/tags/foo\\bar',), ]) def test_parse_rich_vcs_ref_from_changes_bad(test_input): """Invalid refs fail to validate""" with pytest.raises(target.RichHistoryInputValidationError): target.parse_rich_vcs_ref_from_changes({'Vcs-Git-Ref': test_input}) @pytest.mark.parametrize(['test_input'], [ # Normal expected pattern ('0000000000000000000000000000000000000001',), # With surrounding spaces (' 0000000000000000000000000000000000000001 ',), # Hex ('abcd000000000000000000000000000000000001',), # Capitalized hex ('ABCD000000000000000000000000000000000001',), ]) def test_parse_rich_vcs_commit_from_changes_good(test_input): """Valid commit hashes parse correctly""" result = target.parse_rich_vcs_commit_from_changes( {'Vcs-Git-Commit': test_input}, ) assert result == test_input.strip() def test_parse_rich_vcs_commit_from_changes_multiple(): """Multiple commit hashes parse as the first commit hash only""" result = target.parse_rich_vcs_commit_from_changes({ 'Vcs-Git-Commit': '0000000000000000000000000000000000000001' ' 0000000000000000000000000000000000000002', }) assert result == '0000000000000000000000000000000000000001' @pytest.mark.parametrize(['test_input'], [ # Not hex ('ghi',), # Abbreviated ('abc1234',), ]) def test_parse_rich_vcs_commit_from_changes_bad(test_input): """Invalid commit hashes fail to validate""" with pytest.raises(target.RichHistoryInputValidationError): target.parse_rich_vcs_commit_from_changes( {'Vcs-Git-Commit': test_input}, ) @contextlib.contextmanager def temporary_changes_file_url(vcs_url, vcs_commit, vcs_ref, write=True): """Create a temporary "downloadable" changes file with rich history This is a context manager that provides a file:/// URL that a fake changes file can be "downloaded" from. The changes file can contain pointers to rich history. A temporary file is used, which is cleaned up when the context manager exits. :param str vcs_url: the git URL to be specified in the Vcs-Git field. :param str vcs_commit: the git commit to be specified in the Vcs-Git-Commit field. :param str vcs_ref: the git ref to be specified in the Vcs-Git-Ref field. :param bool write: whether to write the file at all, or leave it empty (useful to test the case where rich history is not specified). :rtype: str :returns: a file:/// URL that when fetched contains a fake changes file. """ with tempfile.NamedTemporaryFile(mode='w') as changes_file: if write: changes_file.write( "Vcs-Git: %s\nVcs-Git-Commit: %s\nVcs-Git-Ref: %s\n" % (vcs_url, vcs_commit, vcs_ref) ) changes_file.flush() yield 'file://' + changes_file.name def populate_rich_history(import_repo, uploader_repo): """Populate two git repositories with a simulation of rich history supply The two repositories are: 1. The "import repository": the git ubuntu importer's output 2. The "upload repository": where an uploader supplies rich history To test uploaders supplying rich history, we create a parent commit (version 1-1) in the import repository, a tree of what the uploader "uploaded" (version 1-2) in the import repository, and rich history supplied by the uploader in the upload repository. :rtype: tuple(pygit2.Commit, pygit2.Commit, pygit2.Tree) :returns: a tuple of three elements: 1. the rich commit supplied by the uploader 2. the parent commit (common to both repositories) 3. the tree object as uploaded by the uploader """ repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='1-1'), repo_builder.Commit.from_spec( name='template', mutate='1-2', ), ], tags={ 'importer/import/1-1': Placeholder('1-1'), 'template': Placeholder('template'), }, ).write(import_repo) repo_builder.Repo( commits=[ repo_builder.Commit.from_spec(name='1-1'), repo_builder.Commit.from_spec( parents=[Placeholder('1-1')], name='rich', mutate='1-2', message='Rich history commit', ), ], # no branches: technically not possible but branches are not # relevant to the test branches={}, tags={ 'rich': Placeholder('rich'), }, ).write(uploader_repo) return ( uploader_repo.lookup_reference('refs/tags/rich').peel(pygit2.Commit), import_repo.lookup_reference('refs/tags/importer/import/1-1').peel( pygit2.Commit, ), import_repo.lookup_reference('refs/tags/template').peel(pygit2.Tree), ) @patch('gitubuntu.importer.LAUNCHPAD_GIT_HOSTING_URL_PREFIX', 'file://') @patch('gitubuntu.importer.VCS_GIT_URL_VALIDATION', re.compile(r'.*')) def test_fetch_rich_history_from_changes_file(repo, pygit2_repo): """Rich history specified in a changes file is found and validated""" rich_commit, parent_commit, import_tree = populate_rich_history( import_repo=repo.raw_repo, uploader_repo=pygit2_repo, ) with temporary_changes_file_url( vcs_url='file://' + pygit2_repo.path, vcs_commit=str(rich_commit.id), vcs_ref='refs/tags/rich', ) as changes_file_url: validated_rich_commit = target.fetch_rich_history_from_changes_file( repo=repo, spi=Mock(get_changes_file_url=Mock(return_value=changes_file_url)), import_tree=import_tree, changelog_parents=[str(parent_commit.id)], ) assert validated_rich_commit.id == rich_commit.id def test_fetch_rich_history_from_changes_file_without_changes_file(): """If a changes file is not available, None is returned""" assert target.fetch_rich_history_from_changes_file( repo=None, spi=Mock(get_changes_file_url=Mock(return_value=None)), import_tree=None, changelog_parents=None, ) is None def test_fetch_rich_history_from_changes_file_without_required_fields(): """If rich history is not specified in its entirety, None is returned""" with temporary_changes_file_url( vcs_url=None, vcs_commit=None, vcs_ref=None, write=False, ) as changes_file_url: assert target.fetch_rich_history_from_changes_file( repo=None, spi=Mock(get_changes_file_url=Mock(return_value=changes_file_url)), import_tree=None, changelog_parents=None, ) is None def test_fetch_rich_history_from_changes_file_rejects_non_launchpad(): """If rich history is available but not from Launchpad, an exception is raised """ # Ugly formatting. Sorry. See https://bugs.python.org/issue12782 with \ pytest.raises(target.RichHistoryHasUnacceptableSource), \ temporary_changes_file_url( vcs_url='https://github.com/foo/bar.git', vcs_commit='0000000000000000000000000000000000000000', vcs_ref='refs/qux', ) as changes_file_url: target.fetch_rich_history_from_changes_file( repo=None, spi=Mock(get_changes_file_url=Mock(return_value=changes_file_url)), import_tree=None, changelog_parents=None, ) @patch('gitubuntu.importer.LAUNCHPAD_GIT_HOSTING_URL_PREFIX', 'file://') @patch('gitubuntu.importer.VCS_GIT_URL_VALIDATION', re.compile(r'.*')) def test_fetch_rich_history_from_changes_file_fetch_failure(repo, tmpdir): """If rich history cannot be fetched, an exception is raised""" # Ugly formatting. Sorry. See https://bugs.python.org/issue12782 with \ pytest.raises(target.RichHistoryFetchFailure), \ temporary_changes_file_url( vcs_url='file://' + str(tmpdir.join('nonexistent')), vcs_commit='0000000000000000000000000000000000000000', vcs_ref='refs/heads/nonexistent', ) as changes_file_url: target.fetch_rich_history_from_changes_file( repo=repo, spi=Mock(get_changes_file_url=Mock(return_value=changes_file_url)), import_tree=None, changelog_parents=None, retry_wait=tenacity.wait.wait_none(), # time warp for testing ) @patch('gitubuntu.importer.LAUNCHPAD_GIT_HOSTING_URL_PREFIX', 'file://') @patch('gitubuntu.importer.VCS_GIT_URL_VALIDATION', re.compile(r'.*')) def test_fetch_rich_history_from_changes_file_fetch_failure_retry( repo, pygit2_repo, ): """If rich history cannot be fetched once, it is retried""" rich_commit, parent_commit, import_tree = populate_rich_history( import_repo=repo.raw_repo, uploader_repo=pygit2_repo, ) real_git_run = repo.git_run with \ patch.object(repo, 'git_run') as mock_git_run, \ temporary_changes_file_url( vcs_url='file://' + str(pygit2_repo.path), vcs_commit=str(rich_commit.id), vcs_ref='refs/tags/rich', ) as changes_file_url: # Arrange for the repo.git_run() to fail the first time but succeed as # normal the second time. def mock_git_run_side_effect_2(*args, **kwargs): return real_git_run(*args, **kwargs) def mock_git_run_side_effect_1(*args, **kwargs): mock_git_run.side_effect = mock_git_run_side_effect_2 raise subprocess.CalledProcessError(returncode=1, cmd=None) mock_git_run.side_effect = mock_git_run_side_effect_1 target.fetch_rich_history_from_changes_file( repo=repo, spi=Mock(get_changes_file_url=Mock(return_value=changes_file_url)), import_tree=import_tree, changelog_parents=[str(parent_commit.id)], retry_wait=tenacity.wait.wait_none(), # time warp for testing ) @patch('gitubuntu.importer.LAUNCHPAD_GIT_HOSTING_URL_PREFIX', 'file://') @patch('gitubuntu.importer.VCS_GIT_URL_VALIDATION', re.compile(r'.*')) def test_fetch_rich_history_from_changes_file_missing_ref( repo, pygit2_repo, ): """If rich history is specified but the ref cannot be found, a hard exception is raised """ rich_commit, _, _ = populate_rich_history( import_repo=repo.raw_repo, uploader_repo=pygit2_repo, ) with \ pytest.raises(target.RichHistoryNotFoundError), \ temporary_changes_file_url( vcs_url='file://' + pygit2_repo.path, vcs_commit=str(rich_commit.id), vcs_ref='refs/tags/missing', ) as changes_file_url: target.fetch_rich_history_from_changes_file( repo=repo, spi=Mock(get_changes_file_url=Mock(return_value=changes_file_url)), import_tree=None, changelog_parents=None, ) @patch('gitubuntu.importer.LAUNCHPAD_GIT_HOSTING_URL_PREFIX', 'file://') @patch('gitubuntu.importer.VCS_GIT_URL_VALIDATION', re.compile(r'.*')) def test_fetch_rich_history_from_changes_file_missing_commit( repo, pygit2_repo, ): """If rich history is specified but the commit cannot be found, an exception is raised """ rich_commit, _, _ = populate_rich_history( import_repo=repo.raw_repo, uploader_repo=pygit2_repo, ) wrong_rich_commit_id = '0000000000000000000000000000000000000001' with \ pytest.raises(target.RichHistoryNotFoundError), \ temporary_changes_file_url( vcs_url='file://' + pygit2_repo.path, vcs_commit=wrong_rich_commit_id, vcs_ref='refs/tags/rich', ) as changes_file_url: target.fetch_rich_history_from_changes_file( repo=repo, spi=Mock(get_changes_file_url=Mock(return_value=changes_file_url)), import_tree=None, changelog_parents=None, ) @patch('gitubuntu.importer.LAUNCHPAD_GIT_HOSTING_URL_PREFIX', 'file://') @patch('gitubuntu.importer.VCS_GIT_URL_VALIDATION', re.compile(r'.*')) def test_fetch_rich_history_from_changes_file_not_a_commit(repo, pygit2_repo): """If rich history is specified but the commit hash resolves to something other than a commit, an exception is raised """ rich_commit, _, _ = populate_rich_history( import_repo=repo.raw_repo, uploader_repo=pygit2_repo, ) with \ pytest.raises(target.RichHistoryNotACommitError), \ temporary_changes_file_url( vcs_url='file://' + pygit2_repo.path, vcs_commit=str(rich_commit.peel(pygit2.Tree).id), vcs_ref='refs/tags/rich', ) as changes_file_url: target.fetch_rich_history_from_changes_file( repo=repo, spi=Mock(get_changes_file_url=Mock(return_value=changes_file_url)), import_tree=None, changelog_parents=None, ) def test_git_fetch_missing_ref_constant(tmpdir): """git fetch should return the stderr we expect when fetching a ref that doesn't exist, to ensure that our hard fail heuristic works properly in this case""" remote_dir = tmpdir.join('remote') local_dir = tmpdir.join('local') os.mkdir(remote_dir) os.mkdir(local_dir) subprocess.check_call(['git', 'init'], cwd=remote_dir) subprocess.check_call(['git', 'init'], cwd=local_dir) env = os.environ.copy() env['LC_ALL'] = 'C.UTF-8' with pytest.raises(subprocess.CalledProcessError) as excinfo: subprocess.run( ['git', 'fetch', '../remote', 'refs/heads/foo'], cwd=local_dir, check=True, capture_output=True, env=env, ) assert excinfo.value.stderr.startswith(target.GIT_REMOTE_REF_MISSING_PREFIX)