70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from django import forms
|
|
|
|
from .client import CatalogPlugin
|
|
from .runtime import RuntimeSettings
|
|
|
|
|
|
class LifecycleConfirmForm(forms.Form):
|
|
confirmation = forms.CharField(
|
|
max_length=64,
|
|
label="Plugin-Slug zur Bestätigung",
|
|
help_text="Diese Eingabe verhindert versehentliche Lifecycle-Aktionen.",
|
|
)
|
|
version = forms.ChoiceField(required=False, label="Version")
|
|
dry_run = forms.BooleanField(
|
|
required=False,
|
|
initial=True,
|
|
label="Nur prüfen (Dry-Run)",
|
|
help_text="Plant und validiert den Vorgang, ohne Dateien oder Prozesse zu verändern.",
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
*args,
|
|
plugin: CatalogPlugin,
|
|
action: str,
|
|
runtime: RuntimeSettings,
|
|
**kwargs,
|
|
):
|
|
super().__init__(*args, **kwargs)
|
|
self.plugin = plugin
|
|
self.action = action
|
|
self.runtime = runtime
|
|
self.fields["confirmation"].widget.attrs.update({"autocomplete": "off", "placeholder": plugin.slug})
|
|
self.fields["dry_run"].initial = runtime.default_dry_run
|
|
if action in {"install", "update"}:
|
|
choices = [
|
|
(release.version, release.version)
|
|
for release in plugin.releases
|
|
if (
|
|
release.supports(runtime.netbox_version, plugin)
|
|
and release.approved
|
|
and release.immutable
|
|
and bool(release.sha256)
|
|
and (runtime.execution_mode != "agent" or bool(release.approved_payload_sha256))
|
|
)
|
|
]
|
|
choices.sort(reverse=True)
|
|
self.fields["version"].choices = choices
|
|
self.fields["version"].required = True
|
|
self.fields["version"].initial = plugin.latest_version
|
|
else:
|
|
self.fields.pop("version")
|
|
for field in self.fields.values():
|
|
if not isinstance(field.widget, forms.CheckboxInput):
|
|
field.widget.attrs.setdefault("class", "form-control")
|
|
|
|
def clean_confirmation(self) -> str:
|
|
value = self.cleaned_data["confirmation"]
|
|
if value != self.plugin.slug:
|
|
raise forms.ValidationError("Der Slug stimmt nicht exakt überein.")
|
|
return value
|
|
|
|
def clean(self):
|
|
cleaned = super().clean()
|
|
if self.runtime.execution_mode == "dry_run" and not cleaned.get("dry_run", False):
|
|
self.add_error("dry_run", "Reale Aktionen sind in execution_mode=dry_run deaktiviert.")
|
|
return cleaned
|