test_server.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696
  1. # test_server.py -- Tests for the git server
  2. # Copyright (C) 2010 Google, Inc.
  3. #
  4. # This program is free software; you can redistribute it and/or
  5. # modify it under the terms of the GNU General Public License
  6. # as published by the Free Software Foundation; version 2
  7. # or (at your option) any later version of the License.
  8. #
  9. # This program is distributed in the hope that it will be useful,
  10. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. # GNU General Public License for more details.
  13. #
  14. # You should have received a copy of the GNU General Public License
  15. # along with this program; if not, write to the Free Software
  16. # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  17. # MA 02110-1301, USA.
  18. """Tests for the smart protocol server."""
  19. from cStringIO import StringIO
  20. import os
  21. import tempfile
  22. from dulwich.errors import (
  23. GitProtocolError,
  24. NotGitRepository,
  25. UnexpectedCommandError,
  26. )
  27. from dulwich.repo import (
  28. MemoryRepo,
  29. Repo,
  30. )
  31. from dulwich.server import (
  32. Backend,
  33. DictBackend,
  34. FileSystemBackend,
  35. Handler,
  36. MultiAckGraphWalkerImpl,
  37. MultiAckDetailedGraphWalkerImpl,
  38. _split_proto_line,
  39. serve_command,
  40. ProtocolGraphWalker,
  41. ReceivePackHandler,
  42. SingleAckGraphWalkerImpl,
  43. UploadPackHandler,
  44. )
  45. from dulwich.tests import TestCase
  46. from dulwich.tests.utils import (
  47. make_commit,
  48. )
  49. ONE = '1' * 40
  50. TWO = '2' * 40
  51. THREE = '3' * 40
  52. FOUR = '4' * 40
  53. FIVE = '5' * 40
  54. SIX = '6' * 40
  55. class TestProto(object):
  56. def __init__(self):
  57. self._output = []
  58. self._received = {0: [], 1: [], 2: [], 3: []}
  59. def set_output(self, output_lines):
  60. self._output = ['%s\n' % line.rstrip() for line in output_lines]
  61. def read_pkt_line(self):
  62. if self._output:
  63. return self._output.pop(0)
  64. else:
  65. return None
  66. def write_sideband(self, band, data):
  67. self._received[band].append(data)
  68. def write_pkt_line(self, data):
  69. if data is None:
  70. data = 'None'
  71. self._received[0].append(data)
  72. def get_received_line(self, band=0):
  73. lines = self._received[band]
  74. if lines:
  75. return lines.pop(0)
  76. else:
  77. return None
  78. class TestGenericHandler(Handler):
  79. def __init__(self):
  80. Handler.__init__(self, Backend(), None)
  81. @classmethod
  82. def capabilities(cls):
  83. return ('cap1', 'cap2', 'cap3')
  84. @classmethod
  85. def required_capabilities(cls):
  86. return ('cap2',)
  87. class HandlerTestCase(TestCase):
  88. def setUp(self):
  89. super(HandlerTestCase, self).setUp()
  90. self._handler = TestGenericHandler()
  91. def assertSucceeds(self, func, *args, **kwargs):
  92. try:
  93. func(*args, **kwargs)
  94. except GitProtocolError, e:
  95. self.fail(e)
  96. def test_capability_line(self):
  97. self.assertEquals('cap1 cap2 cap3', self._handler.capability_line())
  98. def test_set_client_capabilities(self):
  99. set_caps = self._handler.set_client_capabilities
  100. self.assertSucceeds(set_caps, ['cap2'])
  101. self.assertSucceeds(set_caps, ['cap1', 'cap2'])
  102. # different order
  103. self.assertSucceeds(set_caps, ['cap3', 'cap1', 'cap2'])
  104. # error cases
  105. self.assertRaises(GitProtocolError, set_caps, ['capxxx', 'cap2'])
  106. self.assertRaises(GitProtocolError, set_caps, ['cap1', 'cap3'])
  107. # ignore innocuous but unknown capabilities
  108. self.assertRaises(GitProtocolError, set_caps, ['cap2', 'ignoreme'])
  109. self.assertFalse('ignoreme' in self._handler.capabilities())
  110. self._handler.innocuous_capabilities = lambda: ('ignoreme',)
  111. self.assertSucceeds(set_caps, ['cap2', 'ignoreme'])
  112. def test_has_capability(self):
  113. self.assertRaises(GitProtocolError, self._handler.has_capability, 'cap')
  114. caps = self._handler.capabilities()
  115. self._handler.set_client_capabilities(caps)
  116. for cap in caps:
  117. self.assertTrue(self._handler.has_capability(cap))
  118. self.assertFalse(self._handler.has_capability('capxxx'))
  119. class UploadPackHandlerTestCase(TestCase):
  120. def setUp(self):
  121. super(UploadPackHandlerTestCase, self).setUp()
  122. self._repo = MemoryRepo.init_bare([], {})
  123. backend = DictBackend({'/': self._repo})
  124. self._handler = UploadPackHandler(
  125. backend, ['/', 'host=lolcathost'], TestProto())
  126. def test_progress(self):
  127. caps = self._handler.required_capabilities()
  128. self._handler.set_client_capabilities(caps)
  129. self._handler.progress('first message')
  130. self._handler.progress('second message')
  131. self.assertEqual('first message',
  132. self._handler.proto.get_received_line(2))
  133. self.assertEqual('second message',
  134. self._handler.proto.get_received_line(2))
  135. self.assertEqual(None, self._handler.proto.get_received_line(2))
  136. def test_no_progress(self):
  137. caps = list(self._handler.required_capabilities()) + ['no-progress']
  138. self._handler.set_client_capabilities(caps)
  139. self._handler.progress('first message')
  140. self._handler.progress('second message')
  141. self.assertEqual(None, self._handler.proto.get_received_line(2))
  142. def test_get_tagged(self):
  143. refs = {
  144. 'refs/tags/tag1': ONE,
  145. 'refs/tags/tag2': TWO,
  146. 'refs/heads/master': FOUR, # not a tag, no peeled value
  147. }
  148. # repo needs to peel this object
  149. self._repo.object_store.add_object(make_commit(id=FOUR))
  150. self._repo.refs._update(refs)
  151. peeled = {
  152. 'refs/tags/tag1': '1234' * 10,
  153. 'refs/tags/tag2': '5678' * 10,
  154. }
  155. self._repo.refs._update_peeled(peeled)
  156. caps = list(self._handler.required_capabilities()) + ['include-tag']
  157. self._handler.set_client_capabilities(caps)
  158. self.assertEquals({'1234' * 10: ONE, '5678' * 10: TWO},
  159. self._handler.get_tagged(refs, repo=self._repo))
  160. # non-include-tag case
  161. caps = self._handler.required_capabilities()
  162. self._handler.set_client_capabilities(caps)
  163. self.assertEquals({}, self._handler.get_tagged(refs, repo=self._repo))
  164. class TestUploadPackHandler(UploadPackHandler):
  165. @classmethod
  166. def required_capabilities(self):
  167. return ()
  168. class ProtocolGraphWalkerTestCase(TestCase):
  169. def setUp(self):
  170. super(ProtocolGraphWalkerTestCase, self).setUp()
  171. # Create the following commit tree:
  172. # 3---5
  173. # /
  174. # 1---2---4
  175. commits = [
  176. make_commit(id=ONE, parents=[], commit_time=111),
  177. make_commit(id=TWO, parents=[ONE], commit_time=222),
  178. make_commit(id=THREE, parents=[ONE], commit_time=333),
  179. make_commit(id=FOUR, parents=[TWO], commit_time=444),
  180. make_commit(id=FIVE, parents=[THREE], commit_time=555),
  181. ]
  182. self._repo = MemoryRepo.init_bare(commits, {})
  183. backend = DictBackend({'/': self._repo})
  184. self._walker = ProtocolGraphWalker(
  185. TestUploadPackHandler(backend, ['/', 'host=lolcats'], TestProto()),
  186. self._repo.object_store, self._repo.get_peeled)
  187. def test_is_satisfied_no_haves(self):
  188. self.assertFalse(self._walker._is_satisfied([], ONE, 0))
  189. self.assertFalse(self._walker._is_satisfied([], TWO, 0))
  190. self.assertFalse(self._walker._is_satisfied([], THREE, 0))
  191. def test_is_satisfied_have_root(self):
  192. self.assertTrue(self._walker._is_satisfied([ONE], ONE, 0))
  193. self.assertTrue(self._walker._is_satisfied([ONE], TWO, 0))
  194. self.assertTrue(self._walker._is_satisfied([ONE], THREE, 0))
  195. def test_is_satisfied_have_branch(self):
  196. self.assertTrue(self._walker._is_satisfied([TWO], TWO, 0))
  197. # wrong branch
  198. self.assertFalse(self._walker._is_satisfied([TWO], THREE, 0))
  199. def test_all_wants_satisfied(self):
  200. self._walker.set_wants([FOUR, FIVE])
  201. # trivial case: wants == haves
  202. self.assertTrue(self._walker.all_wants_satisfied([FOUR, FIVE]))
  203. # cases that require walking the commit tree
  204. self.assertTrue(self._walker.all_wants_satisfied([ONE]))
  205. self.assertFalse(self._walker.all_wants_satisfied([TWO]))
  206. self.assertFalse(self._walker.all_wants_satisfied([THREE]))
  207. self.assertTrue(self._walker.all_wants_satisfied([TWO, THREE]))
  208. def test_split_proto_line(self):
  209. allowed = ('want', 'done', None)
  210. self.assertEquals(('want', ONE),
  211. _split_proto_line('want %s\n' % ONE, allowed))
  212. self.assertEquals(('want', TWO),
  213. _split_proto_line('want %s\n' % TWO, allowed))
  214. self.assertRaises(GitProtocolError, _split_proto_line,
  215. 'want xxxx\n', allowed)
  216. self.assertRaises(UnexpectedCommandError, _split_proto_line,
  217. 'have %s\n' % THREE, allowed)
  218. self.assertRaises(GitProtocolError, _split_proto_line,
  219. 'foo %s\n' % FOUR, allowed)
  220. self.assertRaises(GitProtocolError, _split_proto_line, 'bar', allowed)
  221. self.assertEquals(('done', None), _split_proto_line('done\n', allowed))
  222. self.assertEquals((None, None), _split_proto_line('', allowed))
  223. def test_determine_wants(self):
  224. self.assertEqual(None, self._walker.determine_wants({}))
  225. self.assertEqual('None', self._walker.proto.get_received_line())
  226. self._walker.proto.set_output([
  227. 'want %s multi_ack' % ONE,
  228. 'want %s' % TWO,
  229. ])
  230. heads = {
  231. 'refs/heads/ref1': ONE,
  232. 'refs/heads/ref2': TWO,
  233. 'refs/heads/ref3': THREE,
  234. }
  235. self._repo.refs._update(heads)
  236. self.assertEquals([ONE, TWO], self._walker.determine_wants(heads))
  237. self._walker.proto.set_output(['want %s multi_ack' % FOUR])
  238. self.assertRaises(GitProtocolError, self._walker.determine_wants, heads)
  239. self._walker.proto.set_output([])
  240. self.assertEquals([], self._walker.determine_wants(heads))
  241. self._walker.proto.set_output(['want %s multi_ack' % ONE, 'foo'])
  242. self.assertRaises(GitProtocolError, self._walker.determine_wants, heads)
  243. self._walker.proto.set_output(['want %s multi_ack' % FOUR])
  244. self.assertRaises(GitProtocolError, self._walker.determine_wants, heads)
  245. def test_determine_wants_advertisement(self):
  246. self._walker.proto.set_output([])
  247. # advertise branch tips plus tag
  248. heads = {
  249. 'refs/heads/ref4': FOUR,
  250. 'refs/heads/ref5': FIVE,
  251. 'refs/heads/tag6': SIX,
  252. }
  253. self._repo.refs._update(heads)
  254. self._repo.refs._update_peeled(heads)
  255. self._repo.refs._update_peeled({'refs/heads/tag6': FIVE})
  256. self._walker.determine_wants(heads)
  257. lines = []
  258. while True:
  259. line = self._walker.proto.get_received_line()
  260. if line == 'None':
  261. break
  262. # strip capabilities list if present
  263. if '\x00' in line:
  264. line = line[:line.index('\x00')]
  265. lines.append(line.rstrip())
  266. self.assertEquals([
  267. '%s refs/heads/ref4' % FOUR,
  268. '%s refs/heads/ref5' % FIVE,
  269. '%s refs/heads/tag6^{}' % FIVE,
  270. '%s refs/heads/tag6' % SIX,
  271. ], sorted(lines))
  272. # ensure peeled tag was advertised immediately following tag
  273. for i, line in enumerate(lines):
  274. if line.endswith(' refs/heads/tag6'):
  275. self.assertEquals('%s refs/heads/tag6^{}' % FIVE, lines[i+1])
  276. # TODO: test commit time cutoff
  277. class TestProtocolGraphWalker(object):
  278. def __init__(self):
  279. self.acks = []
  280. self.lines = []
  281. self.done = False
  282. self.http_req = None
  283. self.advertise_refs = False
  284. def read_proto_line(self, allowed):
  285. command, sha = self.lines.pop(0)
  286. if allowed is not None:
  287. assert command in allowed
  288. return command, sha
  289. def send_ack(self, sha, ack_type=''):
  290. self.acks.append((sha, ack_type))
  291. def send_nak(self):
  292. self.acks.append((None, 'nak'))
  293. def all_wants_satisfied(self, haves):
  294. return self.done
  295. def pop_ack(self):
  296. if not self.acks:
  297. return None
  298. return self.acks.pop(0)
  299. class AckGraphWalkerImplTestCase(TestCase):
  300. """Base setup and asserts for AckGraphWalker tests."""
  301. def setUp(self):
  302. super(AckGraphWalkerImplTestCase, self).setUp()
  303. self._walker = TestProtocolGraphWalker()
  304. self._walker.lines = [
  305. ('have', TWO),
  306. ('have', ONE),
  307. ('have', THREE),
  308. ('done', None),
  309. ]
  310. self._impl = self.impl_cls(self._walker)
  311. def assertNoAck(self):
  312. self.assertEquals(None, self._walker.pop_ack())
  313. def assertAcks(self, acks):
  314. for sha, ack_type in acks:
  315. self.assertEquals((sha, ack_type), self._walker.pop_ack())
  316. self.assertNoAck()
  317. def assertAck(self, sha, ack_type=''):
  318. self.assertAcks([(sha, ack_type)])
  319. def assertNak(self):
  320. self.assertAck(None, 'nak')
  321. def assertNextEquals(self, sha):
  322. self.assertEquals(sha, self._impl.next())
  323. class SingleAckGraphWalkerImplTestCase(AckGraphWalkerImplTestCase):
  324. impl_cls = SingleAckGraphWalkerImpl
  325. def test_single_ack(self):
  326. self.assertNextEquals(TWO)
  327. self.assertNoAck()
  328. self.assertNextEquals(ONE)
  329. self._walker.done = True
  330. self._impl.ack(ONE)
  331. self.assertAck(ONE)
  332. self.assertNextEquals(THREE)
  333. self._impl.ack(THREE)
  334. self.assertNoAck()
  335. self.assertNextEquals(None)
  336. self.assertNoAck()
  337. def test_single_ack_flush(self):
  338. # same as ack test but ends with a flush-pkt instead of done
  339. self._walker.lines[-1] = (None, None)
  340. self.assertNextEquals(TWO)
  341. self.assertNoAck()
  342. self.assertNextEquals(ONE)
  343. self._walker.done = True
  344. self._impl.ack(ONE)
  345. self.assertAck(ONE)
  346. self.assertNextEquals(THREE)
  347. self.assertNoAck()
  348. self.assertNextEquals(None)
  349. self.assertNoAck()
  350. def test_single_ack_nak(self):
  351. self.assertNextEquals(TWO)
  352. self.assertNoAck()
  353. self.assertNextEquals(ONE)
  354. self.assertNoAck()
  355. self.assertNextEquals(THREE)
  356. self.assertNoAck()
  357. self.assertNextEquals(None)
  358. self.assertNak()
  359. def test_single_ack_nak_flush(self):
  360. # same as nak test but ends with a flush-pkt instead of done
  361. self._walker.lines[-1] = (None, None)
  362. self.assertNextEquals(TWO)
  363. self.assertNoAck()
  364. self.assertNextEquals(ONE)
  365. self.assertNoAck()
  366. self.assertNextEquals(THREE)
  367. self.assertNoAck()
  368. self.assertNextEquals(None)
  369. self.assertNak()
  370. class MultiAckGraphWalkerImplTestCase(AckGraphWalkerImplTestCase):
  371. impl_cls = MultiAckGraphWalkerImpl
  372. def test_multi_ack(self):
  373. self.assertNextEquals(TWO)
  374. self.assertNoAck()
  375. self.assertNextEquals(ONE)
  376. self._walker.done = True
  377. self._impl.ack(ONE)
  378. self.assertAck(ONE, 'continue')
  379. self.assertNextEquals(THREE)
  380. self._impl.ack(THREE)
  381. self.assertAck(THREE, 'continue')
  382. self.assertNextEquals(None)
  383. self.assertAck(THREE)
  384. def test_multi_ack_partial(self):
  385. self.assertNextEquals(TWO)
  386. self.assertNoAck()
  387. self.assertNextEquals(ONE)
  388. self._impl.ack(ONE)
  389. self.assertAck(ONE, 'continue')
  390. self.assertNextEquals(THREE)
  391. self.assertNoAck()
  392. self.assertNextEquals(None)
  393. # done, re-send ack of last common
  394. self.assertAck(ONE)
  395. def test_multi_ack_flush(self):
  396. self._walker.lines = [
  397. ('have', TWO),
  398. (None, None),
  399. ('have', ONE),
  400. ('have', THREE),
  401. ('done', None),
  402. ]
  403. self.assertNextEquals(TWO)
  404. self.assertNoAck()
  405. self.assertNextEquals(ONE)
  406. self.assertNak() # nak the flush-pkt
  407. self._walker.done = True
  408. self._impl.ack(ONE)
  409. self.assertAck(ONE, 'continue')
  410. self.assertNextEquals(THREE)
  411. self._impl.ack(THREE)
  412. self.assertAck(THREE, 'continue')
  413. self.assertNextEquals(None)
  414. self.assertAck(THREE)
  415. def test_multi_ack_nak(self):
  416. self.assertNextEquals(TWO)
  417. self.assertNoAck()
  418. self.assertNextEquals(ONE)
  419. self.assertNoAck()
  420. self.assertNextEquals(THREE)
  421. self.assertNoAck()
  422. self.assertNextEquals(None)
  423. self.assertNak()
  424. class MultiAckDetailedGraphWalkerImplTestCase(AckGraphWalkerImplTestCase):
  425. impl_cls = MultiAckDetailedGraphWalkerImpl
  426. def test_multi_ack(self):
  427. self.assertNextEquals(TWO)
  428. self.assertNoAck()
  429. self.assertNextEquals(ONE)
  430. self._walker.done = True
  431. self._impl.ack(ONE)
  432. self.assertAcks([(ONE, 'common'), (ONE, 'ready')])
  433. self.assertNextEquals(THREE)
  434. self._impl.ack(THREE)
  435. self.assertAck(THREE, 'ready')
  436. self.assertNextEquals(None)
  437. self.assertAck(THREE)
  438. def test_multi_ack_partial(self):
  439. self.assertNextEquals(TWO)
  440. self.assertNoAck()
  441. self.assertNextEquals(ONE)
  442. self._impl.ack(ONE)
  443. self.assertAck(ONE, 'common')
  444. self.assertNextEquals(THREE)
  445. self.assertNoAck()
  446. self.assertNextEquals(None)
  447. # done, re-send ack of last common
  448. self.assertAck(ONE)
  449. def test_multi_ack_flush(self):
  450. # same as ack test but contains a flush-pkt in the middle
  451. self._walker.lines = [
  452. ('have', TWO),
  453. (None, None),
  454. ('have', ONE),
  455. ('have', THREE),
  456. ('done', None),
  457. ]
  458. self.assertNextEquals(TWO)
  459. self.assertNoAck()
  460. self.assertNextEquals(ONE)
  461. self.assertNak() # nak the flush-pkt
  462. self._walker.done = True
  463. self._impl.ack(ONE)
  464. self.assertAcks([(ONE, 'common'), (ONE, 'ready')])
  465. self.assertNextEquals(THREE)
  466. self._impl.ack(THREE)
  467. self.assertAck(THREE, 'ready')
  468. self.assertNextEquals(None)
  469. self.assertAck(THREE)
  470. def test_multi_ack_nak(self):
  471. self.assertNextEquals(TWO)
  472. self.assertNoAck()
  473. self.assertNextEquals(ONE)
  474. self.assertNoAck()
  475. self.assertNextEquals(THREE)
  476. self.assertNoAck()
  477. self.assertNextEquals(None)
  478. self.assertNak()
  479. def test_multi_ack_nak_flush(self):
  480. # same as nak test but contains a flush-pkt in the middle
  481. self._walker.lines = [
  482. ('have', TWO),
  483. (None, None),
  484. ('have', ONE),
  485. ('have', THREE),
  486. ('done', None),
  487. ]
  488. self.assertNextEquals(TWO)
  489. self.assertNoAck()
  490. self.assertNextEquals(ONE)
  491. self.assertNak()
  492. self.assertNextEquals(THREE)
  493. self.assertNoAck()
  494. self.assertNextEquals(None)
  495. self.assertNak()
  496. def test_multi_ack_stateless(self):
  497. # transmission ends with a flush-pkt
  498. self._walker.lines[-1] = (None, None)
  499. self._walker.http_req = True
  500. self.assertNextEquals(TWO)
  501. self.assertNoAck()
  502. self.assertNextEquals(ONE)
  503. self.assertNoAck()
  504. self.assertNextEquals(THREE)
  505. self.assertNoAck()
  506. self.assertNextEquals(None)
  507. self.assertNak()
  508. class FileSystemBackendTests(TestCase):
  509. """Tests for FileSystemBackend."""
  510. def setUp(self):
  511. super(FileSystemBackendTests, self).setUp()
  512. self.path = tempfile.mkdtemp()
  513. self.repo = Repo.init(self.path)
  514. self.backend = FileSystemBackend()
  515. def test_nonexistant(self):
  516. self.assertRaises(NotGitRepository,
  517. self.backend.open_repository, "/does/not/exist/unless/foo")
  518. def test_absolute(self):
  519. repo = self.backend.open_repository(self.path)
  520. self.assertEquals(repo.path, self.repo.path)
  521. def test_child(self):
  522. self.assertRaises(NotGitRepository,
  523. self.backend.open_repository, os.path.join(self.path, "foo"))
  524. class ServeCommandTests(TestCase):
  525. """Tests for serve_command."""
  526. def setUp(self):
  527. super(ServeCommandTests, self).setUp()
  528. self.backend = DictBackend({})
  529. def serve_command(self, handler_cls, args, inf, outf):
  530. return serve_command(handler_cls, ["test"] + args, backend=self.backend,
  531. inf=inf, outf=outf)
  532. def test_receive_pack(self):
  533. commit = make_commit(id=ONE, parents=[], commit_time=111)
  534. self.backend.repos["/"] = MemoryRepo.init_bare(
  535. [commit], {"refs/heads/master": commit.id})
  536. outf = StringIO()
  537. exitcode = self.serve_command(ReceivePackHandler, ["/"], StringIO("0000"), outf)
  538. outlines = outf.getvalue().splitlines()
  539. self.assertEquals(2, len(outlines))
  540. self.assertEquals("1111111111111111111111111111111111111111 refs/heads/master",
  541. outlines[0][4:].split("\x00")[0])
  542. self.assertEquals("0000", outlines[-1])
  543. self.assertEquals(0, exitcode)