124 lines
6.3 KiB
Python
124 lines
6.3 KiB
Python
"""Scoped Gitea release upload/readback and subsequent publication."""
|
|
import argparse
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tomllib
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
HERE = Path(__file__).resolve().parent
|
|
ORIGIN = 'http://217.177.34.108:3000'
|
|
REPO = ORIGIN + '/api/v1/repos/sorokin.oleg.edu/ARCADIA-Studio'
|
|
BASE = REPO + '/releases/3'
|
|
TAG = 'v0.2.0-preview.contextual.20260911'
|
|
TARGET = 'b2c3e54001fc19a05423bcd5808ceea78d13ff59'
|
|
NAME = 'ArcadiaStudio-0.2.0-preview.contextual.20260911-x86_64-pc-windows-gnullvm.zip'
|
|
FOLDER = ROOT / '.local/contextual-prerelease-20260911/packages'
|
|
EXPECTED = {
|
|
NAME: '9186d4fc9a7b52fb195c183c4847c8e3419ea6da8b3f907e0c06f9ceec58dee0',
|
|
NAME + '.sha256': '21455af55f6330235a72f43f09bd155acaea0d647bd3f68f09a64d06a21fdbcc',
|
|
'VALIDATION.md': '104aa66522a4f1a14972d598ed494750d4b6408445c42a57ff771572b9690474',
|
|
}
|
|
NOTES_SHA = '5f023207d275533a1f2004294aafee4e04c6bf147d34e9d8c3a0889a8b09faac'
|
|
token = tomllib.loads((Path.home() / '.codex/config.toml').read_text(encoding='utf-8'))['mcp_servers']['gitea']['env']['GITEA_ACCESS_TOKEN']
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
raise RuntimeError('Unexpected redirect; authorization was not forwarded')
|
|
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect)
|
|
|
|
def request(url, data=None, method=None, headers=None, binary=False):
|
|
parsed = urllib.parse.urlsplit(url)
|
|
expected_origin = urllib.parse.urlsplit(ORIGIN)
|
|
assert (parsed.scheme, parsed.netloc) == (expected_origin.scheme, expected_origin.netloc)
|
|
assert parsed.username is None and parsed.password is None
|
|
req = urllib.request.Request(url, data=data, method=method,
|
|
headers={'Authorization': 'token ' + token, **(headers or {})})
|
|
try:
|
|
with opener.open(req, timeout=120) as response:
|
|
body = response.read()
|
|
return body if binary else json.loads(body)
|
|
except urllib.error.HTTPError as error:
|
|
detail = error.read(2000).decode('utf-8', 'replace').replace(token, '[redacted]')
|
|
raise SystemExit(f'HTTP {error.code}: {detail}') from None
|
|
|
|
def digest(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
def record(name, value):
|
|
with (HERE / name).open('x', encoding='utf-8', newline='\n') as output:
|
|
json.dump(value, output, indent=2)
|
|
output.write('\n')
|
|
|
|
def check_release(value):
|
|
assert value['id'] == 3 and value['tag_name'] == TAG
|
|
assert value['target_commitish'] == TARGET and value['prerelease'] is True
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('action', choices=('upload', 'publish'))
|
|
args = parser.parse_args()
|
|
release = request(BASE)
|
|
check_release(release)
|
|
assets = request(BASE + '/assets')
|
|
notes_bytes = (HERE / 'RELEASE-NOTES.md').read_bytes()
|
|
assert digest(notes_bytes) == NOTES_SHA
|
|
notes = notes_bytes.decode('utf-8')
|
|
assert 'PACKAGE-RUNTIME-ACCEPTANCE-PENDING' not in notes
|
|
|
|
if args.action == 'upload':
|
|
assert release['draft'] is True
|
|
assert not (HERE / 'assets-verified.json').exists(), 'Completed upload receipt already exists'
|
|
paths = [FOLDER / NAME, FOLDER / (NAME + '.sha256'), HERE / 'VALIDATION.md']
|
|
assert {asset['name'] for asset in assets}.issubset(EXPECTED)
|
|
results = []
|
|
for path in paths:
|
|
content = path.read_bytes()
|
|
assert digest(content) == EXPECTED[path.name], path.name
|
|
matches = [asset for asset in assets if asset['name'] == path.name]
|
|
assert len(matches) <= 1
|
|
if matches:
|
|
asset = matches[0]
|
|
else:
|
|
boundary = 'ArcadiaContextual' + uuid.uuid4().hex
|
|
prefix = (f'--{boundary}\r\nContent-Disposition: form-data; name="attachment"; filename="{path.name}"\r\nContent-Type: application/octet-stream\r\n\r\n').encode()
|
|
body = prefix + content + f'\r\n--{boundary}--\r\n'.encode()
|
|
asset = request(BASE + '/assets?' + urllib.parse.urlencode({'name': path.name}),
|
|
data=body, method='POST', headers={'Content-Type': 'multipart/form-data; boundary=' + boundary})
|
|
received = request(asset['browser_download_url'], binary=True)
|
|
assert received == content and digest(received) == EXPECTED[path.name]
|
|
result = {'id': asset['id'], 'name': path.name, 'bytes': len(content),
|
|
'sha256': digest(content), 'url': asset['browser_download_url'], 'readback': 'PASS'}
|
|
results.append(result)
|
|
print(json.dumps(result), flush=True)
|
|
final_assets = request(BASE + '/assets')
|
|
assert {a['id']: (a['name'], a['size']) for a in final_assets} == {a['id']: (a['name'], a['bytes']) for a in results}
|
|
record('assets-verified.json', {'atUtc': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'releaseId': 3, 'assets': results})
|
|
else:
|
|
assert not (HERE / 'published-verified.json').exists(), 'Completed publication receipt already exists'
|
|
verified = json.loads((HERE / 'assets-verified.json').read_bytes())
|
|
expected_assets = {a['id']: (a['name'], a['bytes']) for a in verified['assets']}
|
|
assert {a['id']: (a['name'], a['size']) for a in assets} == expected_assets
|
|
assert {a['name']: a['sha256'] for a in verified['assets']} == EXPECTED
|
|
request(BASE, data=json.dumps({'draft': False, 'prerelease': True, 'body': notes}).encode(),
|
|
method='PATCH', headers={'Content-Type': 'application/json'})
|
|
readback = request(BASE)
|
|
check_release(readback)
|
|
assert readback['draft'] is False and readback['body'] == notes
|
|
tag = request(REPO + '/tags/' + urllib.parse.quote(TAG, safe=''))
|
|
assert tag['name'] == TAG and tag['commit']['sha'] == TARGET
|
|
final_assets = request(BASE + '/assets')
|
|
assert {a['id']: (a['name'], a['size']) for a in final_assets} == expected_assets
|
|
result = {'atUtc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
'id': readback['id'], 'tag': TAG, 'target': TARGET, 'url': readback['html_url'],
|
|
'draft': readback['draft'], 'prerelease': readback['prerelease'],
|
|
'assetCount': len(final_assets), 'notesSha256': NOTES_SHA,
|
|
'publishedAt': readback['published_at'], 'tagTargetReadback': 'PASS'}
|
|
record('published-verified.json', result)
|
|
print(json.dumps(result, indent=2), flush=True)
|