djadhere/services/models.py

from django.db import models
from django.contrib.gis.db import models as geo_models
from django.db.models import Q
from django.core.validators import MaxValueValidator
from django.utils import timezone
from django.contrib.auth.models import Group
from django.contrib.contenttypes.fields import GenericRelation
from django.core.exceptions import ValidationError
from django.urls import reverse
from django.utils import timezone
from django.utils.html import format_html, mark_safe
from django.core.exceptions import PermissionDenied
from django.core.validators import RegexValidator

from ipaddress import ip_network

from djadhere.utils import get_active_filter, is_overlapping
from adhesions.models import Adhesion
from banking.models import RecurringPayment


def ipprefix_validator(value):
    try:
        ip_network(value)
    except ValueError:
        raise ValidationError('%s n’est pas un préfixe valide' % value)


class IPPrefix(models.Model):
    prefix = models.CharField(max_length=128, verbose_name='Préfixe', validators=[ipprefix_validator], unique=True)

    class Meta:
        ordering = ['prefix']
        verbose_name = 'Réseau'
        verbose_name_plural = 'Réseaux'

    def __str__(self):
        return self.prefix


class IPResourceManager(models.Manager):
    def get_queryset(self):
        qs = super().get_queryset()
        # On rajoute une super annotation « in_use » pour savoir si l’IP est dispo ou non :-)
        qs = qs.annotate(
                    in_use_by_service=models.Exists(
                        ServiceAllocation.objects.filter(resource=models.OuterRef('pk'), active=True)
                    ),
                    in_use_by_antenna=models.Exists(
                        ServiceAllocation.objects.filter(resource=models.OuterRef('pk'), active=True)
                    )
        )
        qs = qs.annotate(
                    in_use=models.Case(
                        models.When(Q(in_use_by_service=True) | Q(in_use_by_antenna=True), then=True),
                        default=False,
                        output_field=models.BooleanField()
                    )
        )
        return qs


class ActiveAllocationManager(models.Manager):
    def get_queryset(self):
        qs = super().get_queryset()
        qs = qs.annotate(
                    active=models.Case(
                        models.When(get_active_filter(), then=True),
                        default=False,
                        output_field=models.BooleanField()
                    )
        )
        return qs


class IPResource(models.Model):
    CATEGORIES = (
        (0, 'IP Public'),
        (1, 'IP Antenne'),
    )

    ip = models.GenericIPAddressField(verbose_name='IP', primary_key=True)
    prefixes = models.ManyToManyField(IPPrefix, verbose_name='préfixes')
    reserved = models.BooleanField(default=False, verbose_name='réservée')
    category = models.IntegerField(choices=CATEGORIES, verbose_name='catégorie')
    notes = models.TextField(blank=True, default='')

    objects = IPResourceManager()

    @property
    def allocations(self):
        if self.category == 0:
            return self.service_allocations
        if self.category == 1:
            return self.antenna_allocations

    class Meta:
        ordering = ['ip']
        verbose_name = 'IP'
        verbose_name_plural = 'IP'

    def __str__(self):
        return str(self.ip)


class ServiceType(models.Model):
    name = models.CharField(max_length=64, verbose_name='Nom', unique=True)

    class Meta:
        ordering = ['name']
        verbose_name = 'type de service'
        verbose_name_plural = 'types de service'

    def __str__(self):
        return self.name


class Service(models.Model):
    adhesion = models.ForeignKey(Adhesion, verbose_name='Adhérent', related_name='services')
    service_type = models.ForeignKey(ServiceType, related_name='services',
                                     verbose_name='Type de service')
    label = models.CharField(blank=True, default='', max_length=128)
    notes = models.TextField(blank=True, default='')
    active = models.BooleanField(default=True, verbose_name='actif')
    created = models.DateTimeField(auto_now_add=True)

    contribution = models.OneToOneField(RecurringPayment)

    def save(self, *args, **kwargs):
        if not hasattr(self, 'contribution'):
            self.contribution = RecurringPayment.objects.create()
        super().save(*args, **kwargs)

    def clean(self):
        super().clean()
        # Vérification de l’unicité par type de service du label
        if self.label != '' and Service.objects.exclude(pk=self.pk).filter(service_type=self.service_type, label=self.label):
            raise ValidationError("Un service du même type existe déjà avec ce label.")

    def get_absolute_url(self):
        return reverse('admin:%s_%s_change' % (self._meta.app_label, self._meta.model_name), args=(self.pk,))

    def __str__(self):
        s = '#%d %s' % (self.pk, self.service_type)
        if self.label:
            s += ' ' + self.label
        return s


class Antenna(models.Model):
    MODE_UNKNOWN = 0
    MODE_AP = 1
    MODE_STA = 2
    MODE_CHOICES = (
        (MODE_UNKNOWN, 'Inconnu'),
        (MODE_AP, 'AP'),
        (MODE_STA, 'Station'),
    )
    label = models.CharField(max_length=128, blank=True, default='')
    mode = models.IntegerField(choices=MODE_CHOICES, default=MODE_UNKNOWN)
    ssid = models.CharField(max_length=64, blank=True, default='', verbose_name='SSID')
    mac = models.CharField(
            blank=True,
            default='',
            max_length=17,
            validators=[
                RegexValidator(r'^([0-9a-fA-F]{2}([:-]?|$)){6}$'),
            ],
            verbose_name='Adresse MAC')
    contact = models.ForeignKey(Adhesion, null=True, blank=True)
    notes = models.TextField(blank=True)
    position = geo_models.PointField(null=True, blank=True)
    orientation = models.IntegerField(verbose_name='Orientation (°)', null=True, blank=True)

    def clean(self):
        super().clean()
        if self.orientation:
            self.orientation = self.orientation % 360

    def get_absolute_url(self):
        return reverse('admin:%s_%s_change' % (self._meta.app_label, self._meta.model_name), args=(self.pk,))

    def get_absolute_link(self):
        name = 'Antenne n°%d' % self.pk
        if self.label:
            name += ' : %s' % self.label
        link = format_html('<a href="{}">{}</a>', self.get_absolute_url(), name)
        if self.allocations.filter(active=True).exists():
            link += ' ('
            link += ', '.join(map(
                            lambda alloc: format_html('<a href="http://{}">{}</a>', alloc.resource, alloc.resource),
                            self.allocations.filter(active=True).all()
            ))
            link += ')'
        return mark_safe(link)

    class Meta:
        verbose_name = 'antenne'

    def __str__(self):
        name = 'Antenne %d' % self.pk
        if self.label:
            name += ' (%s)' % self.label
        return name


class Route(models.Model):
    name = models.CharField(max_length=64, unique=True)

    def __str__(self):
        return self.name


class Tunnel(Route):
    description = models.CharField(max_length=128, blank=True)
    created = models.DateTimeField(default=timezone.now, verbose_name='Date de création')
    ended = models.DateTimeField(null=True, blank=True, verbose_name='Date de désactivation')
    port = models.IntegerField(null=True, blank=True)
    local_ip = models.GenericIPAddressField(null=True, blank=True, verbose_name='IP locale')
    remote_ip = models.GenericIPAddressField(null=True, blank=True, verbose_name='IP distante')
    networks = models.ManyToManyField(IPPrefix, blank=True, verbose_name='Réseaux')
    notes = models.TextField(blank=True, default='')

    def clean(self):
        super().clean()
        if self.ended:
            # Vérification de la cohérence des champs created et ended
            if self.created > self.ended:
                raise ValidationError({'ended': "La date de désactivation doit être postérieur "
                                                "à la date de création du tunnel."})
        elif self.port:
            # Vérification de l’unicité d’un tunnel actif avec un port donné
            if Tunnel.objects.exclude(pk=self.pk).filter(port=self.port, ended__isnull=True).exists():
                raise ValidationError({'port': "Ce numéro de port est déjà utilisé par un autre tunnel."})


class Allocation(models.Model):
    start = models.DateTimeField(verbose_name='Début de la période d’allocation', default=timezone.now)
    end = models.DateTimeField(null=True, blank=True, verbose_name='Fin de la période d’allocation')
    notes = models.TextField(blank=True, default='')

    objects = ActiveAllocationManager()

    def clean(self):
        super().clean()
        # Vérification de la cohérence des champs start et end
        if self.end and self.start > self.end:
            raise ValidationError("La date de début de l’allocation doit être antérieur "
                                  "à la date de fin de l’allocation.")
        if self.resource_id:
            if self.resource.reserved and (not self.end or self.end > timezone.now()):
                raise ValidationError("L’IP sélectionnée est réservée")
            # Vérification de l’abscence de chevauchement de la période d’allocation
            allocations = type(self).objects.filter(resource__pk=self.resource.pk)
            if is_overlapping(self, allocations):
                raise ValidationError("La période d’allocation de cette ressource chevauche "
                                        "avec une période d’allocation précédente.")

    class Meta:
        abstract = True
        ordering = ['-start']

    def __str__(self):
        return str(self.resource)


class ServiceAllocation(Allocation):
    resource = models.ForeignKey(IPResource, verbose_name='Ressource', related_name='service_allocations',
                        related_query_name='service_allocation', limit_choices_to={'category': 0})
    service = models.ForeignKey(Service, related_name='allocations', related_query_name='allocation')
    route = models.ForeignKey(Route, verbose_name='Route', related_name='allocations', related_query_name='allocation')

    class Meta:
        verbose_name = 'allocation'
        verbose_name_plural = 'allocations'


class AntennaAllocation(Allocation):
    resource = models.ForeignKey(IPResource, verbose_name='Ressource', related_name='antenna_allocations',
                        related_query_name='antenna_allocation', limit_choices_to={'category': 1})
    antenna = models.ForeignKey(Antenna, related_name='allocations', related_query_name='allocation')

    class Meta:
        verbose_name = 'allocation'
        verbose_name_plural = 'allocations'