lxc-network/lxc_net/network.py

""" Network objects """

from . import settings
from . import util
from .jinja_template import JinjaTemplate

import asyncio


class Network(util.LibvirtObject):
    class AlreadyExists(Exception):
        def __str__(self):
            return "This network is already instanciated"

    def __init__(self, conn, name=None, enable_v4=True):
        super().__init__(conn)

        if not name:
            name = str(self.id)

        self.name = settings.PREFIX + "_link_" + name
        self.enable_v4 = enable_v4
        self.bridge_id = settings.NETWORK_ID * 0xFF + self.id
        self.bridge_mac = util.MACAddress(self.id, None)
        self.ipv4 = util.Addrv4(self.id, None, host_address=True)
        self.ipv6 = util.Addrv6(self.id, None, host_address=True)
        self.lxc_network = None

    @property
    def up(self):
        return self.lxc_network is not None

    def create(self):
        if self.lxc_network:
            raise self.AlreadyExists()

        xml = JinjaTemplate("network.xml").inst(
            name=self.name,
            uuid=self.uuid,
            bridge_id=self.bridge_id,
            mac=self.bridge_mac,
            addrv4=self.ipv4,
            netmaskv4=self.ipv4.netmask,
            addrv6=self.ipv6,
            prefixv6=self.ipv6.prefix,
        )

        self.lxc_network = self.conn.networkCreateXML(xml)

    async def async_create(self):
        """ Same as `create`, asyncio-enabled """
        executor_create = util.run_in_executor(self.create)
        await executor_create()

    def cleanup(self):
        if self.lxc_network:
            self.lxc_network.destroy()
            self.lxc_network = None

    def __enter__(self):
        self.create()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cleanup()

    def __del__(self):
        self.cleanup()