test_fetch.py 文件源码

def test_fetch_repository(self):
        component_def = self.component_def.copy()
        component_def.update(self.update_def)

        fixture = fixtures.MockPatch('os.path.isdir')
        isdir_mock = self.useFixture(fixture).mock
        isdir_mock.return_value = self.dir_exists

        status = fetch.fetch_repository(component_def)
        git_path = os.path.join(self.tmp_path, component_def['name'])
        isdir_mock.assert_called_once_with(git_path)
        if self.dir_exists:
            self.assertEqual(status['status'],
                             fetch.ALREADY_EXISTED_STATUS)
            self.mock_clone.assert_not_called()

        elif self.clone_side_effect:
            self.mock_clone.assert_called_once_with('theurl', git_path)
            self.assertEqual(status['status'],
                             fetch.CLONE_FAILED_STATUS)
        else:
            git_ref = component_def.get('git_ref')
            self.mock_clone.assert_called_once_with('theurl', git_path)
            if git_ref and self.checkout_side_effect:
                self.repo.git.checkout.assert_called_once_with(git_ref)
                self.assertEqual(status['status'],
                                 fetch.CHECKOUT_FAILED_STATUS)
            elif git_ref:
                self.repo.git.checkout.assert_called_once_with(git_ref)
                self.assertEqual(status['status'],
                                 fetch.FETCH_SUCCEEDED_STATUS)
            else:
                self.assertEqual(status['status'],
                                 fetch.FETCH_SUCCEEDED_STATUS)