ahill/flatpak
1
0
Fork 0
mirror of https://github.com/flatpak/flatpak.git synced 2026-01-26 06:07:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

tests/oci-registry-server.py: Clean up Python style

Browse Source
This commit is contained in:
Sebastian Wick 2024-08-22 02:43:13 -04:00 committed by Georges Basile Stavracas Neto
parent 05e2b083ed
commit 5945a18d97
2 changed files with 88 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.flake8 Normal file
View File

@ -0,0 +1,2 @@
[flake8]
max-line-length = 100

170
tests/oci-registry-server.py
View File

@ -13,65 +13,61 @@ import http.server as http_server
repositories = {}
icons = {}
def get_index():
results = []
for repo_name in sorted(repositories.keys()):
repo = repositories[repo_name]
results.append({
'Name': repo_name,
'Images': repo['images'],
'Lists': [],
})
results.append(
{
"Name": repo_name,
"Images": repo["images"],
"Lists": [],
}
)
return json.dumps({"Registry": "/", "Results": results}, indent=4)
return json.dumps({
'Registry': '/',
'Results': results
}, indent=4)
def cache_icon(data_uri):
prefix = 'data:image/png;base64,'
prefix = "data:image/png;base64,"
assert data_uri.startswith(prefix)
data = base64.b64decode(data_uri[len(prefix):])
data = base64.b64decode(data_uri[len(prefix) :])
h = hashlib.sha256()
h.update(data)
digest = h.hexdigest()
filename = digest + '.png'
filename = digest + ".png"
icons[filename] = data
return '/icons/' + filename
return "/icons/" + filename
serial = 0
server_start_time = int(time.time())
def get_etag():
return str(server_start_time) + '-' + str(serial)
return str(server_start_time) + "-" + str(serial)
def modified():
global serial
serial += 1
def parse_http_date(date):
parsed = parsedate(date)
if parsed is not None:
return timegm(parsed)
else:
return None
class RequestHandler(http_server.BaseHTTPRequestHandler):
def check_route(self, route):
parts = self.path.split('?', 1)
path = parts[0].split('/')
parts = self.path.split("?", 1)
path = parts[0].split("/")
result = []
route_path = route.split('/')
route_path = route.split("/")
print((route_path, path))
if len(route_path) != len(path):
return False
matches = {}
for i in range(1, len(route_path)):
if route_path[i][0] == '@':
if route_path[i][0] == "@":
matches[route_path[i][1:]] = path[i]
elif route_path[i] != path[i]:
return False
@ -92,24 +88,25 @@ class RequestHandler(http_server.BaseHTTPRequestHandler):
add_headers = {}
if self.check_route('/v2/@repo_name/blobs/@digest'):
repo_name = self.matches['repo_name']
digest = self.matches['digest']
response_file = repositories[repo_name]['blobs'][digest]
elif self.check_route('/v2/@repo_name/manifests/@ref'):
repo_name = self.matches['repo_name']
ref = self.matches['ref']
response_file = repositories[repo_name]['manifests'][ref]
elif self.check_route('/index/static') or self.check_route('/index/dynamic'):
if self.check_route("/v2/@repo_name/blobs/@digest"):
repo_name = self.matches["repo_name"]
digest = self.matches["digest"]
response_file = repositories[repo_name]["blobs"][digest]
elif self.check_route("/v2/@repo_name/manifests/@ref"):
repo_name = self.matches["repo_name"]
ref = self.matches["ref"]
response_file = repositories[repo_name]["manifests"][ref]
elif self.check_route("/index/static") or self.check_route("/index/dynamic"):
etag = get_etag()
if self.headers.get("If-None-Match") == etag:
response = 304
else:
response_string = get_index()
add_headers['Etag'] = etag
elif self.check_route('/icons/@filename') :
response_string = icons[self.matches['filename']]
response_content_type = 'image/png'
add_headers["Etag"] = etag
elif self.check_route("/icons/@filename"):
response_string = icons[self.matches["filename"]]
assert isinstance(response_string, bytes)
response_content_type = "image/png"
else:
response = 404
@ -121,86 +118,89 @@ class RequestHandler(http_server.BaseHTTPRequestHandler):
self.send_header("Content-Type", response_content_type)
if response == 200 or response == 304:
self.send_header('Cache-Control', 'no-cache')
self.send_header("Cache-Control", "no-cache")
self.end_headers()
if response == 200:
if response_file:
with open(response_file, 'rb') as f:
with open(response_file, "rb") as f:
response_string = f.read()
if isinstance(response_string, bytes):
self.wfile.write(response_string)
else:
self.wfile.write(response_string.encode('utf-8'))
assert isinstance(response_string, str)
self.wfile.write(response_string.encode("utf-8"))
def do_HEAD(self):
return self.do_GET()
def do_POST(self):
if self.check_route('/testing/@repo_name/@tag'):
repo_name = self.matches['repo_name']
tag = self.matches['tag']
d = self.query['d'][0]
detach_icons = 'detach-icons' in self.query
if self.check_route("/testing/@repo_name/@tag"):
repo_name = self.matches["repo_name"]
tag = self.matches["tag"]
d = self.query["d"][0]
detach_icons = "detach-icons" in self.query
repo = repositories.setdefault(repo_name, {})
blobs = repo.setdefault('blobs', {})
manifests = repo.setdefault('manifests', {})
images = repo.setdefault('images', [])
blobs = repo.setdefault("blobs", {})
manifests = repo.setdefault("manifests", {})
images = repo.setdefault("images", [])
with open(os.path.join(d, 'index.json')) as f:
with open(os.path.join(d, "index.json")) as f:
index = json.load(f)
manifest_digest = index['manifests'][0]['digest']
manifest_path = os.path.join(d, 'blobs', *manifest_digest.split(':'))
manifest_digest = index["manifests"][0]["digest"]
manifest_path = os.path.join(d, "blobs", *manifest_digest.split(":"))
manifests[manifest_digest] = manifest_path
manifests[tag] = manifest_path
with open(manifest_path) as f:
manifest = json.load(f)
config_digest = manifest['config']['digest']
config_path = os.path.join(d, 'blobs', *config_digest.split(':'))
config_digest = manifest["config"]["digest"]
config_path = os.path.join(d, "blobs", *config_digest.split(":"))
with open(config_path) as f:
config = json.load(f)
for dig in os.listdir(os.path.join(d, 'blobs', 'sha256')):
digest = 'sha256:' + dig
path = os.path.join(d, 'blobs', 'sha256', dig)
for dig in os.listdir(os.path.join(d, "blobs", "sha256")):
digest = "sha256:" + dig
path = os.path.join(d, "blobs", "sha256", dig)
if digest != manifest_digest:
blobs[digest] = path
if detach_icons:
for size in (64, 128):
annotation = 'org.freedesktop.appstream.icon-{}'.format(size)
icon = manifest.get('annotations', {}).get(annotation)
annotation = "org.freedesktop.appstream.icon-{}".format(size)
icon = manifest.get("annotations", {}).get(annotation)
if icon:
path = cache_icon(icon)
manifest['annotations'][annotation] = path
manifest["annotations"][annotation] = path
else:
icon = config.get('config', {}).get('Labels', {}).get(annotation)
icon = (
config.get("config", {}).get("Labels", {}).get(annotation)
)
if icon:
path = cache_icon(icon)
config['config']['Labels'][annotation] = path
config["config"]["Labels"][annotation] = path
image = {
"Tags": [tag],
"Digest": manifest_digest,
"MediaType": "application/vnd.oci.image.manifest.v1+json",
"OS": config['os'],
"Architecture": config['architecture'],
"Annotations": manifest.get('annotations', {}),
"Labels": config.get('config', {}).get('Labels', {}),
"OS": config["os"],
"Architecture": config["architecture"],
"Annotations": manifest.get("annotations", {}),
"Labels": config.get("config", {}).get("Labels", {}),
}
# Delete old versions
for i in images:
if tag in i['Tags']:
if tag in i["Tags"]:
images.remove(i)
del manifests[i['Digest']]
del manifests[i["Digest"]]
images.append(image)
@ -214,26 +214,26 @@ class RequestHandler(http_server.BaseHTTPRequestHandler):
return
def do_DELETE(self):
if self.check_route('/testing/@repo_name/@ref'):
repo_name = self.matches['repo_name']
ref = self.matches['ref']
if self.check_route("/testing/@repo_name/@ref"):
repo_name = self.matches["repo_name"]
ref = self.matches["ref"]
repo = repositories.setdefault(repo_name, {})
blobs = repo.setdefault('blobs', {})
manifests = repo.setdefault('manifests', {})
images = repo.setdefault('images', [])
repo.setdefault("blobs", {})
manifests = repo.setdefault("manifests", {})
images = repo.setdefault("images", [])
image = None
for i in images:
if i['Digest'] == ref or ref in i['Tags']:
if i["Digest"] == ref or ref in i["Tags"]:
image = i
break
assert image
images.remove(image)
del manifests[image['Digest']]
for t in image['Tags']:
del manifests[image["Digest"]]
for t in image["Tags"]:
del manifests[t]
modified()
@ -245,22 +245,24 @@ class RequestHandler(http_server.BaseHTTPRequestHandler):
self.end_headers()
return
def run(dir):
RequestHandler.protocol_version = "HTTP/1.0"
httpd = http_server.HTTPServer( ("127.0.0.1", 0), RequestHandler)
httpd = http_server.HTTPServer(("127.0.0.1", 0), RequestHandler)
host, port = httpd.socket.getsockname()[:2]
with open("httpd-port", 'w') as file:
with open("httpd-port", "w") as file:
file.write("%d" % port)
try:
os.write(3, bytes("Started\n", 'utf-8'));
except:
os.write(3, bytes("Started\n", "utf-8"))
except OSError:
pass
print("Serving HTTP on port %d" % port);
print("Serving HTTP on port %d" % port)
if dir:
os.chdir(dir)
httpd.serve_forever()
if __name__ == '__main__':
if __name__ == "__main__":
dir = None
if len(sys.argv) >= 2 and len(sys.argv[1]) > 0:
dir = sys.argv[1]