Add OVH VPS compute driver support via ex_ extension methods

OVH VPS is a separate product line from Public Cloud that uses the
proprietary /vps/ API. This adds 7 extension methods to OvhNodeDriver:
ex_list_vps, ex_get_vps, ex_reboot_vps, ex_start_vps, ex_stop_vps,
ex_rebuild_vps, and ex_list_vps_images. Includes unit tests with mock
fixtures and updated documentation.

Author: Vibe Kanban

docs/compute/drivers/ovh.rst

@@ -70,6 +70,51 @@ Create and attach a volume to a node
 
 .. literalinclude:: /examples/compute/ovh/attach_volume.py
 
+VPS (Virtual Private Server) support
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The OVH driver also supports managing OVH VPS instances, which are a separate
+product line from the Public Cloud. VPS methods use the ``/vps/`` API and are
+available as ``ex_``-prefixed extension methods.
+
+Note: VPS operations do not require ``ex_project_id``, but the driver
+constructor still requires it for Public Cloud operations. You can pass any
+value if you only need VPS functionality.
+
+.. code-block:: python
+
+    from libcloud.compute.types import Provider
+    from libcloud.compute.providers import get_driver
+
+    Ovh = get_driver(Provider.OVH)
+    driver = Ovh(
+        "your_app_key",
+        "your_app_secret",
+        "project_id",
+        "your_consumer_key",
+    )
+
+    # List all VPS
+    vps_names = driver.ex_list_vps()
+    for name in vps_names:
+        node = driver.ex_get_vps(name)
+        print(f"{node.name}: {node.state} - IPs: {node.public_ips}")
+
+    # Reboot a VPS
+    driver.ex_reboot_vps("vps-12345678.vps.ovh.net")
+
+    # List available OS images for a VPS
+    images = driver.ex_list_vps_images("vps-12345678.vps.ovh.net")
+    for image in images:
+        print(f"{image.id}: {image.name}")
+
+    # Rebuild a VPS with a new OS
+    driver.ex_rebuild_vps(
+        "vps-12345678.vps.ovh.net",
+        "img-debian-12",
+        ssh_key=["ssh-rsa AAAA..."],
+    )
+
 API Docs
 --------
 

libcloud/compute/drivers/ovh.py

@@ -26,7 +26,7 @@
     StorageVolume,
     VolumeSnapshot,
 )
-from libcloud.compute.types import Provider, StorageVolumeState, VolumeSnapshotState
+from libcloud.compute.types import NodeState, Provider, StorageVolumeState, VolumeSnapshotState
 from libcloud.compute.drivers.openstack import OpenStackKeyPair, OpenStackNodeDriver
 
 
@@ -50,6 +50,16 @@ class OvhNodeDriver(NodeDriver):
     VOLUME_STATE_MAP = OpenStackNodeDriver.VOLUME_STATE_MAP
     SNAPSHOT_STATE_MAP = OpenStackNodeDriver.SNAPSHOT_STATE_MAP
 
+    VPS_STATE_MAP = {
+        "running": NodeState.RUNNING,
+        "stopped": NodeState.STOPPED,
+        "installing": NodeState.PENDING,
+        "rebooting": NodeState.REBOOTING,
+        "stopping": NodeState.STOPPING,
+        "rescued": NodeState.PENDING,
+        "maintenance": NodeState.PENDING,
+    }
+
     def __init__(self, key, secret, ex_project_id, ex_consumer_key=None, region=None):
         """
         Instantiate the driver with the given API credentials.
@@ -629,5 +639,167 @@ def _to_snapshot(self, obj):
     def _to_snapshots(self, objs):
         return [self._to_snapshot(obj) for obj in objs]
 
+    # ------------------------------------------------------------------
+    # VPS (Virtual Private Server) extension methods
+    # ------------------------------------------------------------------
+    # OVH VPS is a separate product line from Public Cloud instances.
+    # It uses the /vps/ API instead of /cloud/project/.
+    # ------------------------------------------------------------------
+
+    def ex_list_vps(self):
+        """
+        List all VPS on the account.
+
+        :return: List of VPS names
+        :rtype: ``list`` of ``str``
+        """
+        action = "{}/vps".format(API_ROOT)
+        response = self.connection.request(action)
+
+        return response.object
+
+    def ex_get_vps(self, name):
+        """
+        Get VPS details as a :class:`Node`.
+
+        :param name: VPS name (e.g. ``vps-12345678.vps.ovh.net``)
+        :type name: ``str``
+
+        :return: Node representing the VPS
+        :rtype: :class:`Node`
+        """
+        action = "{}/vps/{}".format(API_ROOT, name)
+        response = self.connection.request(action)
+
+        return self._to_vps_node(response.object)
+
+    def ex_reboot_vps(self, name):
+        """
+        Reboot a VPS.
+
+        :param name: VPS name
+        :type name: ``str``
+
+        :return: True on success
+        :rtype: ``bool``
+        """
+        action = "{}/vps/{}/reboot".format(API_ROOT, name)
+        self.connection.request(action, method="POST")
+
+        return True
+
+    def ex_start_vps(self, name):
+        """
+        Start a VPS.
+
+        :param name: VPS name
+        :type name: ``str``
+
+        :return: True on success
+        :rtype: ``bool``
+        """
+        action = "{}/vps/{}/start".format(API_ROOT, name)
+        self.connection.request(action, method="POST")
+
+        return True
+
+    def ex_stop_vps(self, name):
+        """
+        Stop a VPS.
+
+        :param name: VPS name
+        :type name: ``str``
+
+        :return: True on success
+        :rtype: ``bool``
+        """
+        action = "{}/vps/{}/stop".format(API_ROOT, name)
+        self.connection.request(action, method="POST")
+
+        return True
+
+    def ex_rebuild_vps(self, name, image_id, ssh_key=None):
+        """
+        Reinstall a VPS with a new OS image.
+
+        :param name: VPS name
+        :type name: ``str``
+
+        :param image_id: Image ID to install
+        :type image_id: ``str``
+
+        :param ssh_key: SSH public key(s) to install (optional)
+        :type ssh_key: ``list`` of ``str``
+
+        :return: True on success
+        :rtype: ``bool``
+        """
+        action = "{}/vps/{}/rebuild".format(API_ROOT, name)
+        data = {"imageId": image_id}
+
+        if ssh_key:
+            data["sshKey"] = ssh_key
+        self.connection.request(action, data=data, method="POST")
+
+        return True
+
+    def ex_list_vps_images(self, name):
+        """
+        List available OS images for a VPS.
+
+        :param name: VPS name
+        :type name: ``str``
+
+        :return: List of available images
+        :rtype: ``list`` of :class:`NodeImage`
+        """
+        action = "{}/vps/{}/images/available".format(API_ROOT, name)
+        response = self.connection.request(action)
+
+        return self._to_vps_images(response.object)
+
+    def _to_vps_node(self, obj):
+        state = self.VPS_STATE_MAP.get(obj.get("state", ""), NodeState.UNKNOWN)
+
+        public_ips = []
+        if obj.get("ips"):
+            public_ips = obj["ips"]
+        elif obj.get("ip"):
+            public_ips = [obj["ip"]]
+
+        extra = {
+            "model": obj.get("model"),
+            "netbootMode": obj.get("netbootMode"),
+            "offerType": obj.get("offerType"),
+            "vcore": obj.get("vcore"),
+            "memory": obj.get("memory"),
+            "disk": obj.get("disk"),
+            "zone": obj.get("zone"),
+        }
+
+        return Node(
+            id=obj.get("name", ""),
+            name=obj.get("displayName") or obj.get("name", ""),
+            state=state,
+            public_ips=public_ips,
+            private_ips=[],
+            driver=self,
+            extra=extra,
+        )
+
+    def _to_vps_images(self, objs):
+        images = []
+        for obj in objs:
+            images.append(
+                NodeImage(
+                    id=obj.get("id", ""),
+                    name=obj.get("name", ""),
+                    driver=self,
+                    extra={"lastModifiedDate": obj.get("lastModifiedDate")},
+                )
+            )
+
+        return images
+
     def _ex_connection_class_kwargs(self):
         return {"ex_consumer_key": self.consumer_key, "region": self.region}

libcloud/test/compute/fixtures/ovh/vps_get.json

@@ -0,0 +1 @@
+["testvps", "testvps2"]

libcloud/test/compute/fixtures/ovh/vps_get_detail.json

@@ -0,0 +1 @@
+{"name": "testvps", "displayName": "my-vps", "state": "running", "model": {"name": "VPS Value 1-2-40", "offer": "value", "version": "2019v1", "vcore": 1, "memory": 2048, "disk": 40, "maximumAdditionalIp": 16}, "netbootMode": "local", "offerType": "ssd", "vcore": 1, "memory": 2048, "disk": 40, "zone": "eu-west-rbx", "ips": ["198.51.100.42", "2001:db8::1"]}

libcloud/test/compute/fixtures/ovh/vps_images_available_get.json

@@ -0,0 +1 @@
+[{"id": "img-debian-12", "name": "Debian 12", "lastModifiedDate": "2025-01-15T10:00:00Z"}, {"id": "img-ubuntu-2404", "name": "Ubuntu 24.04", "lastModifiedDate": "2025-02-01T12:00:00Z"}, {"id": "img-centos-stream-9", "name": "CentOS Stream 9", "lastModifiedDate": "2024-11-20T08:00:00Z"}]

libcloud/test/compute/fixtures/ovh/vps_reboot_post.json

@@ -0,0 +1 @@
+{"taskId": 12345, "type": "rebootVm", "state": "todo"}

libcloud/test/compute/fixtures/ovh/vps_rebuild_post.json

@@ -0,0 +1 @@
+{"taskId": 12348, "type": "reinstallVm", "state": "todo"}

libcloud/test/compute/fixtures/ovh/vps_start_post.json

@@ -0,0 +1 @@
+{"taskId": 12346, "type": "startVm", "state": "todo"}

libcloud/test/compute/fixtures/ovh/vps_stop_post.json

@@ -0,0 +1 @@
+{"taskId": 12347, "type": "stopVm", "state": "todo"}

libcloud/test/compute/test_ovh.py

@@ -22,6 +22,7 @@
 from libcloud.test.secrets import OVH_PARAMS
 from libcloud.common.exceptions import BaseHTTPError
 from libcloud.test.file_fixtures import ComputeFileFixtures
+from libcloud.compute.types import NodeState
 from libcloud.compute.drivers.ovh import OvhNodeDriver
 from libcloud.test.common.test_ovh import BaseOvhMockHttp
 
@@ -155,6 +156,36 @@ def _json_1_0_cloud_project_project_id_instance_get_invalid_app_key_error(
         body = '{"message":"Invalid application key"}'
         return (httplib.UNAUTHORIZED, body, {}, httplib.responses[httplib.OK])
 
+    # VPS mock handlers
+
+    def _json_1_0_vps_get(self, method, url, body, headers):
+        body = self.fixtures.load("vps_get.json")
+        return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+    def _json_1_0_vps_testvps_get(self, method, url, body, headers):
+        body = self.fixtures.load("vps_get_detail.json")
+        return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+    def _json_1_0_vps_testvps_reboot_post(self, method, url, body, headers):
+        body = self.fixtures.load("vps_reboot_post.json")
+        return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+    def _json_1_0_vps_testvps_start_post(self, method, url, body, headers):
+        body = self.fixtures.load("vps_start_post.json")
+        return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+    def _json_1_0_vps_testvps_stop_post(self, method, url, body, headers):
+        body = self.fixtures.load("vps_stop_post.json")
+        return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+    def _json_1_0_vps_testvps_rebuild_post(self, method, url, body, headers):
+        body = self.fixtures.load("vps_rebuild_post.json")
+        return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
+    def _json_1_0_vps_testvps_images_available_get(self, method, url, body, headers):
+        body = self.fixtures.load("vps_images_available_get.json")
+        return (httplib.OK, body, {}, httplib.responses[httplib.OK])
+
 
 @patch("libcloud.common.ovh.OvhConnection._timedelta", 42)
 class OvhTests(unittest.TestCase):
@@ -305,6 +336,62 @@ def test_destroy_volume_snapshot(self):
     def test_get_pricing(self):
         self.driver.ex_get_pricing("foo-id")
 
+    # VPS tests
+
+    def test_ex_list_vps(self):
+        names = self.driver.ex_list_vps()
+        self.assertEqual(len(names), 2)
+        self.assertEqual(names[0], "testvps")
+        self.assertEqual(names[1], "testvps2")
+
+    def test_ex_get_vps(self):
+        node = self.driver.ex_get_vps("testvps")
+        self.assertEqual(node.id, "testvps")
+        self.assertEqual(node.name, "my-vps")
+        self.assertEqual(node.state, NodeState.RUNNING)
+        self.assertEqual(len(node.public_ips), 2)
+        self.assertIn("198.51.100.42", node.public_ips)
+        self.assertEqual(node.extra["vcore"], 1)
+        self.assertEqual(node.extra["memory"], 2048)
+        self.assertEqual(node.extra["disk"], 40)
+        self.assertEqual(node.extra["zone"], "eu-west-rbx")
+
+    def test_ex_reboot_vps(self):
+        result = self.driver.ex_reboot_vps("testvps")
+        self.assertTrue(result)
+
+    def test_ex_start_vps(self):
+        result = self.driver.ex_start_vps("testvps")
+        self.assertTrue(result)
+
+    def test_ex_stop_vps(self):
+        result = self.driver.ex_stop_vps("testvps")
+        self.assertTrue(result)
+
+    def test_ex_rebuild_vps(self):
+        result = self.driver.ex_rebuild_vps("testvps", "img-debian-12")
+        self.assertTrue(result)
+
+    def test_ex_rebuild_vps_with_ssh_key(self):
+        result = self.driver.ex_rebuild_vps(
+            "testvps", "img-debian-12", ssh_key=["ssh-rsa AAAA..."]
+        )
+        self.assertTrue(result)
+
+    def test_ex_list_vps_images(self):
+        images = self.driver.ex_list_vps_images("testvps")
+        self.assertEqual(len(images), 3)
+        self.assertEqual(images[0].id, "img-debian-12")
+        self.assertEqual(images[0].name, "Debian 12")
+        self.assertEqual(images[1].id, "img-ubuntu-2404")
+        self.assertEqual(images[1].name, "Ubuntu 24.04")
+
+    def test_vps_state_map(self):
+        self.assertEqual(self.driver.VPS_STATE_MAP["running"], NodeState.RUNNING)
+        self.assertEqual(self.driver.VPS_STATE_MAP["stopped"], NodeState.STOPPED)
+        self.assertEqual(self.driver.VPS_STATE_MAP["rebooting"], NodeState.REBOOTING)
+        self.assertEqual(self.driver.VPS_STATE_MAP["installing"], NodeState.PENDING)
+
 
 if __name__ == "__main__":
     sys.exit(unittest.main())