sven
b1c9c7a40b
modified: netbox_services/engine/allocators.py modified: netbox_services/engine/context.py new file: netbox_services/engine/mappers.py modified: netbox_services/engine/resolver.py new file: netbox_services/engine/templating.py modified: netbox_services/forms.py modified: netbox_services/models.py modified: netbox_services/tests/test_allocators.py new file: netbox_services/tests/test_mappers.py modified: netbox_services/tests/test_resolver.py new file: netbox_services/tests/test_templating.py Simplify layout to template and service instances
258 lines
8.0 KiB
Python
258 lines
8.0 KiB
Python
import json
|
|
|
|
from django.contrib.contenttypes.fields import GenericForeignKey
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.exceptions import ValidationError
|
|
from django.db import models
|
|
from django.urls import reverse
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from core.models import ObjectType
|
|
from netbox.models import NetBoxModel, PrimaryModel
|
|
from utilities.choices import ChoiceSet
|
|
|
|
|
|
class ServiceTemplateStatusChoices(ChoiceSet):
|
|
key = "ServiceTemplate.status"
|
|
|
|
STATUS_DRAFT = "draft"
|
|
STATUS_ACTIVE = "active"
|
|
STATUS_DISCONTINUED = "discontinued"
|
|
|
|
CHOICES = [
|
|
(STATUS_DRAFT, _("Draft"), "cyan"),
|
|
(STATUS_ACTIVE, _("Active"), "green"),
|
|
(STATUS_DISCONTINUED, _("Discontinued"), "gray"),
|
|
]
|
|
|
|
|
|
class ServiceStatusChoices(ChoiceSet):
|
|
key = "Service.status"
|
|
|
|
STATUS_PLANNED = "planned"
|
|
STATUS_ACTIVE = "active"
|
|
STATUS_DISCONTINUED = "discontinued"
|
|
|
|
CHOICES = [
|
|
(STATUS_PLANNED, _("Planned"), "cyan"),
|
|
(STATUS_ACTIVE, _("Active"), "green"),
|
|
(STATUS_DISCONTINUED, _("Discontinued"), "gray"),
|
|
]
|
|
|
|
|
|
class RequirementScopeChoices(ChoiceSet):
|
|
key = "ServiceRequirement.scope"
|
|
|
|
SCOPE_SERVICE = "service"
|
|
SCOPE_PARENT = "parent"
|
|
|
|
CHOICES = [
|
|
(SCOPE_SERVICE, _("Service"), "blue"),
|
|
(SCOPE_PARENT, _("Parent"), "green"),
|
|
]
|
|
|
|
|
|
class ResolutionModeChoices(ChoiceSet):
|
|
key = "Resolution.mode"
|
|
|
|
MODE_DRY_RUN = "dry_run"
|
|
MODE_COMMIT = "commit"
|
|
|
|
CHOICES = [
|
|
(MODE_DRY_RUN, _("Dry Run"), "cyan"),
|
|
(MODE_COMMIT, _("Commit"), "green"),
|
|
]
|
|
|
|
|
|
def validate_tree(tree):
|
|
"""Validate the declarative requirement tree structure."""
|
|
if not isinstance(tree, list):
|
|
raise ValidationError({"tree": _("Tree must be a list of requirement nodes.")})
|
|
|
|
seen_names = set()
|
|
|
|
def _validate_node(node, path="root"):
|
|
if not isinstance(node, dict):
|
|
raise ValidationError({"tree": _(f"Node at {path} must be a dict.")})
|
|
|
|
name = node.get("name")
|
|
if not name:
|
|
raise ValidationError({"tree": _(f"Node at {path} must have a 'name'.")})
|
|
|
|
node_path = f"{path}/{name}"
|
|
if name in seen_names:
|
|
raise ValidationError({"tree": _(f"Duplicate node name '{name}'.")})
|
|
seen_names.add(name)
|
|
|
|
object_type = node.get("object_type")
|
|
if not object_type or not isinstance(object_type, str):
|
|
raise ValidationError({"tree": _(f"Node '{name}' must have an 'object_type' string (e.g. 'dcim.device').")})
|
|
|
|
parts = object_type.split(".")
|
|
if len(parts) != 2:
|
|
raise ValidationError({"tree": _(f"Node '{name}': object_type must be 'app_label.model_name'.")})
|
|
|
|
try:
|
|
ot = ObjectType.objects.get_by_natural_key(parts[0], parts[1])
|
|
except ObjectType.DoesNotExist:
|
|
raise ValidationError({"tree": _(f"Node '{name}': object_type '{object_type}' not found.")})
|
|
|
|
if ot.model_class() is None:
|
|
raise ValidationError({"tree": _(f"Node '{name}': object_type '{object_type}' has no model class.")})
|
|
|
|
quantity = node.get("quantity", 1)
|
|
if not isinstance(quantity, int) or quantity < 1:
|
|
raise ValidationError({"tree": _(f"Node '{name}': quantity must be a positive integer.")})
|
|
|
|
selector = node.get("selector")
|
|
allocation = node.get("allocation")
|
|
if not selector and not allocation:
|
|
raise ValidationError({"tree": _(f"Node '{name}': must have either 'selector' or 'allocation'.")})
|
|
if selector and not isinstance(selector, dict):
|
|
raise ValidationError({"tree": _(f"Node '{name}': selector must be a dict.")})
|
|
if allocation and not isinstance(allocation, dict):
|
|
raise ValidationError({"tree": _(f"Node '{name}': allocation must be a dict.")})
|
|
if allocation and "attrs" in allocation and not isinstance(allocation["attrs"], dict):
|
|
raise ValidationError({"tree": _(f"Node '{name}': allocation.attrs must be a dict.")})
|
|
|
|
scope = node.get("scope", "parent")
|
|
if scope not in RequirementScopeChoices.values():
|
|
raise ValidationError({"tree": _(f"Node '{name}': scope must be 'service' or 'parent'.")})
|
|
|
|
children = node.get("children", [])
|
|
if not isinstance(children, list):
|
|
raise ValidationError({"tree": _(f"Node '{name}': children must be a list.")})
|
|
for i, child in enumerate(children):
|
|
_validate_node(child, node_path)
|
|
|
|
for i, node in enumerate(tree):
|
|
_validate_node(node, f"root[{i}]")
|
|
|
|
|
|
class ServiceTemplate(PrimaryModel):
|
|
name = models.CharField(
|
|
verbose_name=_("name"),
|
|
max_length=100,
|
|
unique=True,
|
|
)
|
|
status = models.CharField(
|
|
verbose_name=_("status"),
|
|
max_length=20,
|
|
choices=ServiceTemplateStatusChoices,
|
|
default=ServiceTemplateStatusChoices.STATUS_DRAFT,
|
|
)
|
|
tree = models.JSONField(
|
|
verbose_name=_("requirement tree"),
|
|
default=list,
|
|
blank=True,
|
|
)
|
|
owner = models.ForeignKey(
|
|
to="users.Owner",
|
|
on_delete=models.PROTECT,
|
|
blank=True,
|
|
null=True,
|
|
related_name="netbox_services_templates",
|
|
)
|
|
|
|
clone_fields = ("description", "comments", "tree")
|
|
|
|
class Meta:
|
|
ordering = ("name",)
|
|
verbose_name = _("service template")
|
|
verbose_name_plural = _("service templates")
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
def get_absolute_url(self):
|
|
from utilities.views import get_viewname
|
|
return reverse(get_viewname(self), args=[self.pk])
|
|
|
|
@property
|
|
def is_locked(self):
|
|
return self.services.exists()
|
|
|
|
def clean(self):
|
|
super().clean()
|
|
if not getattr(self, "_tree_validated_by_form", False):
|
|
try:
|
|
validate_tree(self.tree)
|
|
except ValidationError as e:
|
|
raise ValidationError(e.messages)
|
|
|
|
|
|
class Service(PrimaryModel):
|
|
service_template = models.ForeignKey(
|
|
to=ServiceTemplate,
|
|
on_delete=models.PROTECT,
|
|
related_name="services",
|
|
)
|
|
name = models.CharField(
|
|
verbose_name=_("name"),
|
|
max_length=100,
|
|
unique=True,
|
|
)
|
|
tenant = models.ForeignKey(
|
|
to="tenancy.Tenant",
|
|
on_delete=models.PROTECT,
|
|
related_name="netbox_services_services",
|
|
verbose_name=_("tenant"),
|
|
)
|
|
status = models.CharField(
|
|
verbose_name=_("status"),
|
|
max_length=20,
|
|
choices=ServiceStatusChoices,
|
|
default=ServiceStatusChoices.STATUS_PLANNED,
|
|
)
|
|
owner = models.ForeignKey(
|
|
to="users.Owner",
|
|
on_delete=models.PROTECT,
|
|
blank=True,
|
|
null=True,
|
|
related_name="netbox_services_services",
|
|
)
|
|
|
|
clone_fields = ("service_template", "tenant", "description", "comments")
|
|
|
|
class Meta:
|
|
ordering = ("name",)
|
|
verbose_name = _("service")
|
|
verbose_name_plural = _("services")
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
def get_absolute_url(self):
|
|
from utilities.views import get_viewname
|
|
return reverse(get_viewname(self), args=[self.pk])
|
|
|
|
|
|
class ServiceResourceBinding(NetBoxModel):
|
|
_netbox_private = True
|
|
|
|
service = models.ForeignKey(
|
|
to=Service,
|
|
on_delete=models.CASCADE,
|
|
related_name="bindings",
|
|
)
|
|
content_type = models.ForeignKey(
|
|
to=ContentType,
|
|
on_delete=models.PROTECT,
|
|
related_name="+",
|
|
)
|
|
object_id = models.PositiveBigIntegerField()
|
|
content_object = GenericForeignKey("content_type", "object_id")
|
|
|
|
class Meta:
|
|
ordering = ("service", "content_type", "object_id")
|
|
verbose_name = _("service resource binding")
|
|
verbose_name_plural = _("service resource bindings")
|
|
indexes = [
|
|
models.Index(fields=["content_type", "object_id"]),
|
|
]
|
|
|
|
def __str__(self):
|
|
obj = self.content_object
|
|
label = str(obj) if obj is not None else f"{self.content_type} #{self.object_id}"
|
|
return f"{self.service} → {label}"
|