69 lines
4.1 KiB
Python
69 lines
4.1 KiB
Python
"""Upload/read back only this new scoped draft release; never replace assets."""
|
|
from pathlib import Path
|
|
import hashlib, json, sys, tomllib, urllib.error, urllib.parse, urllib.request, uuid
|
|
root = Path(__file__).resolve().parent
|
|
base = 'http://217.177.34.108:3000'
|
|
api = base + '/api/v1/repos/sorokin.oleg.edu/ARCADIA-Studio'
|
|
tag = 'v0.2.0-preview.save-diagnostics.20261001'
|
|
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, *args, **kwargs):
|
|
raise RuntimeError('Unexpected redirect; credentials are not forwarded')
|
|
opener = urllib.request.build_opener(NoRedirect)
|
|
def request(url, method='GET', data=None, content_type=None):
|
|
assert url.startswith(base+'/')
|
|
headers = {'Authorization': 'token '+token}
|
|
if content_type: headers['Content-Type'] = content_type
|
|
with opener.open(urllib.request.Request(url, data=data, headers=headers, method=method), timeout=60) as response:
|
|
return response.read()
|
|
def get(url): return json.loads(request(url))
|
|
def pin(data): return {'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()}
|
|
mode, release_id, source = sys.argv[1:]
|
|
assert mode in ('upload', 'publish', 'verify')
|
|
release_url = api+'/releases/'+str(int(release_id))
|
|
release = get(release_url)
|
|
assert release['tag_name'] == tag and release['target_commitish'] == source
|
|
assert release['prerelease'] is True
|
|
assets = root/'release-assets-ready01'
|
|
files = sorted(p for p in assets.iterdir() if p.is_file())
|
|
assert len(files) == 3
|
|
records = {}
|
|
if mode == 'upload':
|
|
assert release['draft'] is True, 'Only a fresh draft is uploadable'
|
|
for p in files:
|
|
before = get(release_url+'/assets')
|
|
same = [a for a in before if a['name']==p.name]
|
|
assert len(same) <= 1
|
|
data = p.read_bytes()
|
|
if same:
|
|
assert pin(request(same[0]['browser_download_url'])) == pin(data), 'Existing asset differs; never replace'
|
|
asset = same[0]
|
|
else:
|
|
boundary = 'arcadia-'+uuid.uuid4().hex
|
|
payload = ('--'+boundary+'\r\nContent-Disposition: form-data; name="attachment"; filename="'+p.name+'"\r\nContent-Type: application/octet-stream\r\n\r\n').encode()+data+('\r\n--'+boundary+'--\r\n').encode()
|
|
asset = json.loads(request(release_url+'/assets?name='+urllib.parse.quote(p.name), 'POST', payload, 'multipart/form-data; boundary='+boundary))
|
|
downloaded = request(asset['browser_download_url'])
|
|
assert pin(downloaded) == pin(data), p.name
|
|
records[p.name] = {**pin(data), 'asset_id': asset['id'], 'url': asset['browser_download_url']}
|
|
print(json.dumps({'verified_asset': p.name, **pin(data)}), flush=True)
|
|
else:
|
|
existing = get(release_url+'/assets')
|
|
assert {a['name'] for a in existing} == {p.name for p in files}
|
|
for p in files:
|
|
asset = next(a for a in existing if a['name']==p.name)
|
|
assert pin(request(asset['browser_download_url'])) == pin(p.read_bytes())
|
|
records[p.name] = {**pin(p.read_bytes()), 'asset_id': asset['id'], 'url': asset['browser_download_url']}
|
|
if mode == 'publish':
|
|
assert release['draft'] is True
|
|
body = (root/'release-notes.md').read_text(encoding='utf-8')
|
|
request(release_url, 'PATCH', json.dumps({'draft': False, 'prerelease': True, 'body': body}).encode(), 'application/json')
|
|
release = get(release_url)
|
|
if mode in ('publish', 'verify'):
|
|
assert release['draft'] is False and release['prerelease'] is True
|
|
assert release['body'] == (root/'release-notes.md').read_text(encoding='utf-8')
|
|
assert get(api+'/releases/tags/'+urllib.parse.quote(tag, safe=''))['id'] == release['id']
|
|
receipt = {'mode': mode, 'release_id': release['id'], 'tag': tag, 'source': source, 'url': release['html_url'], 'draft': release['draft'], 'assets': records}
|
|
with (root/(mode+'-readback.json')).open('x', encoding='utf-8') as f:
|
|
json.dump(receipt, f, indent=2)
|
|
print(json.dumps({k:v for k,v in receipt.items() if k!='assets'}, indent=2), flush=True)
|