run ./cloudflare-upload-preview.py followed by ./cloudflare-test-preview.py

cloudflare-test-preview.py

#!/usr/bin/env python3.6
import asyncio
import json
import platform
import secrets
import subprocess
import shlex
import sys
from pathlib import Path

import aiohttp
from aiohttp import WSMsgType
from devtools import debug, pformat
from devtools.ansi import sformat

THIS_DIR = Path(__file__).parent
fake_host = 'example.com'
preview_id = (THIS_DIR / '.preview_id').read_text()


def get_browser_opener():
    system = platform.system()
    if system == 'Linux':
        # what about linux without gnome?
        # on some version this should be `gnome-open`
        return 'gvfs-open'
    elif system == 'Windows':
        return 'start chrome'  # random guess, I don't much care
    else:
        return 'open'  # osx at least


def open_browser(path):
    path = path or '/theendpoint/whatever.txt'
    page_url = f'https://cloudflareworkers.com/#{preview_id}:https://{fake_host}{path}'
    subprocess.run(shlex.split(f'{get_browser_opener()} {page_url}'), check=True)


# we don't need all of these, but not clear which we do
msgs = [
    {'id': 1, 'method': 'Profiler.enable'},
    {'id': 2, 'method': 'Runtime.enable'},
    {'id': 3, 'method': 'Debugger.enable'},
    {'id': 4, 'method': 'Debugger.setPauseOnExceptions', 'params': {'state': 'none'}},
    {'id': 5, 'method': 'Debugger.setAsyncCallStackDepth', 'params': {'maxDepth': 32}},
    {'id': 6, 'method': 'Debugger.setBlackboxPatterns', 'params': {'patterns': []}},
    {'id': 7, 'method': 'Runtime.runIfWaitingForDebugger'},
]

ignored_methods = {
    'Runtime.executionContextCreated',
    'Runtime.executionContextDestroyed',
    'Debugger.scriptParsed',
}


async def watch_inspect(session_id):
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(f'wss://cloudflareworkers.com/inspect/{session_id}') as ws:
            for msg in msgs:
                await ws.send_str(json.dumps(msg))
            async for msg in ws:
                if msg.type == WSMsgType.TEXT:
                    data = json.loads(msg.data)
                    method = data.get('method')
                    if not method or method in ignored_methods:
                        continue
                    if method == 'Runtime.consoleAPICalled':
                        args = data['params']['args']
                        level = data['params']['type']
                        s = [sformat(f'{level:>8}:', sformat.blue)]
                        for arg in args:
                            if arg['type'] in {'string', 'number'}:
                                s.append(sformat(arg['value'], sformat.red))
                            else:
                                s.append(str(pformat(arg['preview'], highlight=True)))
                        print(' '.join(s))
                    else:
                        print('unknown message from websocket:')
                        debug(data)


async def run_httpie_websocket(loop, session_id, httpie_args):
    loop.create_task(watch_inspect(session_id))
    await asyncio.sleep(0.2)
    p = await asyncio.create_subprocess_exec(*httpie_args)
    await p.wait()
    # wait longer in case the worker is still running
    await asyncio.sleep(2)
    return p.returncode


def run_httpie():
    args = sys.argv[1:]
    added_url = False
    session_id = secrets.token_hex()[:32]
    httpie_args = ['http']
    for arg in args:
        if arg.startswith('/') and not added_url:
            httpie_args += [
                'https://0000000000000000.cloudflareworkers.com',
                f'Cookie:__ew_fiddle_preview={preview_id}{session_id}{1}{fake_host}{arg}'
            ]
            added_url = True
        else:
            httpie_args.append(arg)

    if not added_url:
        print('path argument not found, should start with "/"')
        return 1
    else:

        loop = asyncio.get_event_loop()
        return loop.run_until_complete(run_httpie_websocket(loop, session_id, httpie_args))


if __name__ == '__main__':
    if len(sys.argv) > 1 and sys.argv[1] == 'browser':
        open_browser(None if len(sys.argv) == 2 else sys.argv[2])
    else:
        sys.exit(run_httpie())

cloudflare-upload-preview.py

#!/usr/bin/env python3.6
import subprocess
from pathlib import Path

import requests

THIS_DIR = Path(__file__).parent

subprocess.run(('yarn', 'build'), check=True)

content = Path('dist/worker.js').read_bytes()
r = requests.post('https://cloudflareworkers.com/script', data=content)
assert r.status_code == 201, (r.status_code, r.text)

preview_id = r.json()['id']

(THIS_DIR / '.preview_id').write_text(preview_id)

print("""
Upload successful.

To test in a browser, run

    ./cloudflare-test-preview.py browser /theendpoint/whatever.txt


to test in the terminal:

    ./cloudflare-test-preview.py POST /theendpoint/whatever.txt
    
Arguments
""")