diff --git a/.flake8 b/.flake8 new file mode 100644 index 00000000..7da1f960 --- /dev/null +++ b/.flake8 @@ -0,0 +1,2 @@ +[flake8] +max-line-length = 100 diff --git a/tests/oci-registry-server.py b/tests/oci-registry-server.py index 33c3b646..3050a883 100755 --- a/tests/oci-registry-server.py +++ b/tests/oci-registry-server.py @@ -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]