Adding support for network forwards

pylxd/managers.py

@@ -49,6 +49,10 @@ class NetworkManager(BaseManager):
     manager_for = "pylxd.models.Network"
 
 
+class NetworkForwardManager(BaseManager):
+    manager_for = "pylxd.models.NetworkForward"
+
+
 class OperationManager(BaseManager):
     manager_for = "pylxd.models.Operation"
 

pylxd/models/__init__.py

@@ -3,7 +3,7 @@
 from pylxd.models.container import Container
 from pylxd.models.image import Image
 from pylxd.models.instance import Instance, Snapshot
-from pylxd.models.network import Network
+from pylxd.models.network import Network, NetworkForward
 from pylxd.models.operation import Operation
 from pylxd.models.profile import Profile
 from pylxd.models.project import Project
@@ -19,6 +19,7 @@
     "Image",
     "Instance",
     "Network",
+    "NetworkForward",
     "Operation",
     "Profile",
     "Project",

pylxd/models/network.py

@@ -13,9 +13,50 @@
 #    under the License.
 import json
 
+from pylxd import managers
 from pylxd.models import _model as model
 
 
+class NetworkForward(model.Model):
+    config = model.Attribute()
+    description = model.Attribute()
+    location = model.Attribute()
+    listen_address = model.Attribute()
+    ports = model.Attribute()
+
+    network = model.Parent()
+
+    @classmethod
+    def get(cls, client, network, listen_address):
+        response = client.api.networks[network.name].forwards[listen_address].get()
+        forward = cls(client, network=network, **response.json()["metadata"])
+        return forward
+
+    @classmethod
+    def create(cls, client, network, config):
+        client.api.networks[network.name].forwards.post(json=config)
+
+        return cls(client, network=network, **config)
+
+    def save(self, *args, **kwargs):
+        self.client.assert_has_api_extension("network")
+        super().save(*args, **kwargs)
+
+    @property
+    def api(self):
+        return self.client.api.networks[self.network.name].forwards[self.listen_address]
+
+    def __str__(self):
+        return json.dumps(self.marshall(skip_readonly=False), indent=2)
+
+    def __repr__(self):
+        attrs = []
+        for attribute, value in self.marshall().items():
+            attrs.append("{}={}".format(attribute, json.dumps(value, sort_keys=True)))
+
+        return "{}({})".format(self.__class__.__name__, ", ".join(sorted(attrs)))
+
+
 class NetworkState(model.AttributeDict):
     """A simple object for representing a network state."""
 
@@ -31,6 +72,13 @@ class Network(model.Model):
     locations = model.Attribute(readonly=True)
     managed = model.Attribute(readonly=True)
     used_by = model.Attribute(readonly=True)
+    _endpoint = "networks"
+
+    forwards = model.Manager()
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.forwards = managers.NetworkForwardManager(self.client, self)
 
     @classmethod
     def exists(cls, client, name):

pylxd/models/tests/test_network.py

@@ -281,3 +281,165 @@ def test_repr(self):
             '"true", "ipv6.address": "none", "ipv6.nat": "false"}, '
             'description="Network description", name="eth0", type="bridge")',
         )
+
+
+class TestNetworkForward(testing.PyLXDTestCase):
+    """Tests for pylxd.models.NetworkForward."""
+
+    def test_get(self):
+        network_name = "eth0"
+        an_network = models.Network.get(self.client, network_name)
+        forward = an_network.forwards.get('192.0.2.1')
+
+        self.assertEqual("Forward description", forward.description)
+
+    def test_get_not_found(self):
+        """LXDAPIException is raised on unknown network."""
+        network_name = "eth0"
+        an_network = models.Network.get(self.client, network_name)
+
+        def not_found(_, context):
+            context.status_code = 404
+            return json.dumps(
+                {"type": "error", "error": "Not found", "error_code": 404}
+            )
+
+        self.add_rule(
+            {
+                "text": not_found,
+                "method": "GET",
+                "url": r"^http://pylxd.test/1.0/networks/eth0/forwards/192.0.2.1$",
+            }
+        )
+
+        self.assertRaises(
+            exceptions.LXDAPIException, models.NetworkForward.get, self.client, an_network, "192.0.2.1"
+        )
+
+    def test_get_error(self):
+        """LXDAPIException is raised on error."""
+        network_name = "eth0"
+        an_network = models.Network.get(self.client, network_name)
+
+        def error(_, context):
+            context.status_code = 500
+            return json.dumps(
+                {
+                    "type": "error",
+                    "error": "Not found",
+                    "error_code": 500,
+                }
+            )
+
+        self.add_rule(
+            {
+                "text": error,
+                "method": "GET",
+                "url": r"^http://pylxd.test/1.0/networks/eth0/forwards/192.0.2.1$",
+            }
+        )
+
+        self.assertRaises(
+            exceptions.LXDAPIException, models.NetworkForward.get, self.client, an_network, "192.0.2.1"
+        )
+
+    def test_create(self):
+        network_name = "eth0"
+        an_network = models.Network.get(self.client, network_name)
+        config = {
+            "config": {},
+            "description": "Forward description",
+            "listen_address": "192.0.2.1",
+            "ports": [
+                {
+                    "description": "Port description",
+                    "listen_port": "80",
+                    "target_address": "192.0.2.2",
+                    "target_port": "80",
+                }
+            ],
+        }
+
+        with mock.patch.object(self.client, "assert_has_api_extension"):
+            forward = models.NetworkForward.create(
+                self.client,
+                network=an_network,
+                config=config,
+            )
+
+        self.assertIsInstance(forward, models.NetworkForward)
+        self.assertEqual("Forward description", forward.description)
+        self.assertEqual("192.0.2.1", forward.listen_address)
+        self.assertEqual(len(forward.ports), 1)
+        self.assertEqual("80", forward.ports[0]["listen_port"])
+        self.assertEqual("192.0.2.2", forward.ports[0]["target_address"])
+        self.assertEqual("80", forward.ports[0]["target_port"])
+        self.assertEqual("Port description", forward.ports[0]["description"])
+
+    def test_update(self):
+        network_name = "eth0"
+        an_network = models.Network.get(self.client, network_name)
+        config = {
+            "config": {},
+            "description": "Forward description",
+            "listen_address": "192.0.2.1",
+            "ports": [
+                {
+                    "description": "Port description",
+                    "listen_port": "80",
+                    "target_address": "192.0.2.2",
+                    "target_port": "80",
+                }
+            ],
+        }
+
+        with mock.patch.object(self.client, "assert_has_api_extension"):
+            forward = models.NetworkForward.create(
+                self.client,
+                network=an_network,
+                config=config,
+            )
+            forward.description = "Updated"
+            forward.save()
+
+        self.assertIsInstance(forward, models.NetworkForward)
+        self.assertEqual("Updated", forward.description)
+        self.assertEqual("192.0.2.1", forward.listen_address)
+        self.assertEqual(len(forward.ports), 1)
+        self.assertEqual("80", forward.ports[0]["listen_port"])
+        self.assertEqual("192.0.2.2", forward.ports[0]["target_address"])
+        self.assertEqual("80", forward.ports[0]["target_port"])
+        self.assertEqual("Port description", forward.ports[0]["description"])
+
+    def test_str(self):
+        network_name = "eth0"
+        an_network = models.Network.get(self.client, network_name)
+        forward = an_network.forwards.get('192.0.2.1')
+        self.assertEqual(
+            json.loads(str(forward)),
+            {
+                "config": {},
+                "description": "Forward description",
+                "location": "eth0",
+                "listen_address": "192.0.2.1",
+                "ports": [
+                    {
+                        "description": "Port description",
+                        "listen_port": "80",
+                        "target_address": "192.0.2.2",
+                        "target_port": "80"
+                    }
+                ]
+            }
+        )
+
+    def test_repr(self):
+        network_name = "eth0"
+        an_network = models.Network.get(self.client, network_name)
+        forward = an_network.forwards.get('192.0.2.1')
+        self.assertEqual(
+            repr(forward),
+            'NetworkForward(config={}, description="Forward description", listen_address="192.0.2.1",'
+            ' location="eth0", ports=[{"description": "Port description", "listen_port": "80",'
+            ' "target_address": "192.0.2.2", "target_port": "80"}])'
+        )

pylxd/tests/mock_lxd.py

@@ -830,6 +830,70 @@ def snapshot_DELETE(request, context):
         "method": "DELETE",
         "url": r"^http://pylxd.test/1.0/networks/eth0$",
     },
+    # Network forwards
+    {
+        "json": {
+            "type": "sync",
+            "metadata": {
+                "config": {},
+                "description": "Forward description",
+                "listen_address": "192.0.2.1",
+                "location": "eth0",
+                "ports": [
+                    {
+                        "description": "Port description",
+                        "listen_port": "80",
+                        "target_address": "192.0.2.2",
+                        "target_port": "80",
+                    }
+                ],
+            },
+        },
+        "method": "GET",
+        "url": r"^http://pylxd.test/1.0/networks/eth0/forwards/192.0.2.1$",
+    },
+    {
+        "json": {
+            "type": "sync",
+            "metadata": {
+                "config": {},
+                "description": "Forward description",
+                "listen_address": "192.0.2.1",
+                "location": "eth0",
+                "ports": [
+                    {
+                        "description": "Port description",
+                        "listen_port": "80",
+                        "target_address": "192.0.2.2",
+                        "target_port": "80",
+                    }
+                ],
+            },
+        },
+        "method": "POST",
+        "url": r"^http://pylxd.test/1.0/networks/eth0/forwards$",
+    },
+    {
+        "json": {
+            "type": "sync",
+            "metadata": {
+                "config": {},
+                "description": "Updated",
+                "listen_address": "192.0.2.1",
+                "location": "eth0",
+                "ports": [
+                    {
+                        "description": "Port description",
+                        "listen_port": "80",
+                        "target_address": "192.0.2.2",
+                        "target_port": "80",
+                    }
+                ],
+            },
+        },
+        "method": "PUT",
+        "url": r"^http://pylxd.test/1.0/networks/eth0/forwards/192.0.2.1$",
+    },
     # Storage Pools
     {
         "json": {