1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
# test_aiohttp.py -- Tests for the aiohttp HTTP server
# Copyright (C) 2025 Jelmer Vernooij <jelmer@jelmer.uk>
#
# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
# General Public License as published by the Free Software Foundation; version 2.0
# or (at your option) any later version. You can redistribute it and/or
# modify it under the terms of either of these two licenses.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# You should have received a copy of the licenses; if not, see
# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
# License, Version 2.0.
#
"""Tests for the aiohttp Git HTTP server."""
try:
from aiohttp.test_utils import AioHTTPTestCase
aiohttp_missing = False
except ImportError:
from unittest import TestCase as AioHTTPTestCase # type: ignore
aiohttp_missing = True
from dulwich.objects import Blob
from dulwich.repo import MemoryRepo
from dulwich.tests.utils import make_object
from . import skipIf
if not aiohttp_missing:
from dulwich.aiohttp.server import create_repo_app
@skipIf(aiohttp_missing, "aiohttp not available")
class AiohttpAppTestCase(AioHTTPTestCase): # type: ignore
"""Test the aiohttp application."""
def setUp(self):
super().setUp()
self.repo = MemoryRepo.init_bare([], {})
self.blob = make_object(Blob, data=b"blob contents")
self.repo.object_store.add_object(self.blob)
self.repo.refs[b"refs/heads/master"] = self.blob.id
def get_app(self):
return create_repo_app(self.repo)
async def test_get_info_refs_dumb(self):
"""Test GET /info/refs without service parameter (dumb protocol)."""
resp = await self.client.request("GET", "/info/refs")
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content_type, "text/plain")
text = await resp.text()
self.assertIn(self.blob.id.decode("ascii"), text)
self.assertIn("refs/heads/master", text)
async def test_get_info_refs_smart(self):
"""Test GET /info/refs?service=git-upload-pack (smart protocol)."""
resp = await self.client.request("GET", "/info/refs?service=git-upload-pack")
self.assertEqual(resp.status, 200)
self.assertIn("git-upload-pack", resp.content_type)
async def test_post_upload_pack(self):
"""Test POST /git-upload-pack."""
# Simple test that the endpoint exists and accepts POST
resp = await self.client.request(
"POST",
"/git-upload-pack",
data=b"0000",
headers={"Content-Type": "application/x-git-upload-pack-request"},
)
# Should respond with 200 even for invalid/minimal input
self.assertEqual(resp.status, 200)
self.assertIn("git-upload-pack", resp.content_type)
@skipIf(aiohttp_missing, "aiohttp not available")
class AiohttpDumbAppTestCase(AioHTTPTestCase): # type: ignore
"""Test the aiohttp application with dumb protocol."""
def setUp(self):
super().setUp()
self.repo = MemoryRepo.init_bare([], {})
self.repo._put_named_file("HEAD", b"ref: refs/heads/master\n")
self.blob = make_object(Blob, data=b"blob contents")
self.repo.object_store.add_object(self.blob)
self.repo.refs[b"refs/heads/master"] = self.blob.id
def get_app(self):
return create_repo_app(self.repo, dumb=True)
async def test_get_head(self):
"""Test GET /HEAD."""
resp = await self.client.request("GET", "/HEAD")
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content_type, "text/plain")
text = await resp.text()
self.assertEqual(text, "ref: refs/heads/master\n")
async def test_get_info_packs(self):
"""Test GET /objects/info/packs."""
resp = await self.client.request("GET", "/objects/info/packs")
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content_type, "text/plain")
async def test_get_loose_object(self):
"""Test GET /objects/{dir}/{file} for loose objects."""
sha = self.blob.id.decode("ascii")
dir_part = sha[:2]
file_part = sha[2:]
resp = await self.client.request("GET", f"/objects/{dir_part}/{file_part}")
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content_type, "application/x-git-loose-object")
body = await resp.read()
self.assertEqual(body, self.blob.as_legacy_object())
async def test_get_loose_object_not_found(self):
"""Test GET /objects/{dir}/{file} for non-existent object."""
resp = await self.client.request("GET", "/objects/ab/cdef" + "0" * 36)
self.assertEqual(resp.status, 404)
|
