| import os |
| import shutil |
| import unittest |
| import pkg_resources |
| from ConfigParser import ConfigParser |
| |
| import mock |
| from ming.base import Object |
| from ming.orm import ThreadLocalORMSession |
| |
| from alluratest.controller import setup_basic_test, setup_global_objects |
| from allura.lib import helpers as h |
| from allura.tests.model.test_repo import RepoImplTestBase |
| from allura import model as M |
| |
| from forgehg import model as HM |
| from forgehg.tests import with_hg |
| |
| class TestNewRepo(unittest.TestCase): |
| |
| def setUp(self): |
| setup_basic_test() |
| self.setup_with_tools() |
| |
| @with_hg |
| def setup_with_tools(self): |
| setup_global_objects() |
| h.set_context('test', 'src-hg', neighborhood='Projects') |
| repo_dir = pkg_resources.resource_filename( |
| 'forgehg', 'tests/data') |
| self.repo = HM.Repository( |
| name='testrepo.hg', |
| fs_path=repo_dir, |
| url_path = '/test/', |
| tool = 'hg', |
| status = 'creating') |
| self.repo.refresh() |
| self.rev = M.repo.Commit.query.get(_id=self.repo.heads[0]['object_id']) |
| self.rev.repo = self.repo |
| ThreadLocalORMSession.flush_all() |
| ThreadLocalORMSession.close_all() |
| |
| def test_redo_trees(self): |
| old_tree = self.rev.tree |
| del self.rev.tree |
| M.repo.Tree.query.remove() |
| ThreadLocalORMSession.close_all() |
| new_tree = self.rev.tree |
| self.assertEqual(old_tree.tree_ids, new_tree.tree_ids) |
| self.assertEqual(old_tree.blob_ids, new_tree.blob_ids) |
| self.assertEqual(old_tree._id, new_tree._id) |
| |
| def test_commit(self): |
| assert self.rev.primary() is self.rev |
| assert self.rev.index_id().startswith('allura/model/repo/Commit#') |
| self.rev.author_url |
| self.rev.committer_url |
| assert self.rev.tree._id == self.rev.tree_id |
| assert self.rev.summary == self.rev.message.splitlines()[0] |
| assert self.rev.shorthand_id() == '[5a0a99]' |
| assert self.rev.symbolic_ids == (['default'], ['tip']) |
| assert self.rev.url() == ( |
| '/p/test/src-hg/ci/' |
| '5a0a993efa9bce7d1983344261393e841fcfd65d/') |
| all_cis = self.repo.log(self.rev._id, 0, 1000) |
| assert len(all_cis) == 6 |
| assert self.repo.log(self.rev._id, 1,1000) == all_cis[1:] |
| assert self.repo.log(self.rev._id, 0,3) == all_cis[:3] |
| assert self.repo.log(self.rev._id, 1,2) == all_cis[1:3] |
| for ci in all_cis: |
| ci.context() |
| self.rev.tree.ls() |
| assert self.rev.tree.readme() == ( |
| 'README', 'This is readme\nAnother line\n') |
| assert self.rev.tree.path() == '/' |
| assert self.rev.tree.url() == ( |
| '/p/test/src-hg/ci/' |
| '5a0a993efa9bce7d1983344261393e841fcfd65d/' |
| 'tree/') |
| self.rev.tree.by_name['README'] |
| assert self.rev.tree.is_blob('README') == True |
| |
| class TestHgRepo(unittest.TestCase, RepoImplTestBase): |
| |
| def setUp(self): |
| setup_basic_test() |
| self.setup_with_tools() |
| |
| @with_hg |
| def setup_with_tools(self): |
| setup_global_objects() |
| h.set_context('test', 'src-hg', neighborhood='Projects') |
| repo_dir = pkg_resources.resource_filename( |
| 'forgehg', 'tests/data') |
| self.repo = HM.Repository( |
| name='testrepo.hg', |
| fs_path=repo_dir, |
| url_path = '/test/', |
| tool = 'hg', |
| status = 'creating') |
| self.repo.refresh() |
| ThreadLocalORMSession.flush_all() |
| ThreadLocalORMSession.close_all() |
| |
| def test_init(self): |
| repo = HM.Repository( |
| name='testrepo.hg', |
| fs_path='/tmp/', |
| url_path = '/test/', |
| tool = 'hg', |
| status = 'creating') |
| dirname = os.path.join(repo.fs_path, repo.name) |
| if os.path.exists(dirname): |
| shutil.rmtree(dirname) |
| repo.init() |
| shutil.rmtree(dirname) |
| |
| def test_fork(self): |
| repo = HM.Repository( |
| name='testrepo.hg', |
| fs_path='/tmp/', |
| url_path = '/test/', |
| tool = 'hg', |
| status = 'creating') |
| repo_path = pkg_resources.resource_filename( |
| 'forgehg', 'tests/data/testrepo.hg') |
| dirname = os.path.join(repo.fs_path, repo.name) |
| if os.path.exists(dirname): |
| shutil.rmtree(dirname) |
| repo.init() |
| repo._impl.clone_from(repo_path) |
| assert len(repo.log()) |
| assert not os.path.exists('/tmp/testrepo.hg/.hg/external-changegroup') |
| assert not os.path.exists('/tmp/testrepo.hg/.hg/nested/nested-file') |
| assert os.path.exists('/tmp/testrepo.hg/.hg/hgrc') |
| cp = ConfigParser() |
| cp.read('/tmp/testrepo.hg/.hg/hgrc') |
| assert not cp.has_section('other') |
| assert cp.has_section('hooks') |
| assert not cp.has_option('hooks', 'changegroup.external') |
| assert not cp.has_option('hooks', 'commit') |
| self.assertEquals(cp.get('hooks', 'changegroup.sourceforge'), 'curl -s http://localhost//auth/refresh_repo/p/test/src-hg/') |
| assert not os.path.exists('/tmp/testrepo.hg/.hg/undo.branch') |
| shutil.rmtree(dirname) |
| |
| @mock.patch('forgehg.model.hg.g.post_event') |
| def test_clone(self, post_event): |
| repo = HM.Repository( |
| name='testrepo.hg', |
| fs_path='/tmp/', |
| url_path = '/test/', |
| tool = 'hg', |
| status = 'creating') |
| repo_path = pkg_resources.resource_filename( |
| 'forgehg', 'tests/data/testrepo.hg') |
| dirname = os.path.join(repo.fs_path, repo.name) |
| if os.path.exists(dirname): |
| shutil.rmtree(dirname) |
| repo.init() |
| repo._impl.clone_from(repo_path) |
| assert len(repo.log()) |
| assert not os.path.exists('/tmp/testrepo.hg/.hg/external-changegroup') |
| assert not os.path.exists('/tmp/testrepo.hg/.hg/nested/nested-file') |
| assert os.path.exists('/tmp/testrepo.hg/.hg/hgrc') |
| cp = ConfigParser() |
| cp.read('/tmp/testrepo.hg/.hg/hgrc') |
| assert not cp.has_section('other') |
| assert cp.has_section('hooks') |
| self.assertEquals(cp.get('hooks', 'changegroup.sourceforge'), 'curl -s http://localhost//auth/refresh_repo/p/test/src-hg/') |
| assert not cp.has_option('hook', 'changegroup.external') |
| assert not cp.has_option('hook', 'commit') |
| assert not os.path.exists('/tmp/testrepo.hg/.hg/undo.branch') |
| shutil.rmtree(dirname) |
| |
| def test_index(self): |
| i = self.repo.index() |
| assert i['type_s'] == 'Hg Repository', i |
| |
| def test_log(self): |
| for entry in self.repo.log(): |
| if entry._id.startswith('00000000'): continue |
| assert entry.committed.email == 'rick446@usa.net' |
| assert entry.message |
| |
| def test_revision(self): |
| entry = self.repo.commit('4a7f7ec0dcf5') |
| assert entry.committed.email == 'rick446@usa.net' |
| assert entry.message |
| assert str(entry.committed.date) == "2012-08-29 13:34:26", str(entry.committed.date) |
| # Test that sha1s for named refs are looked up in cache first, instead |
| # of from disk. |
| with mock.patch('forgehg.model.hg.M.repo.Commit.query') as q: |
| self.repo.heads.append(Object(name='HEAD', object_id='deadbeef')) |
| self.repo.commit('HEAD') |
| q.get.assert_called_with(_id='deadbeef') |
| |
| class TestHgCommit(unittest.TestCase): |
| |
| def setUp(self): |
| setup_basic_test() |
| self.setup_with_tools() |
| |
| @with_hg |
| def setup_with_tools(self): |
| setup_global_objects() |
| h.set_context('test', 'src-hg', neighborhood='Projects') |
| repo_dir = pkg_resources.resource_filename( |
| 'forgehg', 'tests/data') |
| self.repo = HM.Repository( |
| name='testrepo.hg', |
| fs_path=repo_dir, |
| url_path = '/test/', |
| tool = 'hg', |
| status = 'creating') |
| self.repo.refresh() |
| self.rev = self.repo.commit('tip') |
| ThreadLocalORMSession.flush_all() |
| ThreadLocalORMSession.close_all() |
| |
| def test_redo_trees(self): |
| old_tree = self.rev.tree |
| del self.rev.tree |
| M.repo.Tree.query.remove(dict(type='tree')) |
| ThreadLocalORMSession.close_all() |
| new_tree = self.rev.tree |
| self.assertEqual(old_tree.tree_ids, new_tree.tree_ids) |
| self.assertEqual(old_tree.blob_ids, new_tree.blob_ids) |
| self.assertEqual(old_tree._id, new_tree._id) |
| |
| def test_url(self): |
| assert self.rev.url().endswith('cfd65d/'), \ |
| self.rev.url() |
| |
| def test_committer_url(self): |
| assert self.rev.committer_url is None |
| |
| def test_primary(self): |
| assert self.rev.primary() == self.rev |
| |
| def test_shorthand(self): |
| assert len(self.rev.shorthand_id()) == 8 |
| |
| def test_diff(self): |
| diffs = (self.rev.diffs.added |
| +self.rev.diffs.removed |
| +self.rev.diffs.changed |
| +self.rev.diffs.copied) |
| for d in diffs: |
| print d |
