xos/models.py

# models.py -  ExampleService Models

from core.models import Service, TenantWithContainer
from django.db import models, transaction

ACTIVETEST_KIND = 'ActiveTest'
ACTIVETEST_SERVICE_NAME = 'activetest'
ACTIVETEST_SERVICE_NAME_VERBOSE = 'ACTIVETEST Service'
ACTIVETEST_SERVICE_NAME_VERBOSE_PLURAL = 'ACTIVETEST Services'
ACTIVETEST_TENANT_NAME_VERBOSE = 'ACTIVETEST Tenant'
ACTIVETEST_TENANT_NAME_VERBOSE_PLURAL = 'ACTIVETEST Tenants'

QT600_SERVICE_NAME = 'qt600'
QT600_SERVICE_NAME_VERBOSE = 'QT600 Service'
QT600_SERVICE_NAME_VERBOSE_PLURAL = 'QT600 Services'
QT600_TENANT_NAME_VERBOSE = 'QT600 Tenant'
QT600_TENANT_NAME_VERBOSE_PLURAL = 'QT600 Tenants'

VPMA_SERVICE_NAME = 'vpma'
VPMA_SERVICE_NAME_VERBOSE = 'VPMA Service'
VPMA_SERVICE_NAME_VERBOSE_PLURAL = 'VPMA Services'
VPMA_TENANT_NAME_VERBOSE = 'VPMA Tenant'
VPMA_TENANT_NAME_VERBOSE_PLURAL = 'VPMA Tenants'

VTA_SERVICE_NAME = 'vta'
VTA_SERVICE_NAME_VERBOSE = 'VTA Service'
VTA_SERVICE_NAME_VERBOSE_PLURAL = 'VTA Services'
VTA_TENANT_NAME_VERBOSE = 'VTA Tenant'
VTA_TENANT_NAME_VERBOSE_PLURAL = 'VTA Tenants'

class ActiveTestService(Service):

    KIND = ACTIVETEST_KIND

    class Meta:
        app_label = ACTIVETEST_SERVICE_NAME
        verbose_name = ACTIVETEST_SERVICE_NAME_VERBOSE
        proxy = True

class Qt600Service(Service):

    KIND = ACTIVETEST_KIND
    class Meta:
        app_label = QT600_SERVICE_NAME
        verbose_name = QT600_SERVICE_NAME_VERBOSE
        proxy = True

class VpmaService(Service):

    KIND = ACTIVETEST_KIND
    class Meta:
        app_label = VPMA_SERVICE_NAME
        verbose_name = VPMA_SERVICE_NAME_VERBOSE
        proxy = True

class VtaService(Service):

    KIND = ACTIVETEST_KIND
    class Meta:
        app_label = VTA_SERVICE_NAME
        verbose_name = VTA_SERVICE_NAME_VERBOSE
        proxy = True

class ActiveTestTenant(TenantWithContainer):

    KIND = ACTIVETEST_KIND

    class Meta:
        verbose_name = ACTIVETEST_TENANT_NAME_VERBOSE

    def __init__(self, *args, **kwargs):
        activetestservice = ActiveTestService.get_service_objects().all()
        if activetestservice:
            self._meta.get_field('provider_service').default = activetestservice[0].id
        super(ActiveTestTenant, self).__init__(*args, **kwargs)

    def save(self, *args, **kwargs):
        super(ActiveTestTenant, self).save(*args, **kwargs)
        model_policy_activetesttenant(self.pk)

    def delete(self, *args, **kwargs):
        self.cleanup_container()
        super(ActiveTestTenant, self).delete(*args, **kwargs)

    @property
    def public_ip(self):
        for port in self.instance.ports.all():
            if "public" in port.network.name.lower():
                return port.ip
        return None

def model_policy_activetesttenant(pk):
    with transaction.atomic():
        tenant = ActiveTestTenant.objects.select_for_update().filter(pk=pk)
        if not tenant:
            return
        tenant = tenant[0]
        tenant.manage_container()

class Qt600Tenant(TenantWithContainer):

    KIND = ACTIVETEST_KIND
    sync_attributes = ["is_reflector","is_initiator","reflector_ip","reflector_public_ip","private_ip","public_ip"]
    
    is_reflector = models.BooleanField(default=False,help_text="True if this tenant should be a TWAMP reflector")
    is_initiator = models.BooleanField(default=False,help_text="True if this tenant should be a TWAMP initiator")
    reflector = models.CharField(max_length=254,default="", blank=True, help_text="The reflector node to use for the test")

    class Meta:
        verbose_name = QT600_TENANT_NAME_VERBOSE

    def __init__(self, *args, **kwargs):
        service = Qt600Service.get_service_objects().all()
        if service:
            self._meta.get_field('provider_service').default = service[0].id
        super(Qt600Tenant, self).__init__(*args, **kwargs)

    def save(self, *args, **kwargs):
        super(Qt600Tenant, self).save(*args, **kwargs)
        model_policy_qt600tenant(self.pk)

    def delete(self, *args, **kwargs):
        self.cleanup_container()
        super(Qt600TenantTenant, self).delete(*args, **kwargs)

    @property
    def private_ip(self):
        for port in self.instance.ports.all():
            if "test_network" in port.network.name.lower():
                return port.ip
        return None

    @property
    def public_ip(self):
        for port in self.instance.ports.all():
            if "public" in port.network.name.lower():
                return port.ip
        return None

    def get_reflector_vm(self):
        service = [s for s in ActiveTestService.get_service_objects().all() if "qt600" in s.name.lower()]
        if service and self.reflector:
            slices = service[0].slices.all()
            for slice in slices:
                instance = [i for i in slice.instances.all() if self.reflector in i.instance_name.lower()]
                if len(instance) > 0:
                    return instance[0]
        return None
        
    @property
    def reflector_ip(self):
        instance = self.get_reflector_vm()
        if instance:
            for port in instance.ports.all():
                if "test_network" in port.network.name.lower():
                    return port.ip
        return None

    @property
    def reflector_public_ip(self):
        instance = self.get_reflector_vm()
        if instance:
            for port in instance.ports.all():
                if "public" in port.network.name.lower():
                    return port.ip
        return None

    @property
    def controller_ip(self):
        service = [s for s in ActiveTestService.get_service_objects().all() if "activetest" in s.name.lower()]
        if service:
            slices = service[0].slices.all()
            for slice in slices:
                for instance in slice.instances.all():
                    for port in instance.ports.all():
                        if "public" in port.network.name.lower():
                            return port.ip
        return None


def model_policy_qt600tenant(pk):
    with transaction.atomic():
        tenant = Qt600Tenant.objects.select_for_update().filter(pk=pk)
        if not tenant:
            return
        tenant = tenant[0]
        tenant.manage_container()

class VpmaTenant(TenantWithContainer):

    KIND = ACTIVETEST_KIND
    sync_attributes = ["is_reflector","is_initiator","reflector_ip","reflector_public_ip","private_ip","public_ip"]
    
    is_reflector = models.BooleanField(default=False,help_text="True if this tenant should be a TWAMP reflector")
    is_initiator = models.BooleanField(default=False,help_text="True if this tenant should be a TWAMP initiator")
    reflector = models.CharField(max_length=254,default="", blank=True, help_text="The reflector node to use for the test")

    class Meta:
        verbose_name = VPMA_TENANT_NAME_VERBOSE

    def __init__(self, *args, **kwargs):
        service = VpmaService.get_service_objects().all()
        if service:
            self._meta.get_field('provider_service').default = service[0].id
        super(VpmaTenant, self).__init__(*args, **kwargs)

    def save(self, *args, **kwargs):
        super(VpmaTenant, self).save(*args, **kwargs)
        model_policy_vpmatenant(self.pk)

    def delete(self, *args, **kwargs):
        self.cleanup_container()
        super(VpmaTenantTenant, self).delete(*args, **kwargs)

    @property
    def private_ip(self):
        for port in self.instance.ports.all():
            if "test_network" in port.network.name.lower():
                return port.ip
        return None

    @property
    def public_ip(self):
        for port in self.instance.ports.all():
            if "public" in port.network.name.lower():
                return port.ip
        return None

    def get_reflector_vm(self):
        service = [s for s in ActiveTestService.get_service_objects().all() if "vpma" in s.name.lower()]
        if service and self.reflector:
            slices = service[0].slices.all()
            for slice in slices:
                instance = [i for i in slice.instances.all() if self.reflector in i.instance_name.lower()]
                if len(instance) > 0:
                    return instance[0]
        return None
        
    @property
    def reflector_ip(self):
        instance = self.get_reflector_vm()
        if instance:
            for port in instance.ports.all():
                if "test_network" in port.network.name.lower():
                    return port.ip
        return None

    @property
    def reflector_public_ip(self):
        instance = self.get_reflector_vm()
        if instance:
            for port in instance.ports.all():
                if "public" in port.network.name.lower():
                    return port.ip
        return None

    @property
    def controller_ip(self):
        service = [s for s in ActiveTestService.get_service_objects().all() if "activetest" in s.name.lower()]
        if service:
            slices = service[0].slices.all()
            for slice in slices:
                for instance in slice.instances.all():
                    for port in instance.ports.all():
                        if "public" in port.network.name.lower():
                            return port.ip
        return None


def model_policy_vpmatenant(pk):
    with transaction.atomic():
        tenant = VpmaTenant.objects.select_for_update().filter(pk=pk)
        if not tenant:
            return
        tenant = tenant[0]
        tenant.manage_container()

class VtaTenant(TenantWithContainer):

    KIND = ACTIVETEST_KIND
    sync_attributes = ["private_ip"]

    class Meta:
        verbose_name = VTA_TENANT_NAME_VERBOSE

    def __init__(self, *args, **kwargs):
        service = VtaService.get_service_objects().all()
        if service:
            self._meta.get_field('provider_service').default = service[0].id
        super(VtaTenant, self).__init__(*args, **kwargs)

    def save(self, *args, **kwargs):
        super(VtaTenant, self).save(*args, **kwargs)
        model_policy_vtatenant(self.pk)

    def delete(self, *args, **kwargs):
        self.cleanup_container()
        super(VtaTenantTenant, self).delete(*args, **kwargs)

    @property
    def private_ip(self):
        for port in self.instance.ports.all():
            if "test_network" in port.network.name.lower():
                return port.ip
        return None

    @property
    def public_ip(self):
        for port in self.instance.ports.all():
            if "public" in port.network.name.lower():
                return port.ip
        return None

    @property
    def controller_ip(self):
        service = [s for s in ActiveTestService.get_service_objects().all() if "activetest" in s.name.lower()]
        if service:
            slices = service[0].slices.all()
            for slice in slices:
                for instance in slice.instances.all():
                    for port in instance.ports.all():
                        if "public" in port.network.name.lower():
                            return port.ip
        return None

def model_policy_vtatenant(pk):
    with transaction.atomic():
        tenant = VtaTenant.objects.select_for_update().filter(pk=pk)
        if not tenant:
            return
        tenant = tenant[0]
        tenant.manage_container()