mirrored a minute ago
0
HappySixFix thread lock in AWS, VirtualBox, and VMware (#49) * Initailize aws support * Add README for the VM server * Refactor OSWorld for supporting more cloud services. * Initialize vmware and aws implementation v1, waiting for verification * Initlize files for azure, gcp and virtualbox support * Debug on the VMware provider * Fix on aws interface mapping * Fix instance type * Refactor * Clean * Add Azure provider * hk region; debug * Fix lock * Remove print * Remove key_name requirements when allocating aws vm * Clean README * Fix reset * Fix bugs * Add VirtualBox and Azure providers * Add VirtualBox OVF link * Raise exception on macOS host * Init RAEDME for VBox * Update VirtualBox VM download link * Update requirements and setup.py; Improve robustness on Windows * Fix network adapter * Go through on Windows machine * Add default adapter option * Fix minor error * Change resolution before creating snapshot * Fix small error * Change default provider option * Fix thread lock * Refactor for more smooth VMware support --------- Co-authored-by: Timothyxxx <384084775@qq.com> Co-authored-by: XinyuanWangCS <xywang626@gmail.com> Co-authored-by: Tianbao Xie <47296835+Timothyxxx@users.noreply.github.com>df70b11
from abc import ABC, abstractmethod


class Provider(ABC):
    def __init__(self, region: str = None):
        """
        Region of the cloud service.
        """
        self.region = region

    @abstractmethod
    def start_emulator(self, path_to_vm: str, headless: bool):
        """
        Method to start the emulator.
        """
        pass

    @abstractmethod
    def get_ip_address(self, path_to_vm: str) -> str:
        """
        Method to get the private IP address of the VM. Private IP means inside the VPC.
        """
        pass

    @abstractmethod
    def save_state(self, path_to_vm: str, snapshot_name: str):
        """
        Method to save the state of the VM.
        """
        pass

    @abstractmethod
    def revert_to_snapshot(self, path_to_vm: str, snapshot_name: str) -> str:
        """
        Method to revert the VM to a given snapshot.
        """
        pass

    @abstractmethod
    def stop_emulator(self, path_to_vm: str):
        """
        Method to stop the emulator.
        """
        pass


class VMManager(ABC):
    checked_and_cleaned = False

    @abstractmethod
    def initialize_registry(self, **kwargs):
        """
        Initialize registry.
        """
        pass

    @abstractmethod
    def add_vm(self, vm_path, **kwargs):
        """
        Add the path of new VM to the registration.
        """
        pass

    @abstractmethod
    def delete_vm(self, vm_path, **kwargs):
        """
        Delete the registration of VM by path.
        """
        pass

    @abstractmethod
    def occupy_vm(self, vm_path, pid, **kwargs):
        """
        Mark the path of VM occupied by the pid.
        """
        pass

    @abstractmethod
    def list_free_vms(self, **kwargs):
        """
        List the paths of VM that are free to use allocated.
        """
        pass

    @abstractmethod
    def check_and_clean(self, **kwargs):
        """
        Check the registration list, and remove the paths of VM that are not in use.
        """
        pass

    @abstractmethod
    def get_vm_path(self, **kwargs):
        """
        Get a virtual machine that is not occupied, generate a new one if no free VM.
        """
        pass