1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586 |
- import time
- from dulwich.object_store import MemoryObjectStore
- from dulwich.objects import Blob, Commit, Tree, parse_timezone
- from . import TestCase, skipIf
- try:
- import gevent
- gevent_support = True
- except ImportError:
- gevent_support = False
- if gevent_support:
- from dulwich.greenthreads import GreenThreadsMissingObjectFinder
- skipmsg = "Gevent library is not installed"
- def create_commit(marker=None):
- blob = Blob.from_string(b"The blob content " + marker)
- tree = Tree()
- tree.add(b"thefile " + marker, 0o100644, blob.id)
- cmt = Commit()
- cmt.tree = tree.id
- cmt.author = cmt.committer = b"John Doe <john@doe.net>"
- cmt.message = marker
- tz = parse_timezone(b"-0200")[0]
- cmt.commit_time = cmt.author_time = int(time.time())
- cmt.commit_timezone = cmt.author_timezone = tz
- return cmt, tree, blob
- def init_store(store, count=1):
- ret = []
- for i in range(count):
- objs = create_commit(marker=("%d" % i).encode("ascii"))
- for obj in objs:
- ret.append(obj)
- store.add_object(obj)
- return ret
- @skipIf(not gevent_support, skipmsg)
- class TestGreenThreadsMissingObjectFinder(TestCase):
- def setUp(self):
- super().setUp()
- self.store = MemoryObjectStore()
- self.cmt_amount = 10
- self.objs = init_store(self.store, self.cmt_amount)
- def test_finder(self):
- wants = [sha.id for sha in self.objs if isinstance(sha, Commit)]
- finder = GreenThreadsMissingObjectFinder(self.store, (), wants)
- self.assertEqual(len(finder.sha_done), 0)
- self.assertEqual(len(finder.objects_to_send), self.cmt_amount)
- finder = GreenThreadsMissingObjectFinder(
- self.store, wants[0 : int(self.cmt_amount / 2)], wants
- )
-
- self.assertEqual(len(finder.sha_done), (self.cmt_amount / 2) * 2)
- self.assertEqual(len(finder.objects_to_send), self.cmt_amount / 2)
|