check-merged.py

def terminal_websocket(name: str, namespace: str, command: list[str], container: str = None):
    load_config()
    core_v1_api = client.CoreV1Api()
    resp = stream(
        core_v1_api.connect_get_namespaced_pod_exec,
        name=name,
        namespace=namespace,
        container=container,
        command=command,
        stderr=True,
        stdin=True,
        stdout=True,
        tty=False,
        _preload_content=False,
    )

    return resp


@router.websocket("/terminal/ws/{namespace}/{name}")
async def websocket_endpoint(
    websocket: WebSocket,
    namespace: str,
    name: str,
):
    await websocket.accept()
    try:
        resp = terminal_websocket(
            name=name,
            namespace=namespace,
            command=["sh"],
        )
        while resp.is_open():
            resp.update(timeout=1)
            data = await websocket.receive_text()
            message = ""
            if data:
                resp.write_stdin(data)

            if resp.peek_stdout():
                message = resp.read_stdout()
                await websocket.send_text(message)

            if resp.peek_stderr():
                message = resp.read_stderr()
                await websocket.send_text(message)

        resp.write_stdin("date\n")
        resp.write_stdin("whoami\n")
        print(resp)
    except WebSocketDisconnect:
        resp.close()

exec_terminal.js

frappe.pages['exec-terminal'].on_page_load = async wrapper => {
	const page = frappe.ui.make_app_page({
		parent: wrapper,
		title: 'Terminal',
		single_column: true
	});
	await configure(page);
}

const configure = async page => {
	const query = new URL(window.location.href).searchParams;
	const namespace = query.get('namespace');
	const name = query.get('name');
	const command = query.get('command') || 'sh';

	if (!name) {
		throw new Error("name required");
	}

	if (!namespace) {
		throw new Error("namespace required");
	}

	const { message } = await frappe.call({ method: "k8s_bench_interface.endpoints.get_pod_exec_info" });
	$(`<div id="terminal"></div>`).appendTo(page.main);
	const wsUrl = message.terminal_socket_endpoint // hardcode this to ws://localhost:3000/terminal
		+ `/ws/${namespace}/${name}?command=${command}&token=`
		+ message.token;

	const term = new Terminal({ cursorBlink: true, convertEol: true});
	const socket = new WebSocket(wsUrl);
	const attachAddon = new AttachAddon(socket);
	term.loadAddon(attachAddon);
	const fitAddon = new FitAddon();
	term.loadAddon(fitAddon);
	fitAddon.fit();
	term.reset();
	term.open(document.getElementById('terminal'));

	socket.onopen = () => socket.send("hostname\r");
	socket.onclose = event => {
		term.writeln("");
		term.writeln("  \u001b[31m[!] Lost connection");
	}

	document.addEventListener("keydown", (zEvent) => {
		if (zEvent.ctrlKey && zEvent.shiftKey && zEvent.key === "C") {
		  zEvent.preventDefault();
		  copyToClipBoard(term.getSelection());
		}
	});

	window.onresize = () => {
		fitAddon.fit();
	};
}