83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
"""Isolated browser-test server: all VM and terminal operations are fakes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import uvicorn
|
|
|
|
from nest.server import create_app
|
|
from tests.test_server import FakeLifecycle, fake_application
|
|
from tests.test_sessions import FakeTerminal
|
|
from uvm.errors import ConflictError, NotFoundError
|
|
from uvm.lifecycle import ListedVm
|
|
|
|
|
|
class BrowserTerminal(FakeTerminal):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.command = bytearray()
|
|
|
|
async def write(self, data):
|
|
await self.events.put({"type": "output", "data": data.replace(b"\r", b"\r\n")})
|
|
for byte in data:
|
|
if byte == 13:
|
|
await self.events.put({"type": "output", "data": b"result:" + bytes(self.command) + b"\r\nguest$ "})
|
|
self.command.clear()
|
|
elif byte == 3:
|
|
self.command.clear()
|
|
else:
|
|
self.command.append(byte)
|
|
|
|
|
|
class BrowserLifecycle(FakeLifecycle):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.vm.username = "worker"
|
|
self.records = {self.vm.id: self.vm, "vm-other": replace(self.vm, id="vm-other", name="second-vm", status="stopped")}
|
|
|
|
def get(self, vm_id):
|
|
try:
|
|
return self.records[vm_id]
|
|
except KeyError as error:
|
|
raise NotFoundError("VM not found") from error
|
|
|
|
def inspect(self, identifier):
|
|
for vm in self.records.values():
|
|
if identifier in {vm.id, vm.name, vm.guest_ip}:
|
|
return ListedVm(vm, vm.status, "enforced" if vm.status == "running" else "inactive")
|
|
raise NotFoundError("VM not found")
|
|
|
|
def list_vms(self):
|
|
return [self.inspect(vm_id) for vm_id in self.records]
|
|
|
|
def open_terminal(self, vm_id, **_size):
|
|
if self.get(vm_id).status != "running":
|
|
raise ConflictError("Start the machine first")
|
|
return BrowserTerminal(), "worker"
|
|
|
|
def install_guest_agent(self, vm_id):
|
|
return self.get(vm_id)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--base", type=Path, required=True)
|
|
parser.add_argument("--port", type=int, required=True)
|
|
args = parser.parse_args()
|
|
application = fake_application(api_token="browser-test-token")
|
|
application.settings.base = args.base
|
|
application.lifecycle = BrowserLifecycle()
|
|
application.state_store = SimpleNamespace(get=application.lifecycle.get)
|
|
api = create_app(application, host="127.0.0.1")
|
|
server = uvicorn.Server(uvicorn.Config(api, host="127.0.0.1", port=args.port, log_level="warning"))
|
|
api.state.uvicorn_server = server
|
|
server.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|