test_client.py 50 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489
  1. # test_client.py -- Tests for the git protocol, client side
  2. # Copyright (C) 2009 Jelmer Vernooij <jelmer@jelmer.uk>
  3. #
  4. # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
  5. # General Public License as public by the Free Software Foundation; version 2.0
  6. # or (at your option) any later version. You can redistribute it and/or
  7. # modify it under the terms of either of these two licenses.
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. #
  15. # You should have received a copy of the licenses; if not, see
  16. # <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
  17. # and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
  18. # License, Version 2.0.
  19. #
  20. from io import BytesIO
  21. import base64
  22. import os
  23. import sys
  24. import shutil
  25. import tempfile
  26. import warnings
  27. from urllib.parse import (
  28. quote as urlquote,
  29. urlparse,
  30. )
  31. import dulwich
  32. from dulwich import (
  33. client,
  34. )
  35. from dulwich.client import (
  36. InvalidWants,
  37. LocalGitClient,
  38. TraditionalGitClient,
  39. TCPGitClient,
  40. SSHGitClient,
  41. HttpGitClient,
  42. FetchPackResult,
  43. ReportStatusParser,
  44. SendPackError,
  45. StrangeHostname,
  46. SubprocessSSHVendor,
  47. PLinkSSHVendor,
  48. HangupException,
  49. GitProtocolError,
  50. check_wants,
  51. default_urllib3_manager,
  52. get_credentials_from_store,
  53. get_transport_and_path,
  54. get_transport_and_path_from_url,
  55. parse_rsync_url,
  56. _remote_error_from_stderr,
  57. )
  58. from dulwich.config import (
  59. ConfigDict,
  60. )
  61. from dulwich.tests import (
  62. TestCase,
  63. )
  64. from dulwich.protocol import (
  65. TCP_GIT_PORT,
  66. Protocol,
  67. )
  68. from dulwich.pack import (
  69. pack_objects_to_data,
  70. write_pack_data,
  71. write_pack_objects,
  72. )
  73. from dulwich.objects import Commit, Tree
  74. from dulwich.repo import (
  75. MemoryRepo,
  76. Repo,
  77. )
  78. from dulwich.tests import skipIf
  79. from dulwich.tests.utils import (
  80. open_repo,
  81. tear_down_repo,
  82. setup_warning_catcher,
  83. )
  84. class DummyClient(TraditionalGitClient):
  85. def __init__(self, can_read, read, write):
  86. self.can_read = can_read
  87. self.read = read
  88. self.write = write
  89. TraditionalGitClient.__init__(self)
  90. def _connect(self, service, path):
  91. return Protocol(self.read, self.write), self.can_read, None
  92. class DummyPopen:
  93. def __init__(self, *args, **kwards):
  94. self.stdin = BytesIO(b"stdin")
  95. self.stdout = BytesIO(b"stdout")
  96. self.stderr = BytesIO(b"stderr")
  97. self.returncode = 0
  98. self.args = args
  99. self.kwargs = kwards
  100. def communicate(self, *args, **kwards):
  101. return ("Running", "")
  102. def wait(self, *args, **kwards):
  103. return False
  104. # TODO(durin42): add unit-level tests of GitClient
  105. class GitClientTests(TestCase):
  106. def setUp(self):
  107. super(GitClientTests, self).setUp()
  108. self.rout = BytesIO()
  109. self.rin = BytesIO()
  110. self.client = DummyClient(lambda x: True, self.rin.read, self.rout.write)
  111. def test_caps(self):
  112. agent_cap = ("agent=dulwich/%d.%d.%d" % dulwich.__version__).encode("ascii")
  113. self.assertEqual(
  114. set(
  115. [
  116. b"multi_ack",
  117. b"side-band-64k",
  118. b"ofs-delta",
  119. b"thin-pack",
  120. b"multi_ack_detailed",
  121. b"shallow",
  122. agent_cap,
  123. ]
  124. ),
  125. set(self.client._fetch_capabilities),
  126. )
  127. self.assertEqual(
  128. set(
  129. [
  130. b"delete-refs",
  131. b"ofs-delta",
  132. b"report-status",
  133. b"side-band-64k",
  134. agent_cap,
  135. ]
  136. ),
  137. set(self.client._send_capabilities),
  138. )
  139. def test_archive_ack(self):
  140. self.rin.write(b"0009NACK\n" b"0000")
  141. self.rin.seek(0)
  142. self.client.archive(b"bla", b"HEAD", None, None)
  143. self.assertEqual(self.rout.getvalue(), b"0011argument HEAD0000")
  144. def test_fetch_empty(self):
  145. self.rin.write(b"0000")
  146. self.rin.seek(0)
  147. def check_heads(heads, **kwargs):
  148. self.assertEqual(heads, {})
  149. return []
  150. ret = self.client.fetch_pack(b"/", check_heads, None, None)
  151. self.assertEqual({}, ret.refs)
  152. self.assertEqual({}, ret.symrefs)
  153. def test_fetch_pack_ignores_magic_ref(self):
  154. self.rin.write(
  155. b"00000000000000000000000000000000000000000000 capabilities^{}"
  156. b"\x00 multi_ack "
  157. b"thin-pack side-band side-band-64k ofs-delta shallow no-progress "
  158. b"include-tag\n"
  159. b"0000"
  160. )
  161. self.rin.seek(0)
  162. def check_heads(heads, **kwargs):
  163. self.assertEqual({}, heads)
  164. return []
  165. ret = self.client.fetch_pack(b"bla", check_heads, None, None, None)
  166. self.assertEqual({}, ret.refs)
  167. self.assertEqual({}, ret.symrefs)
  168. self.assertEqual(self.rout.getvalue(), b"0000")
  169. def test_fetch_pack_none(self):
  170. self.rin.write(
  171. b"008855dcc6bf963f922e1ed5c4bbaaefcfacef57b1d7 HEAD\x00multi_ack "
  172. b"thin-pack side-band side-band-64k ofs-delta shallow no-progress "
  173. b"include-tag\n"
  174. b"0000"
  175. )
  176. self.rin.seek(0)
  177. ret = self.client.fetch_pack(b"bla", lambda heads, **kwargs: [], None, None, None)
  178. self.assertEqual(
  179. {b"HEAD": b"55dcc6bf963f922e1ed5c4bbaaefcfacef57b1d7"}, ret.refs
  180. )
  181. self.assertEqual({}, ret.symrefs)
  182. self.assertEqual(self.rout.getvalue(), b"0000")
  183. def test_send_pack_no_sideband64k_with_update_ref_error(self):
  184. # No side-bank-64k reported by server shouldn't try to parse
  185. # side band data
  186. pkts = [
  187. b"55dcc6bf963f922e1ed5c4bbaaefcfacef57b1d7 capabilities^{}"
  188. b"\x00 report-status delete-refs ofs-delta\n",
  189. b"",
  190. b"unpack ok",
  191. b"ng refs/foo/bar pre-receive hook declined",
  192. b"",
  193. ]
  194. for pkt in pkts:
  195. if pkt == b"":
  196. self.rin.write(b"0000")
  197. else:
  198. self.rin.write(("%04x" % (len(pkt) + 4)).encode("ascii") + pkt)
  199. self.rin.seek(0)
  200. tree = Tree()
  201. commit = Commit()
  202. commit.tree = tree
  203. commit.parents = []
  204. commit.author = commit.committer = b"test user"
  205. commit.commit_time = commit.author_time = 1174773719
  206. commit.commit_timezone = commit.author_timezone = 0
  207. commit.encoding = b"UTF-8"
  208. commit.message = b"test message"
  209. def update_refs(refs):
  210. return {
  211. b"refs/foo/bar": commit.id,
  212. }
  213. def generate_pack_data(have, want, ofs_delta=False):
  214. return pack_objects_to_data(
  215. [
  216. (commit, None),
  217. (tree, ""),
  218. ]
  219. )
  220. result = self.client.send_pack("blah", update_refs, generate_pack_data)
  221. self.assertEqual(
  222. {b"refs/foo/bar": "pre-receive hook declined"}, result.ref_status
  223. )
  224. self.assertEqual({b"refs/foo/bar": commit.id}, result.refs)
  225. def test_send_pack_none(self):
  226. # Set ref to current value
  227. self.rin.write(
  228. b"0078310ca9477129b8586fa2afc779c1f57cf64bba6c "
  229. b"refs/heads/master\x00 report-status delete-refs "
  230. b"side-band-64k quiet ofs-delta\n"
  231. b"0000"
  232. )
  233. self.rin.seek(0)
  234. def update_refs(refs):
  235. return {b"refs/heads/master": b"310ca9477129b8586fa2afc779c1f57cf64bba6c"}
  236. def generate_pack_data(have, want, ofs_delta=False):
  237. return 0, []
  238. self.client.send_pack(b"/", update_refs, generate_pack_data)
  239. self.assertEqual(self.rout.getvalue(), b"0000")
  240. def test_send_pack_keep_and_delete(self):
  241. self.rin.write(
  242. b"0063310ca9477129b8586fa2afc779c1f57cf64bba6c "
  243. b"refs/heads/master\x00report-status delete-refs ofs-delta\n"
  244. b"003f310ca9477129b8586fa2afc779c1f57cf64bba6c refs/heads/keepme\n"
  245. b"0000000eunpack ok\n"
  246. b"0019ok refs/heads/master\n"
  247. b"0000"
  248. )
  249. self.rin.seek(0)
  250. def update_refs(refs):
  251. return {b"refs/heads/master": b"0" * 40}
  252. def generate_pack_data(have, want, ofs_delta=False):
  253. return 0, []
  254. self.client.send_pack(b"/", update_refs, generate_pack_data)
  255. self.assertEqual(
  256. self.rout.getvalue(),
  257. b"008b310ca9477129b8586fa2afc779c1f57cf64bba6c "
  258. b"0000000000000000000000000000000000000000 "
  259. b"refs/heads/master\x00delete-refs ofs-delta report-status0000",
  260. )
  261. def test_send_pack_delete_only(self):
  262. self.rin.write(
  263. b"0063310ca9477129b8586fa2afc779c1f57cf64bba6c "
  264. b"refs/heads/master\x00report-status delete-refs ofs-delta\n"
  265. b"0000000eunpack ok\n"
  266. b"0019ok refs/heads/master\n"
  267. b"0000"
  268. )
  269. self.rin.seek(0)
  270. def update_refs(refs):
  271. return {b"refs/heads/master": b"0" * 40}
  272. def generate_pack_data(have, want, ofs_delta=False):
  273. return 0, []
  274. self.client.send_pack(b"/", update_refs, generate_pack_data)
  275. self.assertEqual(
  276. self.rout.getvalue(),
  277. b"008b310ca9477129b8586fa2afc779c1f57cf64bba6c "
  278. b"0000000000000000000000000000000000000000 "
  279. b"refs/heads/master\x00delete-refs ofs-delta report-status0000",
  280. )
  281. def test_send_pack_new_ref_only(self):
  282. self.rin.write(
  283. b"0063310ca9477129b8586fa2afc779c1f57cf64bba6c "
  284. b"refs/heads/master\x00report-status delete-refs ofs-delta\n"
  285. b"0000000eunpack ok\n"
  286. b"0019ok refs/heads/blah12\n"
  287. b"0000"
  288. )
  289. self.rin.seek(0)
  290. def update_refs(refs):
  291. return {
  292. b"refs/heads/blah12": b"310ca9477129b8586fa2afc779c1f57cf64bba6c",
  293. b"refs/heads/master": b"310ca9477129b8586fa2afc779c1f57cf64bba6c",
  294. }
  295. def generate_pack_data(have, want, ofs_delta=False):
  296. return 0, []
  297. f = BytesIO()
  298. write_pack_objects(f, {})
  299. self.client.send_pack("/", update_refs, generate_pack_data)
  300. self.assertEqual(
  301. self.rout.getvalue(),
  302. b"008b0000000000000000000000000000000000000000 "
  303. b"310ca9477129b8586fa2afc779c1f57cf64bba6c "
  304. b"refs/heads/blah12\x00delete-refs ofs-delta report-status0000"
  305. + f.getvalue(),
  306. )
  307. def test_send_pack_new_ref(self):
  308. self.rin.write(
  309. b"0064310ca9477129b8586fa2afc779c1f57cf64bba6c "
  310. b"refs/heads/master\x00 report-status delete-refs ofs-delta\n"
  311. b"0000000eunpack ok\n"
  312. b"0019ok refs/heads/blah12\n"
  313. b"0000"
  314. )
  315. self.rin.seek(0)
  316. tree = Tree()
  317. commit = Commit()
  318. commit.tree = tree
  319. commit.parents = []
  320. commit.author = commit.committer = b"test user"
  321. commit.commit_time = commit.author_time = 1174773719
  322. commit.commit_timezone = commit.author_timezone = 0
  323. commit.encoding = b"UTF-8"
  324. commit.message = b"test message"
  325. def update_refs(refs):
  326. return {
  327. b"refs/heads/blah12": commit.id,
  328. b"refs/heads/master": b"310ca9477129b8586fa2afc779c1f57cf64bba6c",
  329. }
  330. def generate_pack_data(have, want, ofs_delta=False):
  331. return pack_objects_to_data(
  332. [
  333. (commit, None),
  334. (tree, b""),
  335. ]
  336. )
  337. f = BytesIO()
  338. write_pack_data(f, *generate_pack_data(None, None))
  339. self.client.send_pack(b"/", update_refs, generate_pack_data)
  340. self.assertEqual(
  341. self.rout.getvalue(),
  342. b"008b0000000000000000000000000000000000000000 "
  343. + commit.id
  344. + b" refs/heads/blah12\x00delete-refs ofs-delta report-status0000"
  345. + f.getvalue(),
  346. )
  347. def test_send_pack_no_deleteref_delete_only(self):
  348. pkts = [
  349. b"310ca9477129b8586fa2afc779c1f57cf64bba6c refs/heads/master"
  350. b"\x00 report-status ofs-delta\n",
  351. b"",
  352. b"",
  353. ]
  354. for pkt in pkts:
  355. if pkt == b"":
  356. self.rin.write(b"0000")
  357. else:
  358. self.rin.write(("%04x" % (len(pkt) + 4)).encode("ascii") + pkt)
  359. self.rin.seek(0)
  360. def update_refs(refs):
  361. return {b"refs/heads/master": b"0" * 40}
  362. def generate_pack_data(have, want, ofs_delta=False):
  363. return 0, []
  364. result = self.client.send_pack(b"/", update_refs, generate_pack_data)
  365. self.assertEqual(
  366. result.ref_status,
  367. {b"refs/heads/master": "remote does not support deleting refs"},
  368. )
  369. self.assertEqual(
  370. result.refs,
  371. {b"refs/heads/master": b"310ca9477129b8586fa2afc779c1f57cf64bba6c"},
  372. )
  373. self.assertEqual(self.rout.getvalue(), b"0000")
  374. class TestGetTransportAndPath(TestCase):
  375. def test_tcp(self):
  376. c, path = get_transport_and_path("git://foo.com/bar/baz")
  377. self.assertTrue(isinstance(c, TCPGitClient))
  378. self.assertEqual("foo.com", c._host)
  379. self.assertEqual(TCP_GIT_PORT, c._port)
  380. self.assertEqual("/bar/baz", path)
  381. def test_tcp_port(self):
  382. c, path = get_transport_and_path("git://foo.com:1234/bar/baz")
  383. self.assertTrue(isinstance(c, TCPGitClient))
  384. self.assertEqual("foo.com", c._host)
  385. self.assertEqual(1234, c._port)
  386. self.assertEqual("/bar/baz", path)
  387. def test_git_ssh_explicit(self):
  388. c, path = get_transport_and_path("git+ssh://foo.com/bar/baz")
  389. self.assertTrue(isinstance(c, SSHGitClient))
  390. self.assertEqual("foo.com", c.host)
  391. self.assertEqual(None, c.port)
  392. self.assertEqual(None, c.username)
  393. self.assertEqual("/bar/baz", path)
  394. def test_ssh_explicit(self):
  395. c, path = get_transport_and_path("ssh://foo.com/bar/baz")
  396. self.assertTrue(isinstance(c, SSHGitClient))
  397. self.assertEqual("foo.com", c.host)
  398. self.assertEqual(None, c.port)
  399. self.assertEqual(None, c.username)
  400. self.assertEqual("/bar/baz", path)
  401. def test_ssh_port_explicit(self):
  402. c, path = get_transport_and_path("git+ssh://foo.com:1234/bar/baz")
  403. self.assertTrue(isinstance(c, SSHGitClient))
  404. self.assertEqual("foo.com", c.host)
  405. self.assertEqual(1234, c.port)
  406. self.assertEqual("/bar/baz", path)
  407. def test_username_and_port_explicit_unknown_scheme(self):
  408. c, path = get_transport_and_path("unknown://git@server:7999/dply/stuff.git")
  409. self.assertTrue(isinstance(c, SSHGitClient))
  410. self.assertEqual("unknown", c.host)
  411. self.assertEqual("//git@server:7999/dply/stuff.git", path)
  412. def test_username_and_port_explicit(self):
  413. c, path = get_transport_and_path("ssh://git@server:7999/dply/stuff.git")
  414. self.assertTrue(isinstance(c, SSHGitClient))
  415. self.assertEqual("git", c.username)
  416. self.assertEqual("server", c.host)
  417. self.assertEqual(7999, c.port)
  418. self.assertEqual("/dply/stuff.git", path)
  419. def test_ssh_abspath_doubleslash(self):
  420. c, path = get_transport_and_path("git+ssh://foo.com//bar/baz")
  421. self.assertTrue(isinstance(c, SSHGitClient))
  422. self.assertEqual("foo.com", c.host)
  423. self.assertEqual(None, c.port)
  424. self.assertEqual(None, c.username)
  425. self.assertEqual("//bar/baz", path)
  426. def test_ssh_port(self):
  427. c, path = get_transport_and_path("git+ssh://foo.com:1234/bar/baz")
  428. self.assertTrue(isinstance(c, SSHGitClient))
  429. self.assertEqual("foo.com", c.host)
  430. self.assertEqual(1234, c.port)
  431. self.assertEqual("/bar/baz", path)
  432. def test_ssh_implicit(self):
  433. c, path = get_transport_and_path("foo:/bar/baz")
  434. self.assertTrue(isinstance(c, SSHGitClient))
  435. self.assertEqual("foo", c.host)
  436. self.assertEqual(None, c.port)
  437. self.assertEqual(None, c.username)
  438. self.assertEqual("/bar/baz", path)
  439. def test_ssh_host(self):
  440. c, path = get_transport_and_path("foo.com:/bar/baz")
  441. self.assertTrue(isinstance(c, SSHGitClient))
  442. self.assertEqual("foo.com", c.host)
  443. self.assertEqual(None, c.port)
  444. self.assertEqual(None, c.username)
  445. self.assertEqual("/bar/baz", path)
  446. def test_ssh_user_host(self):
  447. c, path = get_transport_and_path("user@foo.com:/bar/baz")
  448. self.assertTrue(isinstance(c, SSHGitClient))
  449. self.assertEqual("foo.com", c.host)
  450. self.assertEqual(None, c.port)
  451. self.assertEqual("user", c.username)
  452. self.assertEqual("/bar/baz", path)
  453. def test_ssh_relpath(self):
  454. c, path = get_transport_and_path("foo:bar/baz")
  455. self.assertTrue(isinstance(c, SSHGitClient))
  456. self.assertEqual("foo", c.host)
  457. self.assertEqual(None, c.port)
  458. self.assertEqual(None, c.username)
  459. self.assertEqual("bar/baz", path)
  460. def test_ssh_host_relpath(self):
  461. c, path = get_transport_and_path("foo.com:bar/baz")
  462. self.assertTrue(isinstance(c, SSHGitClient))
  463. self.assertEqual("foo.com", c.host)
  464. self.assertEqual(None, c.port)
  465. self.assertEqual(None, c.username)
  466. self.assertEqual("bar/baz", path)
  467. def test_ssh_user_host_relpath(self):
  468. c, path = get_transport_and_path("user@foo.com:bar/baz")
  469. self.assertTrue(isinstance(c, SSHGitClient))
  470. self.assertEqual("foo.com", c.host)
  471. self.assertEqual(None, c.port)
  472. self.assertEqual("user", c.username)
  473. self.assertEqual("bar/baz", path)
  474. def test_local(self):
  475. c, path = get_transport_and_path("foo.bar/baz")
  476. self.assertTrue(isinstance(c, LocalGitClient))
  477. self.assertEqual("foo.bar/baz", path)
  478. @skipIf(sys.platform != "win32", "Behaviour only happens on windows.")
  479. def test_local_abs_windows_path(self):
  480. c, path = get_transport_and_path("C:\\foo.bar\\baz")
  481. self.assertTrue(isinstance(c, LocalGitClient))
  482. self.assertEqual("C:\\foo.bar\\baz", path)
  483. def test_error(self):
  484. # Need to use a known urlparse.uses_netloc URL scheme to get the
  485. # expected parsing of the URL on Python versions less than 2.6.5
  486. c, path = get_transport_and_path("prospero://bar/baz")
  487. self.assertTrue(isinstance(c, SSHGitClient))
  488. def test_http(self):
  489. url = "https://github.com/jelmer/dulwich"
  490. c, path = get_transport_and_path(url)
  491. self.assertTrue(isinstance(c, HttpGitClient))
  492. self.assertEqual("/jelmer/dulwich", path)
  493. def test_http_auth(self):
  494. url = "https://user:passwd@github.com/jelmer/dulwich"
  495. c, path = get_transport_and_path(url)
  496. self.assertTrue(isinstance(c, HttpGitClient))
  497. self.assertEqual("/jelmer/dulwich", path)
  498. self.assertEqual("user", c._username)
  499. self.assertEqual("passwd", c._password)
  500. def test_http_auth_with_username(self):
  501. url = "https://github.com/jelmer/dulwich"
  502. c, path = get_transport_and_path(url, username="user2", password="blah")
  503. self.assertTrue(isinstance(c, HttpGitClient))
  504. self.assertEqual("/jelmer/dulwich", path)
  505. self.assertEqual("user2", c._username)
  506. self.assertEqual("blah", c._password)
  507. def test_http_auth_with_username_and_in_url(self):
  508. url = "https://user:passwd@github.com/jelmer/dulwich"
  509. c, path = get_transport_and_path(url, username="user2", password="blah")
  510. self.assertTrue(isinstance(c, HttpGitClient))
  511. self.assertEqual("/jelmer/dulwich", path)
  512. self.assertEqual("user", c._username)
  513. self.assertEqual("passwd", c._password)
  514. def test_http_no_auth(self):
  515. url = "https://github.com/jelmer/dulwich"
  516. c, path = get_transport_and_path(url)
  517. self.assertTrue(isinstance(c, HttpGitClient))
  518. self.assertEqual("/jelmer/dulwich", path)
  519. self.assertIs(None, c._username)
  520. self.assertIs(None, c._password)
  521. class TestGetTransportAndPathFromUrl(TestCase):
  522. def test_tcp(self):
  523. c, path = get_transport_and_path_from_url("git://foo.com/bar/baz")
  524. self.assertTrue(isinstance(c, TCPGitClient))
  525. self.assertEqual("foo.com", c._host)
  526. self.assertEqual(TCP_GIT_PORT, c._port)
  527. self.assertEqual("/bar/baz", path)
  528. def test_tcp_port(self):
  529. c, path = get_transport_and_path_from_url("git://foo.com:1234/bar/baz")
  530. self.assertTrue(isinstance(c, TCPGitClient))
  531. self.assertEqual("foo.com", c._host)
  532. self.assertEqual(1234, c._port)
  533. self.assertEqual("/bar/baz", path)
  534. def test_ssh_explicit(self):
  535. c, path = get_transport_and_path_from_url("git+ssh://foo.com/bar/baz")
  536. self.assertTrue(isinstance(c, SSHGitClient))
  537. self.assertEqual("foo.com", c.host)
  538. self.assertEqual(None, c.port)
  539. self.assertEqual(None, c.username)
  540. self.assertEqual("/bar/baz", path)
  541. def test_ssh_port_explicit(self):
  542. c, path = get_transport_and_path_from_url("git+ssh://foo.com:1234/bar/baz")
  543. self.assertTrue(isinstance(c, SSHGitClient))
  544. self.assertEqual("foo.com", c.host)
  545. self.assertEqual(1234, c.port)
  546. self.assertEqual("/bar/baz", path)
  547. def test_ssh_homepath(self):
  548. c, path = get_transport_and_path_from_url("git+ssh://foo.com/~/bar/baz")
  549. self.assertTrue(isinstance(c, SSHGitClient))
  550. self.assertEqual("foo.com", c.host)
  551. self.assertEqual(None, c.port)
  552. self.assertEqual(None, c.username)
  553. self.assertEqual("/~/bar/baz", path)
  554. def test_ssh_port_homepath(self):
  555. c, path = get_transport_and_path_from_url("git+ssh://foo.com:1234/~/bar/baz")
  556. self.assertTrue(isinstance(c, SSHGitClient))
  557. self.assertEqual("foo.com", c.host)
  558. self.assertEqual(1234, c.port)
  559. self.assertEqual("/~/bar/baz", path)
  560. def test_ssh_host_relpath(self):
  561. self.assertRaises(
  562. ValueError, get_transport_and_path_from_url, "foo.com:bar/baz"
  563. )
  564. def test_ssh_user_host_relpath(self):
  565. self.assertRaises(
  566. ValueError, get_transport_and_path_from_url, "user@foo.com:bar/baz"
  567. )
  568. def test_local_path(self):
  569. self.assertRaises(ValueError, get_transport_and_path_from_url, "foo.bar/baz")
  570. def test_error(self):
  571. # Need to use a known urlparse.uses_netloc URL scheme to get the
  572. # expected parsing of the URL on Python versions less than 2.6.5
  573. self.assertRaises(
  574. ValueError, get_transport_and_path_from_url, "prospero://bar/baz"
  575. )
  576. def test_http(self):
  577. url = "https://github.com/jelmer/dulwich"
  578. c, path = get_transport_and_path_from_url(url)
  579. self.assertTrue(isinstance(c, HttpGitClient))
  580. self.assertEqual("https://github.com", c.get_url(b"/"))
  581. self.assertEqual("/jelmer/dulwich", path)
  582. def test_http_port(self):
  583. url = "https://github.com:9090/jelmer/dulwich"
  584. c, path = get_transport_and_path_from_url(url)
  585. self.assertEqual("https://github.com:9090", c.get_url(b"/"))
  586. self.assertTrue(isinstance(c, HttpGitClient))
  587. self.assertEqual("/jelmer/dulwich", path)
  588. def test_file(self):
  589. c, path = get_transport_and_path_from_url("file:///home/jelmer/foo")
  590. self.assertTrue(isinstance(c, LocalGitClient))
  591. self.assertEqual("/home/jelmer/foo", path)
  592. class TestSSHVendor(object):
  593. def __init__(self):
  594. self.host = None
  595. self.command = ""
  596. self.username = None
  597. self.port = None
  598. self.password = None
  599. self.key_filename = None
  600. def run_command(
  601. self,
  602. host,
  603. command,
  604. username=None,
  605. port=None,
  606. password=None,
  607. key_filename=None,
  608. ):
  609. self.host = host
  610. self.command = command
  611. self.username = username
  612. self.port = port
  613. self.password = password
  614. self.key_filename = key_filename
  615. class Subprocess:
  616. pass
  617. setattr(Subprocess, "read", lambda: None)
  618. setattr(Subprocess, "write", lambda: None)
  619. setattr(Subprocess, "close", lambda: None)
  620. setattr(Subprocess, "can_read", lambda: None)
  621. return Subprocess()
  622. class SSHGitClientTests(TestCase):
  623. def setUp(self):
  624. super(SSHGitClientTests, self).setUp()
  625. self.server = TestSSHVendor()
  626. self.real_vendor = client.get_ssh_vendor
  627. client.get_ssh_vendor = lambda: self.server
  628. self.client = SSHGitClient("git.samba.org")
  629. def tearDown(self):
  630. super(SSHGitClientTests, self).tearDown()
  631. client.get_ssh_vendor = self.real_vendor
  632. def test_get_url(self):
  633. path = "/tmp/repo.git"
  634. c = SSHGitClient("git.samba.org")
  635. url = c.get_url(path)
  636. self.assertEqual("ssh://git.samba.org/tmp/repo.git", url)
  637. def test_get_url_with_username_and_port(self):
  638. path = "/tmp/repo.git"
  639. c = SSHGitClient("git.samba.org", port=2222, username="user")
  640. url = c.get_url(path)
  641. self.assertEqual("ssh://user@git.samba.org:2222/tmp/repo.git", url)
  642. def test_default_command(self):
  643. self.assertEqual(b"git-upload-pack", self.client._get_cmd_path(b"upload-pack"))
  644. def test_alternative_command_path(self):
  645. self.client.alternative_paths[b"upload-pack"] = b"/usr/lib/git/git-upload-pack"
  646. self.assertEqual(
  647. b"/usr/lib/git/git-upload-pack",
  648. self.client._get_cmd_path(b"upload-pack"),
  649. )
  650. def test_alternative_command_path_spaces(self):
  651. self.client.alternative_paths[
  652. b"upload-pack"
  653. ] = b"/usr/lib/git/git-upload-pack -ibla"
  654. self.assertEqual(
  655. b"/usr/lib/git/git-upload-pack -ibla",
  656. self.client._get_cmd_path(b"upload-pack"),
  657. )
  658. def test_connect(self):
  659. server = self.server
  660. client = self.client
  661. client.username = b"username"
  662. client.port = 1337
  663. client._connect(b"command", b"/path/to/repo")
  664. self.assertEqual(b"username", server.username)
  665. self.assertEqual(1337, server.port)
  666. self.assertEqual("git-command '/path/to/repo'", server.command)
  667. client._connect(b"relative-command", b"/~/path/to/repo")
  668. self.assertEqual("git-relative-command '~/path/to/repo'", server.command)
  669. class ReportStatusParserTests(TestCase):
  670. def test_invalid_pack(self):
  671. parser = ReportStatusParser()
  672. parser.handle_packet(b"unpack error - foo bar")
  673. parser.handle_packet(b"ok refs/foo/bar")
  674. parser.handle_packet(None)
  675. self.assertRaises(SendPackError, list, parser.check())
  676. def test_update_refs_error(self):
  677. parser = ReportStatusParser()
  678. parser.handle_packet(b"unpack ok")
  679. parser.handle_packet(b"ng refs/foo/bar need to pull")
  680. parser.handle_packet(None)
  681. self.assertEqual([(b"refs/foo/bar", "need to pull")], list(parser.check()))
  682. def test_ok(self):
  683. parser = ReportStatusParser()
  684. parser.handle_packet(b"unpack ok")
  685. parser.handle_packet(b"ok refs/foo/bar")
  686. parser.handle_packet(None)
  687. self.assertEqual([(b"refs/foo/bar", None)], list(parser.check()))
  688. class LocalGitClientTests(TestCase):
  689. def test_get_url(self):
  690. path = "/tmp/repo.git"
  691. c = LocalGitClient()
  692. url = c.get_url(path)
  693. self.assertEqual("file:///tmp/repo.git", url)
  694. def test_fetch_into_empty(self):
  695. c = LocalGitClient()
  696. t = MemoryRepo()
  697. s = open_repo("a.git")
  698. self.addCleanup(tear_down_repo, s)
  699. self.assertEqual(s.get_refs(), c.fetch(s.path, t).refs)
  700. def test_fetch_empty(self):
  701. c = LocalGitClient()
  702. s = open_repo("a.git")
  703. self.addCleanup(tear_down_repo, s)
  704. out = BytesIO()
  705. walker = {}
  706. ret = c.fetch_pack(
  707. s.path, lambda heads, **kwargs: [], graph_walker=walker, pack_data=out.write
  708. )
  709. self.assertEqual(
  710. {
  711. b"HEAD": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  712. b"refs/heads/master": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  713. b"refs/tags/mytag": b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a",
  714. b"refs/tags/mytag-packed": b"b0931cadc54336e78a1d980420e3268903b57a50",
  715. },
  716. ret.refs,
  717. )
  718. self.assertEqual({b"HEAD": b"refs/heads/master"}, ret.symrefs)
  719. self.assertEqual(
  720. b"PACK\x00\x00\x00\x02\x00\x00\x00\x00\x02\x9d\x08"
  721. b"\x82;\xd8\xa8\xea\xb5\x10\xadj\xc7\\\x82<\xfd>\xd3\x1e",
  722. out.getvalue(),
  723. )
  724. def test_fetch_pack_none(self):
  725. c = LocalGitClient()
  726. s = open_repo("a.git")
  727. self.addCleanup(tear_down_repo, s)
  728. out = BytesIO()
  729. walker = MemoryRepo().get_graph_walker()
  730. ret = c.fetch_pack(
  731. s.path,
  732. lambda heads, **kwargs: [b"a90fa2d900a17e99b433217e988c4eb4a2e9a097"],
  733. graph_walker=walker,
  734. pack_data=out.write,
  735. )
  736. self.assertEqual({b"HEAD": b"refs/heads/master"}, ret.symrefs)
  737. self.assertEqual(
  738. {
  739. b"HEAD": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  740. b"refs/heads/master": b"a90fa2d900a17e99b433217e988c4eb4a2e9a097",
  741. b"refs/tags/mytag": b"28237f4dc30d0d462658d6b937b08a0f0b6ef55a",
  742. b"refs/tags/mytag-packed": b"b0931cadc54336e78a1d980420e3268903b57a50",
  743. },
  744. ret.refs,
  745. )
  746. # Hardcoding is not ideal, but we'll fix that some other day..
  747. self.assertTrue(
  748. out.getvalue().startswith(b"PACK\x00\x00\x00\x02\x00\x00\x00\x07")
  749. )
  750. def test_send_pack_without_changes(self):
  751. local = open_repo("a.git")
  752. self.addCleanup(tear_down_repo, local)
  753. target = open_repo("a.git")
  754. self.addCleanup(tear_down_repo, target)
  755. self.send_and_verify(b"master", local, target)
  756. def test_send_pack_with_changes(self):
  757. local = open_repo("a.git")
  758. self.addCleanup(tear_down_repo, local)
  759. target_path = tempfile.mkdtemp()
  760. self.addCleanup(shutil.rmtree, target_path)
  761. with Repo.init_bare(target_path) as target:
  762. self.send_and_verify(b"master", local, target)
  763. def test_get_refs(self):
  764. local = open_repo("refs.git")
  765. self.addCleanup(tear_down_repo, local)
  766. client = LocalGitClient()
  767. refs = client.get_refs(local.path)
  768. self.assertDictEqual(local.refs.as_dict(), refs)
  769. def send_and_verify(self, branch, local, target):
  770. """Send branch from local to remote repository and verify it worked."""
  771. client = LocalGitClient()
  772. ref_name = b"refs/heads/" + branch
  773. result = client.send_pack(
  774. target.path,
  775. lambda _: {ref_name: local.refs[ref_name]},
  776. local.generate_pack_data,
  777. )
  778. self.assertEqual(local.refs[ref_name], result.refs[ref_name])
  779. self.assertIs(None, result.agent)
  780. self.assertEqual({}, result.ref_status)
  781. obj_local = local.get_object(result.refs[ref_name])
  782. obj_target = target.get_object(result.refs[ref_name])
  783. self.assertEqual(obj_local, obj_target)
  784. class HttpGitClientTests(TestCase):
  785. def test_get_url(self):
  786. base_url = "https://github.com/jelmer/dulwich"
  787. path = "/jelmer/dulwich"
  788. c = HttpGitClient(base_url)
  789. url = c.get_url(path)
  790. self.assertEqual("https://github.com/jelmer/dulwich", url)
  791. def test_get_url_bytes_path(self):
  792. base_url = "https://github.com/jelmer/dulwich"
  793. path_bytes = b"/jelmer/dulwich"
  794. c = HttpGitClient(base_url)
  795. url = c.get_url(path_bytes)
  796. self.assertEqual("https://github.com/jelmer/dulwich", url)
  797. def test_get_url_with_username_and_passwd(self):
  798. base_url = "https://github.com/jelmer/dulwich"
  799. path = "/jelmer/dulwich"
  800. c = HttpGitClient(base_url, username="USERNAME", password="PASSWD")
  801. url = c.get_url(path)
  802. self.assertEqual("https://github.com/jelmer/dulwich", url)
  803. def test_init_username_passwd_set(self):
  804. url = "https://github.com/jelmer/dulwich"
  805. c = HttpGitClient(url, config=None, username="user", password="passwd")
  806. self.assertEqual("user", c._username)
  807. self.assertEqual("passwd", c._password)
  808. basic_auth = c.pool_manager.headers["authorization"]
  809. auth_string = "%s:%s" % ("user", "passwd")
  810. b64_credentials = base64.b64encode(auth_string.encode("latin1"))
  811. expected_basic_auth = "Basic %s" % b64_credentials.decode("latin1")
  812. self.assertEqual(basic_auth, expected_basic_auth)
  813. def test_init_no_username_passwd(self):
  814. url = "https://github.com/jelmer/dulwich"
  815. c = HttpGitClient(url, config=None)
  816. self.assertIs(None, c._username)
  817. self.assertIs(None, c._password)
  818. self.assertNotIn("authorization", c.pool_manager.headers)
  819. def test_from_parsedurl_on_url_with_quoted_credentials(self):
  820. original_username = "john|the|first"
  821. quoted_username = urlquote(original_username)
  822. original_password = "Ya#1$2%3"
  823. quoted_password = urlquote(original_password)
  824. url = "https://{username}:{password}@github.com/jelmer/dulwich".format(
  825. username=quoted_username, password=quoted_password
  826. )
  827. c = HttpGitClient.from_parsedurl(urlparse(url))
  828. self.assertEqual(original_username, c._username)
  829. self.assertEqual(original_password, c._password)
  830. basic_auth = c.pool_manager.headers["authorization"]
  831. auth_string = "%s:%s" % (original_username, original_password)
  832. b64_credentials = base64.b64encode(auth_string.encode("latin1"))
  833. expected_basic_auth = "Basic %s" % b64_credentials.decode("latin1")
  834. self.assertEqual(basic_auth, expected_basic_auth)
  835. def test_url_redirect_location(self):
  836. from urllib3.response import HTTPResponse
  837. test_data = {
  838. "https://gitlab.com/inkscape/inkscape/": {
  839. "redirect_url": "https://gitlab.com/inkscape/inkscape.git/",
  840. "refs_data": (
  841. b"001e# service=git-upload-pack\n00000032"
  842. b"fb2bebf4919a011f0fd7cec085443d0031228e76 "
  843. b"HEAD\n0000"
  844. ),
  845. },
  846. "https://github.com/jelmer/dulwich/": {
  847. "redirect_url": "https://github.com/jelmer/dulwich/",
  848. "refs_data": (
  849. b"001e# service=git-upload-pack\n00000032"
  850. b"3ff25e09724aa4d86ea5bca7d5dd0399a3c8bfcf "
  851. b"HEAD\n0000"
  852. ),
  853. },
  854. }
  855. tail = "info/refs?service=git-upload-pack"
  856. # we need to mock urllib3.PoolManager as this test will fail
  857. # otherwise without an active internet connection
  858. class PoolManagerMock:
  859. def __init__(self):
  860. self.headers = {}
  861. def request(self, method, url, fields=None, headers=None, redirect=True):
  862. base_url = url[: -len(tail)]
  863. redirect_base_url = test_data[base_url]["redirect_url"]
  864. redirect_url = redirect_base_url + tail
  865. headers = {
  866. "Content-Type": "application/x-git-upload-pack-advertisement"
  867. }
  868. body = test_data[base_url]["refs_data"]
  869. # urllib3 handles automatic redirection by default
  870. status = 200
  871. request_url = redirect_url
  872. # simulate urllib3 behavior when redirect parameter is False
  873. if redirect is False:
  874. request_url = url
  875. if redirect_base_url != base_url:
  876. body = ""
  877. headers["location"] = redirect_url
  878. status = 301
  879. return HTTPResponse(
  880. body=body,
  881. headers=headers,
  882. request_method=method,
  883. request_url=request_url,
  884. status=status,
  885. )
  886. pool_manager = PoolManagerMock()
  887. for base_url in test_data.keys():
  888. # instantiate HttpGitClient with mocked pool manager
  889. c = HttpGitClient(base_url, pool_manager=pool_manager, config=None)
  890. # call method that detects url redirection
  891. _, _, processed_url = c._discover_references(b"git-upload-pack", base_url)
  892. # send the same request as the method above without redirection
  893. resp = c.pool_manager.request("GET", base_url + tail, redirect=False)
  894. # check expected behavior of urllib3
  895. redirect_location = resp.get_redirect_location()
  896. if resp.status == 200:
  897. self.assertFalse(redirect_location)
  898. if redirect_location:
  899. # check that url redirection has been correctly detected
  900. self.assertEqual(processed_url, redirect_location[: -len(tail)])
  901. else:
  902. # check also the no redirection case
  903. self.assertEqual(processed_url, base_url)
  904. class TCPGitClientTests(TestCase):
  905. def test_get_url(self):
  906. host = "github.com"
  907. path = "/jelmer/dulwich"
  908. c = TCPGitClient(host)
  909. url = c.get_url(path)
  910. self.assertEqual("git://github.com/jelmer/dulwich", url)
  911. def test_get_url_with_port(self):
  912. host = "github.com"
  913. path = "/jelmer/dulwich"
  914. port = 9090
  915. c = TCPGitClient(host, port=port)
  916. url = c.get_url(path)
  917. self.assertEqual("git://github.com:9090/jelmer/dulwich", url)
  918. class DefaultUrllib3ManagerTest(TestCase):
  919. def test_no_config(self):
  920. manager = default_urllib3_manager(config=None)
  921. self.assertEqual(manager.connection_pool_kw["cert_reqs"], "CERT_REQUIRED")
  922. def test_config_no_proxy(self):
  923. import urllib3
  924. manager = default_urllib3_manager(config=ConfigDict())
  925. self.assertNotIsInstance(manager, urllib3.ProxyManager)
  926. self.assertIsInstance(manager, urllib3.PoolManager)
  927. def test_config_no_proxy_custom_cls(self):
  928. import urllib3
  929. class CustomPoolManager(urllib3.PoolManager):
  930. pass
  931. manager = default_urllib3_manager(
  932. config=ConfigDict(), pool_manager_cls=CustomPoolManager
  933. )
  934. self.assertIsInstance(manager, CustomPoolManager)
  935. def test_config_ssl(self):
  936. config = ConfigDict()
  937. config.set(b"http", b"sslVerify", b"true")
  938. manager = default_urllib3_manager(config=config)
  939. self.assertEqual(manager.connection_pool_kw["cert_reqs"], "CERT_REQUIRED")
  940. def test_config_no_ssl(self):
  941. config = ConfigDict()
  942. config.set(b"http", b"sslVerify", b"false")
  943. manager = default_urllib3_manager(config=config)
  944. self.assertEqual(manager.connection_pool_kw["cert_reqs"], "CERT_NONE")
  945. def test_config_proxy(self):
  946. import urllib3
  947. config = ConfigDict()
  948. config.set(b"http", b"proxy", b"http://localhost:3128/")
  949. manager = default_urllib3_manager(config=config)
  950. self.assertIsInstance(manager, urllib3.ProxyManager)
  951. self.assertTrue(hasattr(manager, "proxy"))
  952. self.assertEqual(manager.proxy.scheme, "http")
  953. self.assertEqual(manager.proxy.host, "localhost")
  954. self.assertEqual(manager.proxy.port, 3128)
  955. def test_environment_proxy(self):
  956. import urllib3
  957. config = ConfigDict()
  958. os.environ["http_proxy"] = "http://myproxy:8080"
  959. manager = default_urllib3_manager(config=config)
  960. self.assertIsInstance(manager, urllib3.ProxyManager)
  961. self.assertTrue(hasattr(manager, "proxy"))
  962. self.assertEqual(manager.proxy.scheme, "http")
  963. self.assertEqual(manager.proxy.host, "myproxy")
  964. self.assertEqual(manager.proxy.port, 8080)
  965. del os.environ["http_proxy"]
  966. def test_config_proxy_custom_cls(self):
  967. import urllib3
  968. class CustomProxyManager(urllib3.ProxyManager):
  969. pass
  970. config = ConfigDict()
  971. config.set(b"http", b"proxy", b"http://localhost:3128/")
  972. manager = default_urllib3_manager(
  973. config=config, proxy_manager_cls=CustomProxyManager
  974. )
  975. self.assertIsInstance(manager, CustomProxyManager)
  976. def test_config_no_verify_ssl(self):
  977. manager = default_urllib3_manager(config=None, cert_reqs="CERT_NONE")
  978. self.assertEqual(manager.connection_pool_kw["cert_reqs"], "CERT_NONE")
  979. class SubprocessSSHVendorTests(TestCase):
  980. def setUp(self):
  981. # Monkey Patch client subprocess popen
  982. self._orig_popen = dulwich.client.subprocess.Popen
  983. dulwich.client.subprocess.Popen = DummyPopen
  984. def tearDown(self):
  985. dulwich.client.subprocess.Popen = self._orig_popen
  986. def test_run_command_dashes(self):
  987. vendor = SubprocessSSHVendor()
  988. self.assertRaises(
  989. StrangeHostname,
  990. vendor.run_command,
  991. "--weird-host",
  992. "git-clone-url",
  993. )
  994. def test_run_command_password(self):
  995. vendor = SubprocessSSHVendor()
  996. self.assertRaises(
  997. NotImplementedError,
  998. vendor.run_command,
  999. "host",
  1000. "git-clone-url",
  1001. password="12345",
  1002. )
  1003. def test_run_command_password_and_privkey(self):
  1004. vendor = SubprocessSSHVendor()
  1005. self.assertRaises(
  1006. NotImplementedError,
  1007. vendor.run_command,
  1008. "host",
  1009. "git-clone-url",
  1010. password="12345",
  1011. key_filename="/tmp/id_rsa",
  1012. )
  1013. def test_run_command_with_port_username_and_privkey(self):
  1014. expected = [
  1015. "ssh",
  1016. "-x",
  1017. "-p",
  1018. "2200",
  1019. "-i",
  1020. "/tmp/id_rsa",
  1021. "user@host",
  1022. "git-clone-url",
  1023. ]
  1024. vendor = SubprocessSSHVendor()
  1025. command = vendor.run_command(
  1026. "host",
  1027. "git-clone-url",
  1028. username="user",
  1029. port="2200",
  1030. key_filename="/tmp/id_rsa",
  1031. )
  1032. args = command.proc.args
  1033. self.assertListEqual(expected, args[0])
  1034. class PLinkSSHVendorTests(TestCase):
  1035. def setUp(self):
  1036. # Monkey Patch client subprocess popen
  1037. self._orig_popen = dulwich.client.subprocess.Popen
  1038. dulwich.client.subprocess.Popen = DummyPopen
  1039. def tearDown(self):
  1040. dulwich.client.subprocess.Popen = self._orig_popen
  1041. def test_run_command_dashes(self):
  1042. vendor = PLinkSSHVendor()
  1043. self.assertRaises(
  1044. StrangeHostname,
  1045. vendor.run_command,
  1046. "--weird-host",
  1047. "git-clone-url",
  1048. )
  1049. def test_run_command_password_and_privkey(self):
  1050. vendor = PLinkSSHVendor()
  1051. warnings.simplefilter("always", UserWarning)
  1052. self.addCleanup(warnings.resetwarnings)
  1053. warnings_list, restore_warnings = setup_warning_catcher()
  1054. self.addCleanup(restore_warnings)
  1055. command = vendor.run_command(
  1056. "host",
  1057. "git-clone-url",
  1058. password="12345",
  1059. key_filename="/tmp/id_rsa",
  1060. )
  1061. expected_warning = UserWarning(
  1062. "Invoking PLink with a password exposes the password in the "
  1063. "process list."
  1064. )
  1065. for w in warnings_list:
  1066. if type(w) == type(expected_warning) and w.args == expected_warning.args:
  1067. break
  1068. else:
  1069. raise AssertionError(
  1070. "Expected warning %r not in %r" % (expected_warning, warnings_list)
  1071. )
  1072. args = command.proc.args
  1073. if sys.platform == "win32":
  1074. binary = ["plink.exe", "-ssh"]
  1075. else:
  1076. binary = ["plink", "-ssh"]
  1077. expected = binary + [
  1078. "-pw",
  1079. "12345",
  1080. "-i",
  1081. "/tmp/id_rsa",
  1082. "host",
  1083. "git-clone-url",
  1084. ]
  1085. self.assertListEqual(expected, args[0])
  1086. def test_run_command_password(self):
  1087. if sys.platform == "win32":
  1088. binary = ["plink.exe", "-ssh"]
  1089. else:
  1090. binary = ["plink", "-ssh"]
  1091. expected = binary + ["-pw", "12345", "host", "git-clone-url"]
  1092. vendor = PLinkSSHVendor()
  1093. warnings.simplefilter("always", UserWarning)
  1094. self.addCleanup(warnings.resetwarnings)
  1095. warnings_list, restore_warnings = setup_warning_catcher()
  1096. self.addCleanup(restore_warnings)
  1097. command = vendor.run_command("host", "git-clone-url", password="12345")
  1098. expected_warning = UserWarning(
  1099. "Invoking PLink with a password exposes the password in the "
  1100. "process list."
  1101. )
  1102. for w in warnings_list:
  1103. if type(w) == type(expected_warning) and w.args == expected_warning.args:
  1104. break
  1105. else:
  1106. raise AssertionError(
  1107. "Expected warning %r not in %r" % (expected_warning, warnings_list)
  1108. )
  1109. args = command.proc.args
  1110. self.assertListEqual(expected, args[0])
  1111. def test_run_command_with_port_username_and_privkey(self):
  1112. if sys.platform == "win32":
  1113. binary = ["plink.exe", "-ssh"]
  1114. else:
  1115. binary = ["plink", "-ssh"]
  1116. expected = binary + [
  1117. "-P",
  1118. "2200",
  1119. "-i",
  1120. "/tmp/id_rsa",
  1121. "user@host",
  1122. "git-clone-url",
  1123. ]
  1124. vendor = PLinkSSHVendor()
  1125. command = vendor.run_command(
  1126. "host",
  1127. "git-clone-url",
  1128. username="user",
  1129. port="2200",
  1130. key_filename="/tmp/id_rsa",
  1131. )
  1132. args = command.proc.args
  1133. self.assertListEqual(expected, args[0])
  1134. class RsyncUrlTests(TestCase):
  1135. def test_simple(self):
  1136. self.assertEqual(parse_rsync_url("foo:bar/path"), (None, "foo", "bar/path"))
  1137. self.assertEqual(
  1138. parse_rsync_url("user@foo:bar/path"), ("user", "foo", "bar/path")
  1139. )
  1140. def test_path(self):
  1141. self.assertRaises(ValueError, parse_rsync_url, "/path")
  1142. class CheckWantsTests(TestCase):
  1143. def test_fine(self):
  1144. check_wants(
  1145. [b"2f3dc7a53fb752a6961d3a56683df46d4d3bf262"],
  1146. {b"refs/heads/blah": b"2f3dc7a53fb752a6961d3a56683df46d4d3bf262"},
  1147. )
  1148. def test_missing(self):
  1149. self.assertRaises(
  1150. InvalidWants,
  1151. check_wants,
  1152. [b"2f3dc7a53fb752a6961d3a56683df46d4d3bf262"],
  1153. {b"refs/heads/blah": b"3f3dc7a53fb752a6961d3a56683df46d4d3bf262"},
  1154. )
  1155. def test_annotated(self):
  1156. self.assertRaises(
  1157. InvalidWants,
  1158. check_wants,
  1159. [b"2f3dc7a53fb752a6961d3a56683df46d4d3bf262"],
  1160. {
  1161. b"refs/heads/blah": b"3f3dc7a53fb752a6961d3a56683df46d4d3bf262",
  1162. b"refs/heads/blah^{}": b"2f3dc7a53fb752a6961d3a56683df46d4d3bf262",
  1163. },
  1164. )
  1165. class FetchPackResultTests(TestCase):
  1166. def test_eq(self):
  1167. self.assertEqual(
  1168. FetchPackResult(
  1169. {b"refs/heads/master": b"2f3dc7a53fb752a6961d3a56683df46d4d3bf262"},
  1170. {},
  1171. b"user/agent",
  1172. ),
  1173. FetchPackResult(
  1174. {b"refs/heads/master": b"2f3dc7a53fb752a6961d3a56683df46d4d3bf262"},
  1175. {},
  1176. b"user/agent",
  1177. ),
  1178. )
  1179. class GitCredentialStoreTests(TestCase):
  1180. @classmethod
  1181. def setUpClass(cls):
  1182. with tempfile.NamedTemporaryFile(delete=False) as f:
  1183. f.write(b"https://user:pass@example.org\n")
  1184. cls.fname = f.name
  1185. @classmethod
  1186. def tearDownClass(cls):
  1187. os.unlink(cls.fname)
  1188. def test_nonmatching_scheme(self):
  1189. self.assertEqual(
  1190. get_credentials_from_store(b"http", b"example.org", fnames=[self.fname]),
  1191. None,
  1192. )
  1193. def test_nonmatching_hostname(self):
  1194. self.assertEqual(
  1195. get_credentials_from_store(b"https", b"noentry.org", fnames=[self.fname]),
  1196. None,
  1197. )
  1198. def test_match_without_username(self):
  1199. self.assertEqual(
  1200. get_credentials_from_store(b"https", b"example.org", fnames=[self.fname]),
  1201. (b"user", b"pass"),
  1202. )
  1203. def test_match_with_matching_username(self):
  1204. self.assertEqual(
  1205. get_credentials_from_store(
  1206. b"https", b"example.org", b"user", fnames=[self.fname]
  1207. ),
  1208. (b"user", b"pass"),
  1209. )
  1210. def test_no_match_with_nonmatching_username(self):
  1211. self.assertEqual(
  1212. get_credentials_from_store(
  1213. b"https", b"example.org", b"otheruser", fnames=[self.fname]
  1214. ),
  1215. None,
  1216. )
  1217. class RemoteErrorFromStderrTests(TestCase):
  1218. def test_nothing(self):
  1219. self.assertEqual(_remote_error_from_stderr(None), HangupException())
  1220. def test_error_line(self):
  1221. b = BytesIO(
  1222. b"""\
  1223. This is some random output.
  1224. ERROR: This is the actual error
  1225. with a tail
  1226. """
  1227. )
  1228. self.assertEqual(
  1229. _remote_error_from_stderr(b),
  1230. GitProtocolError("This is the actual error"),
  1231. )
  1232. def test_no_error_line(self):
  1233. b = BytesIO(
  1234. b"""\
  1235. This is output without an error line.
  1236. And this line is just random noise, too.
  1237. """
  1238. )
  1239. self.assertEqual(
  1240. _remote_error_from_stderr(b),
  1241. HangupException(
  1242. [
  1243. b"This is output without an error line.",
  1244. b"And this line is just random noise, too.",
  1245. ]
  1246. ),
  1247. )