Coverage for dcim/forms/bulk

1from django import forms

2from django.contrib.contenttypes.models import ContentType

3from django.contrib.postgres.forms.array import SimpleArrayField

4from django.core.exceptions import NON_FIELD_ERRORS, MultipleObjectsReturned, ObjectDoesNotExist

5from django.utils.functional import lazy

6from django.utils.html import format_html

7from django.utils.safestring import SafeString, mark_safe

8from django.utils.translation import gettext_lazy as _

10from dcim.choices import *

11from dcim.constants import *

12from dcim.models import *

13from dcim.utils import reconcile_port_mappings

14from extras.models import ConfigTemplate

15from ipam.choices import VLANQinQRoleChoices

16from ipam.models import VLAN, VRF, IPAddress, VLANGroup

17from netbox.choices import *

18from netbox.forms import (

19 NestedGroupModelImportForm,

20 NetBoxModelImportForm,

21 OrganizationalModelImportForm,

22 OwnerCSVMixin,

23 PrimaryModelImportForm,

24)

25from tenancy.models import Tenant

26from utilities.forms.fields import (

27 CSVChoiceField,

28 CSVContentTypeField,

29 CSVModelChoiceField,

30 CSVModelMultipleChoiceField,

31 CSVTypedChoiceField,

32 SlugField,

33)

34from virtualization.models import Cluster, VirtualMachine, VMInterface

35from wireless.choices import WirelessRoleChoices

37from .common import ModuleCommonForm

39__all__ = (

40 'CableBundleImportForm',

41 'CableImportForm',

42 'ConsolePortImportForm',

43 'ConsoleServerPortImportForm',

44 'CoolingFeedImportForm',

45 'CoolingIntakeImportForm',

46 'CoolingOutflowImportForm',

47 'CoolingSourceImportForm',

48 'DeviceBayImportForm',

49 'DeviceImportForm',

50 'DeviceRoleImportForm',

51 'DeviceTypeImportForm',

52 'FrontPortImportForm',

53 'InterfaceImportForm',

54 'InventoryItemImportForm',

55 'InventoryItemRoleImportForm',

56 'LocationImportForm',

57 'MACAddressImportForm',

58 'ManufacturerImportForm',

59 'ModuleBayImportForm',

60 'ModuleBayTypeImportForm',

61 'ModuleImportForm',

62 'ModuleTypeImportForm',

63 'ModuleTypeProfileImportForm',

64 'PlatformImportForm',

65 'PowerFeedImportForm',

66 'PowerOutletImportForm',

67 'PowerPanelImportForm',

68 'PowerPortImportForm',

69 'RackGroupImportForm',

70 'RackImportForm',

71 'RackReservationImportForm',

72 'RackRoleImportForm',

73 'RackTypeImportForm',

74 'RearPortImportForm',

75 'RegionImportForm',

76 'SiteGroupImportForm',

77 'SiteImportForm',

78 'VirtualChassisImportForm',

79 'VirtualDeviceContextImportForm'

80)

82# A lazily evaluated format_html(), for help text which must not resolve its translated content until

83# the field is rendered. Unlike mark_safe(), this escapes the interpolated arguments.

84format_html_lazy = lazy(format_html, SafeString)

87class RegionImportForm(NestedGroupModelImportForm):

88 parent = CSVModelChoiceField(

89 label=_('Parent'),

90 queryset=Region.objects.all(),

91 required=False,

92 to_field_name='name',

93 help_text=_('Name of parent region')

94 )

96 class Meta:

97 model = Region

98 fields = ('name', 'slug', 'parent', 'description', 'owner', 'comments', 'tags')

100

101class SiteGroupImportForm(NestedGroupModelImportForm):

102 parent = CSVModelChoiceField(

103 label=_('Parent'),

104 queryset=SiteGroup.objects.all(),

105 required=False,

106 to_field_name='name',

107 help_text=_('Name of parent site group')

108 )

109

110 class Meta:

111 model = SiteGroup

112 fields = ('name', 'slug', 'parent', 'description', 'owner', 'comments', 'tags')

113

114

115class SiteImportForm(PrimaryModelImportForm):

116 status = CSVChoiceField(

117 label=_('Status'),

118 choices=SiteStatusChoices,

119 help_text=_('Operational status')

120 )

121 region = CSVModelChoiceField(

122 label=_('Region'),

123 queryset=Region.objects.all(),

124 required=False,

125 to_field_name='name',

126 help_text=_('Assigned region')

127 )

128 group = CSVModelChoiceField(

129 label=_('Group'),

130 queryset=SiteGroup.objects.all(),

131 required=False,

132 to_field_name='name',

133 help_text=_('Assigned group')

134 )

135 tenant = CSVModelChoiceField(

136 label=_('Tenant'),

137 queryset=Tenant.objects.all(),

138 required=False,

139 to_field_name='name',

140 help_text=_('Assigned tenant')

141 )

142

143 class Meta:

144 model = Site

145 fields = (

146 'name', 'slug', 'status', 'region', 'group', 'tenant', 'facility', 'time_zone', 'description',

147 'physical_address', 'shipping_address', 'latitude', 'longitude', 'owner', 'comments', 'tags'

148 )

149 help_texts = {

150 'time_zone': mark_safe(

151 '{} (<a href="https://en.wikipedia.org/wiki/List_of_tz_database_time_zones">{}</a>)'.format(

152 _('Time zone'), _('available options')

153 )

154 )

155 }

156

157

158class LocationImportForm(NestedGroupModelImportForm):

159 site = CSVModelChoiceField(

160 label=_('Site'),

161 queryset=Site.objects.all(),

162 to_field_name='name',

163 help_text=_('Assigned site')

164 )

165 parent = CSVModelChoiceField(

166 label=_('Parent'),

167 queryset=Location.objects.all(),

168 required=False,

169 to_field_name='name',

170 help_text=_('Parent location'),

171 error_messages={

172 'invalid_choice': _('Location not found.'),

173 }

174 )

175 status = CSVChoiceField(

176 label=_('Status'),

177 choices=LocationStatusChoices,

178 help_text=_('Operational status')

179 )

180 tenant = CSVModelChoiceField(

181 label=_('Tenant'),

182 queryset=Tenant.objects.all(),

183 required=False,

184 to_field_name='name',

185 help_text=_('Assigned tenant')

186 )

187

188 class Meta:

189 model = Location

190 fields = (

191 'site', 'parent', 'name', 'slug', 'status', 'tenant', 'facility', 'description', 'owner', 'comments',

192 'tags',

193 )

194

195 def __init__(self, data=None, *args, **kwargs):

196 super().__init__(data, *args, **kwargs)

197

198 if data:

199 # Limit location queryset by assigned site

200 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

201 self.fields['parent'].queryset = self.fields['parent'].queryset.filter(**params)

202

203

204class RackGroupImportForm(OrganizationalModelImportForm):

205

206 class Meta:

207 model = RackGroup

208 fields = ('name', 'slug', 'description', 'owner', 'comments', 'tags')

209

210

211class RackRoleImportForm(OrganizationalModelImportForm):

212

213 class Meta:

214 model = RackRole

215 fields = ('name', 'slug', 'color', 'description', 'owner', 'comments', 'tags')

216

217

218class RackTypeImportForm(PrimaryModelImportForm):

219 manufacturer = forms.ModelChoiceField(

220 label=_('Manufacturer'),

221 queryset=Manufacturer.objects.all(),

222 to_field_name='name',

223 help_text=_('The manufacturer of this rack type')

224 )

225 form_factor = CSVChoiceField(

226 label=_('Type'),

227 choices=RackFormFactorChoices,

228 help_text=_('Form factor')

229 )

230 starting_unit = forms.IntegerField(

231 required=False,

232 min_value=1,

233 help_text=_('The lowest-numbered position in the rack')

234 )

235 width = forms.ChoiceField(

236 label=_('Width'),

237 choices=RackWidthChoices,

238 help_text=_('Rail-to-rail width (in inches)')

239 )

240 outer_unit = CSVChoiceField(

241 label=_('Outer unit'),

242 choices=RackDimensionUnitChoices,

243 required=False,

244 help_text=_('Unit for outer dimensions')

245 )

246 weight_unit = CSVChoiceField(

247 label=_('Weight unit'),

248 choices=WeightUnitChoices,

249 required=False,

250 help_text=_('Unit for rack weights')

251 )

252 cooling_capability = CSVChoiceField(

253 label=_('Cooling capability'),

254 choices=RackCoolingCapabilityChoices,

255 required=False,

256 help_text=_('Cooling capability')

257 )

258

259 class Meta:

260 model = RackType

261 fields = (

262 'manufacturer', 'model', 'slug', 'form_factor', 'width', 'u_height', 'starting_unit', 'desc_units',

263 'outer_width', 'outer_height', 'outer_depth', 'outer_unit', 'mounting_depth', 'weight', 'max_weight',

264 'weight_unit', 'cooling_capability', 'cooling_capacity', 'description', 'owner', 'comments', 'tags',

265 )

266

267 def __init__(self, data=None, *args, **kwargs):

268 super().__init__(data, *args, **kwargs)

269

270

271class RackImportForm(PrimaryModelImportForm):

272 site = CSVModelChoiceField(

273 label=_('Site'),

274 queryset=Site.objects.all(),

275 to_field_name='name'

276 )

277 location = CSVModelChoiceField(

278 label=_('Location'),

279 queryset=Location.objects.all(),

280 required=False,

281 to_field_name='name'

282 )

283 tenant = CSVModelChoiceField(

284 label=_('Tenant'),

285 queryset=Tenant.objects.all(),

286 required=False,

287 to_field_name='name',

288 help_text=_('Name of assigned tenant')

289 )

290 group = CSVModelChoiceField(

291 label=_('Rack group'),

292 queryset=RackGroup.objects.all(),

293 required=False,

294 to_field_name='name',

295 help_text=_('Name of assigned group')

296 )

297 status = CSVChoiceField(

298 label=_('Status'),

299 choices=RackStatusChoices,

300 help_text=_('Operational status')

301 )

302 role = CSVModelChoiceField(

303 label=_('Role'),

304 queryset=RackRole.objects.all(),

305 required=False,

306 to_field_name='name',

307 help_text=_('Name of assigned role')

308 )

309 rack_type = CSVModelChoiceField(

310 label=_('Rack type'),

311 queryset=RackType.objects.all(),

312 to_field_name='model',

313 required=False,

314 help_text=_('Rack type model')

315 )

316 form_factor = CSVChoiceField(

317 label=_('Type'),

318 choices=RackFormFactorChoices,

319 required=False,

320 help_text=_('Form factor')

321 )

322 width = forms.ChoiceField(

323 label=_('Width'),

324 choices=RackWidthChoices,

325 required=False,

326 help_text=_('Rail-to-rail width (in inches)')

327 )

328 u_height = forms.IntegerField(

329 required=False,

330 label=_('Height (U)')

331 )

332 outer_unit = CSVChoiceField(

333 label=_('Outer unit'),

334 choices=RackDimensionUnitChoices,

335 required=False,

336 help_text=_('Unit for outer dimensions')

337 )

338 airflow = CSVChoiceField(

339 label=_('Airflow'),

340 choices=RackAirflowChoices,

341 required=False,

342 help_text=_('Airflow direction')

343 )

344 cooling_capability = CSVChoiceField(

345 label=_('Cooling capability'),

346 choices=RackCoolingCapabilityChoices,

347 required=False,

348 help_text=_('Cooling capability')

349 )

350 weight_unit = CSVChoiceField(

351 label=_('Weight unit'),

352 choices=WeightUnitChoices,

353 required=False,

354 help_text=_('Unit for rack weights')

355 )

356

357 class Meta:

358 model = Rack

359 fields = (

360 'site', 'location', 'group', 'name', 'facility_id', 'tenant', 'status', 'role', 'rack_type', 'form_factor',

361 'serial', 'asset_tag', 'width', 'u_height', 'desc_units', 'outer_width', 'outer_height', 'outer_depth',

362 'outer_unit', 'mounting_depth', 'airflow', 'cooling_capability', 'cooling_capacity', 'weight',

363 'max_weight', 'weight_unit', 'description', 'owner', 'comments', 'tags',

364 )

365

366 def __init__(self, data=None, *args, **kwargs):

367 super().__init__(data, *args, **kwargs)

368

369 if data:

370

371 # Limit location queryset by assigned site

372 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

373 self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)

374

375 def clean(self):

376 super().clean()

377

378 # width & u_height must be set if not specifying a rack type on import

379 if not self.instance.pk:

380 if not self.cleaned_data.get('rack_type') and not self.cleaned_data.get('width'):

381 raise forms.ValidationError(_("Width must be set if not specifying a rack type."))

382 if not self.cleaned_data.get('rack_type') and not self.cleaned_data.get('u_height'):

383 raise forms.ValidationError(_("U height must be set if not specifying a rack type."))

384

385

386class RackReservationImportForm(PrimaryModelImportForm):

387 site = CSVModelChoiceField(

388 label=_('Site'),

389 queryset=Site.objects.all(),

390 to_field_name='name',

391 help_text=_('Parent site')

392 )

393 location = CSVModelChoiceField(

394 label=_('Location'),

395 queryset=Location.objects.all(),

396 to_field_name='name',

397 required=False,

398 help_text=_("Rack's location (if any)")

399 )

400 rack = CSVModelChoiceField(

401 label=_('Rack'),

402 queryset=Rack.objects.all(),

403 to_field_name='name',

404 help_text=_('Rack')

405 )

406 units = SimpleArrayField(

407 label=_('Units'),

408 base_field=forms.IntegerField(),

409 required=True,

410 help_text=_('Comma-separated list of individual unit numbers')

411 )

412 status = CSVChoiceField(

413 label=_('Status'),

414 choices=RackReservationStatusChoices,

415 help_text=_('Operational status')

416 )

417 tenant = CSVModelChoiceField(

418 label=_('Tenant'),

419 queryset=Tenant.objects.all(),

420 required=False,

421 to_field_name='name',

422 help_text=_('Assigned tenant')

423 )

424

425 class Meta:

426 model = RackReservation

427 fields = ('site', 'location', 'rack', 'units', 'status', 'tenant', 'description', 'owner', 'comments', 'tags')

428

429 def __init__(self, data=None, *args, **kwargs):

430 super().__init__(data, *args, **kwargs)

431

432 if data:

433

434 # Limit location queryset by assigned site

435 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

436 self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)

437

438 # Limit rack queryset by assigned site and group

439 params = {

440 f"site__{self.fields['site'].to_field_name}": data.get('site'),

441 f"location__{self.fields['location'].to_field_name}": data.get('location'),

442 }

443 self.fields['rack'].queryset = self.fields['rack'].queryset.filter(**params)

444

445

446class ManufacturerImportForm(OrganizationalModelImportForm):

447

448 class Meta:

449 model = Manufacturer

450 fields = ('name', 'slug', 'description', 'owner', 'comments', 'tags')

451

452

453class DeviceTypeImportForm(PrimaryModelImportForm):

454 manufacturer = CSVModelChoiceField(

455 label=_('Manufacturer'),

456 queryset=Manufacturer.objects.all(),

457 to_field_name='name',

458 help_text=_('The manufacturer which produces this device type')

459 )

460 default_platform = CSVModelChoiceField(

461 label=_('Default platform'),

462 queryset=Platform.objects.all(),

463 to_field_name='name',

464 required=False,

465 help_text=_('The default platform for devices of this type (optional)')

466 )

467 weight = forms.DecimalField(

468 label=_('Weight'),

469 required=False,

470 help_text=_('Device weight'),

471 )

472 weight_unit = CSVChoiceField(

473 label=_('Weight unit'),

474 choices=WeightUnitChoices,

475 required=False,

476 help_text=_('Unit for device weight')

477 )

478

479 class Meta:

480 model = DeviceType

481 fields = [

482 'manufacturer', 'default_platform', 'model', 'slug', 'part_number', 'u_height', 'exclude_from_utilization',

483 'is_full_depth', 'subdevice_role', 'airflow', 'cooling_method', 'description', 'weight', 'weight_unit',

484 'end_of_life', 'owner', 'comments', 'tags',

485 ]

486

487

488class ModuleBayTypeImportForm(PrimaryModelImportForm):

489 manufacturer = CSVModelChoiceField(

490 label=_('Manufacturer'),

491 queryset=Manufacturer.objects.all(),

492 to_field_name='name',

493 required=False,

494 )

495

496 class Meta:

497 model = ModuleBayType

498 fields = [

499 'name', 'slug', 'manufacturer', 'color', 'description', 'owner', 'comments', 'tags',

500 ]

501

502

503class ModuleTypeProfileImportForm(PrimaryModelImportForm):

504

505 class Meta:

506 model = ModuleTypeProfile

507 fields = [

508 'name', 'description', 'schema', 'owner', 'comments', 'tags',

509 ]

510

511

512class ModuleTypeImportForm(PrimaryModelImportForm):

513 profile = forms.ModelChoiceField(

514 label=_('Profile'),

515 queryset=ModuleTypeProfile.objects.all(),

516 to_field_name='name',

517 required=False

518 )

519 manufacturer = forms.ModelChoiceField(

520 label=_('Manufacturer'),

521 queryset=Manufacturer.objects.all(),

522 to_field_name='name'

523 )

524 airflow = CSVChoiceField(

525 label=_('Airflow'),

526 choices=ModuleAirflowChoices,

527 required=False,

528 help_text=_('Airflow direction')

529 )

530 cooling_method = CSVChoiceField(

531 label=_('Cooling method'),

532 choices=CoolingMethodChoices,

533 required=False,

534 help_text=_('Cooling method')

535 )

536 weight = forms.DecimalField(

537 label=_('Weight'),

538 required=False,

539 help_text=_('Module weight'),

540 )

541 weight_unit = CSVChoiceField(

542 label=_('Weight unit'),

543 choices=WeightUnitChoices,

544 required=False,

545 help_text=_('Unit for module weight')

546 )

547 attribute_data = forms.JSONField(

548 label=_('Attributes'),

549 required=False,

550 help_text=_('Attribute values for the assigned profile, passed as a dictionary')

551 )

552

553 class Meta:

554 model = ModuleType

555 fields = [

556 'manufacturer', 'model', 'part_number', 'description', 'cooling_method', 'airflow', 'weight', 'weight_unit',

557 'end_of_life', 'profile', 'attribute_data', 'owner', 'comments', 'tags',

558 ]

559

560 def clean(self):

561 super().clean()

562

563 # Attribute data may be included only if a profile is specified

564 if self.cleaned_data.get('attribute_data') and not self.cleaned_data.get('profile'):

565 raise forms.ValidationError(_("Profile must be specified if attribute data is provided."))

566

567 # Default attribute_data to an empty dictionary if a profile is specified (to enforce schema validation)

568 if self.cleaned_data.get('profile') and not self.cleaned_data.get('attribute_data'):

569 self.cleaned_data['attribute_data'] = {}

570

571

572class DeviceRoleImportForm(NestedGroupModelImportForm):

573 parent = CSVModelChoiceField(

574 label=_('Parent'),

575 queryset=DeviceRole.objects.all(),

576 required=False,

577 to_field_name='name',

578 help_text=_('Parent Device Role'),

579 error_messages={

580 'invalid_choice': _('Device role not found.'),

581 }

582 )

583 config_template = CSVModelChoiceField(

584 label=_('Config template'),

585 queryset=ConfigTemplate.objects.all(),

586 to_field_name='name',

587 required=False,

588 help_text=_('Config template')

589 )

590

591 class Meta:

592 model = DeviceRole

593 fields = (

594 'name', 'slug', 'parent', 'color', 'vm_role', 'config_template', 'description', 'owner', 'comments', 'tags'

595 )

596

597

598class PlatformImportForm(NestedGroupModelImportForm):

599 parent = CSVModelChoiceField(

600 label=_('Parent'),

601 queryset=Platform.objects.all(),

602 required=False,

603 to_field_name='name',

604 help_text=_('Parent platform'),

605 error_messages={

606 'invalid_choice': _('Platform not found.'),

607 }

608 )

609 manufacturer = CSVModelChoiceField(

610 label=_('Manufacturer'),

611 queryset=Manufacturer.objects.all(),

612 required=False,

613 to_field_name='name',

614 help_text=_('Limit platform assignments to this manufacturer')

615 )

616 config_template = CSVModelChoiceField(

617 label=_('Config template'),

618 queryset=ConfigTemplate.objects.all(),

619 to_field_name='name',

620 required=False,

621 help_text=_('Config template')

622 )

623

624 class Meta:

625 model = Platform

626 fields = (

627 'name', 'slug', 'parent', 'manufacturer', 'config_template', 'description', 'owner', 'comments', 'tags',

628 )

629

630

631class BaseDeviceImportForm(PrimaryModelImportForm):

632 role = CSVModelChoiceField(

633 label=_('Device role'),

634 queryset=DeviceRole.objects.all(),

635 to_field_name='name',

636 help_text=_('Assigned role')

637 )

638 tenant = CSVModelChoiceField(

639 label=_('Tenant'),

640 queryset=Tenant.objects.all(),

641 required=False,

642 to_field_name='name',

643 help_text=_('Assigned tenant')

644 )

645 manufacturer = CSVModelChoiceField(

646 label=_('Manufacturer'),

647 queryset=Manufacturer.objects.all(),

648 to_field_name='name',

649 help_text=_('Device type manufacturer')

650 )

651 device_type = CSVModelChoiceField(

652 label=_('Device type'),

653 queryset=DeviceType.objects.all(),

654 to_field_name='model',

655 help_text=_('Device type model')

656 )

657 platform = CSVModelChoiceField(

658 label=_('Platform'),

659 queryset=Platform.objects.all(),

660 required=False,

661 to_field_name='name',

662 help_text=_('Assigned platform')

663 )

664 status = CSVChoiceField(

665 label=_('Status'),

666 choices=DeviceStatusChoices,

667 help_text=_('Operational status')

668 )

669 virtual_chassis = CSVModelChoiceField(

670 label=_('Virtual chassis'),

671 queryset=VirtualChassis.objects.all(),

672 to_field_name='name',

673 required=False,

674 help_text=_('Virtual chassis')

675 )

676 cluster = CSVModelChoiceField(

677 label=_('Cluster'),

678 queryset=Cluster.objects.all(),

679 to_field_name='name',

680 required=False,

681 help_text=_('Virtualization cluster')

682 )

683

684 class Meta:

685 fields = []

686 model = Device

687

688 def __init__(self, data=None, *args, **kwargs):

689 super().__init__(data, *args, **kwargs)

690

691 if data:

692

693 # Limit device type queryset by manufacturer

694 params = {f"manufacturer__{self.fields['manufacturer'].to_field_name}": data.get('manufacturer')}

695 self.fields['device_type'].queryset = self.fields['device_type'].queryset.filter(**params)

696

697

698class DeviceImportForm(BaseDeviceImportForm):

699 site = CSVModelChoiceField(

700 label=_('Site'),

701 queryset=Site.objects.all(),

702 to_field_name='name',

703 help_text=_('Assigned site')

704 )

705 location = CSVModelChoiceField(

706 label=_('Location'),

707 queryset=Location.objects.all(),

708 to_field_name='name',

709 required=False,

710 help_text=_("Assigned location (if any)")

711 )

712 rack = CSVModelChoiceField(

713 label=_('Rack'),

714 queryset=Rack.objects.all(),

715 to_field_name='name',

716 required=False,

717 help_text=_("Assigned rack (if any)")

718 )

719 face = CSVChoiceField(

720 label=_('Face'),

721 choices=DeviceFaceChoices,

722 required=False,

723 help_text=_('Mounted rack face')

724 )

725 parent = CSVModelChoiceField(

726 label=_('Parent'),

727 queryset=Device.objects.all(),

728 to_field_name='name',

729 required=False,

730 help_text=_('Parent device (for child devices)')

731 )

732 device_bay = CSVModelChoiceField(

733 label=_('Device bay'),

734 queryset=DeviceBay.objects.all(),

735 to_field_name='name',

736 required=False,

737 help_text=_('Device bay in which this device is installed (for child devices)')

738 )

739 airflow = CSVChoiceField(

740 label=_('Airflow'),

741 choices=DeviceAirflowChoices,

742 required=False,

743 help_text=_('Airflow direction')

744 )

745 cooling_method = CSVChoiceField(

746 label=_('Cooling method'),

747 choices=CoolingMethodChoices,

748 required=False,

749 help_text=_('Cooling method')

750 )

751 config_template = CSVModelChoiceField(

752 label=_('Config template'),

753 queryset=ConfigTemplate.objects.all(),

754 to_field_name='name',

755 required=False,

756 help_text=_('Config template')

757 )

758

759 class Meta(BaseDeviceImportForm.Meta):

760 fields = [

761 'name', 'role', 'tenant', 'manufacturer', 'device_type', 'platform', 'serial', 'asset_tag', 'status',

762 'site', 'location', 'rack', 'position', 'face', 'latitude', 'longitude', 'parent', 'device_bay', 'airflow',

763 'cooling_method', 'virtual_chassis', 'vc_position', 'vc_priority', 'cluster', 'description',

764 'config_template', 'owner', 'comments', 'tags',

765 ]

766

767 def __init__(self, data=None, *args, **kwargs):

768 super().__init__(data, *args, **kwargs)

769

770 if data:

771

772 # Limit location queryset by assigned site

773 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

774 self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)

775 self.fields['parent'].queryset = self.fields['parent'].queryset.filter(**params)

776

777 # Limit rack queryset by assigned site and location

778 params = {

779 f"site__{self.fields['site'].to_field_name}": data.get('site'),

780 }

781 if location := data.get('location'):

782 params.update({

783 f"location__{self.fields['location'].to_field_name}": location,

784 })

785 self.fields['rack'].queryset = self.fields['rack'].queryset.filter(**params)

786

787 # Limit platform queryset by manufacturer

788 params = {f"manufacturer__{self.fields['manufacturer'].to_field_name}": data.get('manufacturer')}

789 self.fields['platform'].queryset = self.fields['platform'].queryset.filter(

790 Q(**params) | Q(manufacturer=None)

791 )

792

793 # Limit device bay queryset by parent device

794 if parent := data.get('parent'):

795 params = {f"device__{self.fields['parent'].to_field_name}": parent}

796 self.fields['device_bay'].queryset = self.fields['device_bay'].queryset.filter(**params)

797

798 def clean(self):

799 super().clean()

800

801 # Inherit site and rack from parent device

802 if parent := self.cleaned_data.get('parent'):

803 self.instance.site = parent.site

804 self.instance.rack = parent.rack

805

806 # Set parent_bay reverse relationship

807 if device_bay := self.cleaned_data.get('device_bay'):

808 self.instance.parent_bay = device_bay

809

810

811class ModuleImportForm(ModuleCommonForm, PrimaryModelImportForm):

812 device = CSVModelChoiceField(

813 label=_('Device'),

814 queryset=Device.objects.all(),

815 to_field_name='name',

816 help_text=_('The device in which this module is installed')

817 )

818 module_bay = CSVModelChoiceField(

819 label=_('Module bay'),

820 queryset=ModuleBay.objects.all(),

821 to_field_name='name',

822 help_text=_('The module bay in which this module is installed')

823 )

824 module_type = CSVModelChoiceField(

825 label=_('Module type'),

826 queryset=ModuleType.objects.all(),

827 to_field_name='model',

828 help_text=_('The type of module')

829 )

830 status = CSVChoiceField(

831 label=_('Status'),

832 choices=ModuleStatusChoices,

833 help_text=_('Operational status')

834 )

835 replicate_components = forms.BooleanField(

836 label=_('Replicate components'),

837 required=False,

838 help_text=_('Automatically populate components associated with this module type (enabled by default)')

839 )

840 adopt_components = forms.BooleanField(

841 label=_('Adopt components'),

842 required=False,

843 help_text=_('Adopt already existing components')

844 )

845

846 class Meta:

847 model = Module

848 fields = (

849 'device', 'module_bay', 'module_type', 'serial', 'asset_tag', 'status', 'description', 'owner', 'comments',

850 'replicate_components', 'adopt_components', 'tags',

851 )

852

853 def __init__(self, data=None, *args, **kwargs):

854 super().__init__(data, *args, **kwargs)

855

856 if data:

857 # Limit module_bay queryset by assigned device

858 params = {f"device__{self.fields['device'].to_field_name}": data.get('device')}

859 self.fields['module_bay'].queryset = self.fields['module_bay'].queryset.filter(**params)

860

861 def clean_replicate_components(self):

862 # Make sure replicate_components is True when it's not included in the uploaded data

863 if 'replicate_components' not in self.data:

864 return True

865 return self.cleaned_data['replicate_components']

866

867

868#

869# Device components

870#

871

872class ConsolePortImportForm(OwnerCSVMixin, NetBoxModelImportForm):

873 device = CSVModelChoiceField(

874 label=_('Device'),

875 queryset=Device.objects.all(),

876 to_field_name='name'

877 )

878 type = CSVChoiceField(

879 label=_('Type'),

880 choices=ConsolePortTypeChoices,

881 required=False,

882 help_text=_('Port type')

883 )

884 speed = CSVTypedChoiceField(

885 label=_('Speed'),

886 choices=ConsolePortSpeedChoices,

887 coerce=int,

888 empty_value=None,

889 required=False,

890 help_text=_('Port speed in bps')

891 )

892

893 class Meta:

894 model = ConsolePort

895 fields = ('device', 'name', 'label', 'type', 'speed', 'mark_connected', 'description', 'owner', 'tags')

896

897

898class ConsoleServerPortImportForm(OwnerCSVMixin, NetBoxModelImportForm):

899 device = CSVModelChoiceField(

900 label=_('Device'),

901 queryset=Device.objects.all(),

902 to_field_name='name'

903 )

904 type = CSVChoiceField(

905 label=_('Type'),

906 choices=ConsolePortTypeChoices,

907 required=False,

908 help_text=_('Port type')

909 )

910 speed = CSVTypedChoiceField(

911 label=_('Speed'),

912 choices=ConsolePortSpeedChoices,

913 coerce=int,

914 empty_value=None,

915 required=False,

916 help_text=_('Port speed in bps')

917 )

918

919 class Meta:

920 model = ConsoleServerPort

921 fields = ('device', 'name', 'label', 'type', 'speed', 'mark_connected', 'description', 'owner', 'tags')

922

923

924class PowerPortImportForm(OwnerCSVMixin, NetBoxModelImportForm):

925 device = CSVModelChoiceField(

926 label=_('Device'),

927 queryset=Device.objects.all(),

928 to_field_name='name'

929 )

930 type = CSVChoiceField(

931 label=_('Type'),

932 choices=PowerPortTypeChoices,

933 required=False,

934 help_text=_('Port type')

935 )

936

937 class Meta:

938 model = PowerPort

939 fields = (

940 'device', 'name', 'label', 'type', 'mark_connected', 'maximum_draw', 'allocated_draw', 'description',

941 'owner', 'tags',

942 )

943

944

945class PowerOutletImportForm(OwnerCSVMixin, NetBoxModelImportForm):

946 device = CSVModelChoiceField(

947 label=_('Device'),

948 queryset=Device.objects.all(),

949 to_field_name='name'

950 )

951 type = CSVChoiceField(

952 label=_('Type'),

953 choices=PowerOutletTypeChoices,

954 required=False,

955 help_text=_('Outlet type')

956 )

957 power_port = CSVModelChoiceField(

958 label=_('Power port'),

959 queryset=PowerPort.objects.all(),

960 required=False,

961 to_field_name='name',

962 help_text=_('Local power port which feeds this outlet')

963 )

964 feed_leg = CSVChoiceField(

965 label=_('Feed leg'),

966 choices=PowerOutletFeedLegChoices,

967 required=False,

968 help_text=_('Electrical phase (for three-phase circuits)')

969 )

970

971 class Meta:

972 model = PowerOutlet

973 fields = (

974 'device', 'name', 'label', 'type', 'color', 'mark_connected', 'power_port', 'feed_leg', 'description',

975 'owner', 'tags',

976 )

977

978 def __init__(self, *args, **kwargs):

979 super().__init__(*args, **kwargs)

980

981 # Limit PowerPort choices to those belonging to this device (or VC master)

982 if self.is_bound and 'device' in self.data:

983 try:

984 device = self.fields['device'].to_python(self.data['device'])

985 except forms.ValidationError:

986 device = None

987 else:

988 try:

989 device = self.instance.device

990 except Device.DoesNotExist:

991 device = None

992

993 if device:

994 self.fields['power_port'].queryset = PowerPort.objects.filter(

995 device__in=[device, device.get_vc_master()]

996 )

997 else:

998 self.fields['power_port'].queryset = PowerPort.objects.none()

999

1000

1001class CoolingIntakeImportForm(OwnerCSVMixin, NetBoxModelImportForm):

1002 device = CSVModelChoiceField(

1003 label=_('Device'),

1004 queryset=Device.objects.all(),

1005 to_field_name='name'

1006 )

1007 type = CSVChoiceField(

1008 label=_('Type'),

1009 choices=CoolingConnectorTypeChoices,

1010 required=False,

1011 help_text=_('Physical connector type')

1012 )

1013 diameter_unit = CSVChoiceField(

1014 label=_('Diameter unit'),

1015 choices=DiameterUnitChoices,

1016 required=False,

1017 help_text=_('Diameter unit')

1018 )

1019 max_flow_unit = CSVChoiceField(

1020 label=_('Max flow unit'),

1021 choices=FlowRateUnitChoices,

1022 required=False,

1023 help_text=_('Unit for maximum flow')

1024 )

1025 cooling_outflow_device = CSVModelChoiceField(

1026 label=_('Cooling outflow device'),

1027 queryset=Device.objects.all(),

1028 required=False,

1029 to_field_name='name',

1030 help_text=_('Device bearing the upstream cooling outflow (defaults to this intake\'s device)')

1031 )

1032 cooling_outflow = CSVModelChoiceField(

1033 label=_('Cooling outflow'),

1034 queryset=CoolingOutflow.objects.all(),

1035 required=False,

1036 to_field_name='name',

1037 help_text=_('Upstream cooling outflow which feeds this intake')

1038 )

1039

1040 class Meta:

1041 model = CoolingIntake

1042 fields = (

1043 'device', 'name', 'label', 'type', 'diameter', 'diameter_unit', 'max_flow',

1044 'max_flow_unit', 'cooling_outflow_device', 'cooling_outflow', 'description', 'owner', 'tags',

1045 )

1046

1047 def __init__(self, *args, **kwargs):

1048 super().__init__(*args, **kwargs)

1049

1050 # The supplying outflow typically belongs to an upstream device (e.g. a CDU or manifold), so scope the

1051 # CoolingOutflow choices to cooling_outflow_device where given, falling back to this intake's own device.

1052 if self.is_bound and 'device' in self.data:

1053 try:

1054 device = self.fields['device'].to_python(self.data['device'])

1055 except forms.ValidationError:

1056 device = None

1057 else:

1058 try:

1059 device = self.instance.device

1060 except Device.DoesNotExist:

1061 device = None

1062

1063 outflow_device = None

1064 if self.is_bound and self.data.get('cooling_outflow_device'):

1065 try:

1066 outflow_device = self.fields['cooling_outflow_device'].to_python(

1067 self.data['cooling_outflow_device']

1068 )

1069 except forms.ValidationError:

1070 outflow_device = None

1071

1072 if outflow_device:

1073 self.fields['cooling_outflow'].queryset = CoolingOutflow.objects.filter(device=outflow_device)

1074 elif device:

1075 self.fields['cooling_outflow'].queryset = CoolingOutflow.objects.filter(

1076 device__in=[device, device.get_vc_master()]

1077 )

1078 else:

1079 self.fields['cooling_outflow'].queryset = CoolingOutflow.objects.none()

1080

1081

1082class CoolingOutflowImportForm(OwnerCSVMixin, NetBoxModelImportForm):

1083 device = CSVModelChoiceField(

1084 label=_('Device'),

1085 queryset=Device.objects.all(),

1086 to_field_name='name'

1087 )

1088 type = CSVChoiceField(

1089 label=_('Type'),

1090 choices=CoolingConnectorTypeChoices,

1091 required=False,

1092 help_text=_('Physical connector type')

1093 )

1094 diameter_unit = CSVChoiceField(

1095 label=_('Diameter unit'),

1096 choices=DiameterUnitChoices,

1097 required=False,

1098 help_text=_('Diameter unit')

1099 )

1100 cooling_intake = CSVModelChoiceField(

1101 label=_('Cooling intake'),

1102 queryset=CoolingIntake.objects.all(),

1103 required=False,

1104 to_field_name='name',

1105 help_text=_('Local cooling intake which feeds this outflow')

1106 )

1107

1108 class Meta:

1109 model = CoolingOutflow

1110 fields = (

1111 'device', 'name', 'label', 'type', 'diameter', 'diameter_unit',

1112 'cooling_intake', 'description', 'owner', 'tags',

1113 )

1114

1115 def __init__(self, *args, **kwargs):

1116 super().__init__(*args, **kwargs)

1117

1118 # Limit CoolingIntake choices to those belonging to this device (or VC master)

1119 if self.is_bound and 'device' in self.data:

1120 try:

1121 device = self.fields['device'].to_python(self.data['device'])

1122 except forms.ValidationError:

1123 device = None

1124 else:

1125 try:

1126 device = self.instance.device

1127 except Device.DoesNotExist:

1128 device = None

1129

1130 if device:

1131 self.fields['cooling_intake'].queryset = CoolingIntake.objects.filter(

1132 device__in=[device, device.get_vc_master()]

1133 )

1134 else:

1135 self.fields['cooling_intake'].queryset = CoolingIntake.objects.none()

1136

1137

1138class InterfaceImportForm(OwnerCSVMixin, NetBoxModelImportForm):

1139 device = CSVModelChoiceField(

1140 label=_('Device'),

1141 queryset=Device.objects.all(),

1142 to_field_name='name'

1143 )

1144 parent = CSVModelChoiceField(

1145 label=_('Parent'),

1146 queryset=Interface.objects.all(),

1147 required=False,

1148 to_field_name='name',

1149 help_text=_('Parent interface')

1150 )

1151 bridge = CSVModelChoiceField(

1152 label=_('Bridge'),

1153 queryset=Interface.objects.all(),

1154 required=False,

1155 to_field_name='name',

1156 help_text=_('Bridged interface')

1157 )

1158 lag = CSVModelChoiceField(

1159 label=_('Lag'),

1160 queryset=Interface.objects.all(),

1161 required=False,

1162 to_field_name='name',

1163 help_text=_('Parent LAG interface')

1164 )

1165 vdcs = CSVModelMultipleChoiceField(

1166 label=_('Vdcs'),

1167 queryset=VirtualDeviceContext.objects.all(),

1168 required=False,

1169 to_field_name='name',

1170 help_text=mark_safe(

1171 _('VDC names separated by commas, encased with double quotes. Example:') + ' <code>"vdc1,vdc2,vdc3"</code>'

1172 )

1173 )

1174 type = CSVChoiceField(

1175 label=_('Type'),

1176 choices=InterfaceTypeChoices,

1177 help_text=_('Physical medium')

1178 )

1179 duplex = CSVChoiceField(

1180 label=_('Duplex'),

1181 choices=InterfaceDuplexChoices,

1182 required=False

1183 )

1184 poe_mode = CSVChoiceField(

1185 label=_('Poe mode'),

1186 choices=InterfacePoEModeChoices,

1187 required=False,

1188 help_text=_('PoE mode')

1189 )

1190 poe_type = CSVChoiceField(

1191 label=_('Poe type'),

1192 choices=InterfacePoETypeChoices,

1193 required=False,

1194 help_text=_('PoE type')

1195 )

1196 mode = CSVChoiceField(

1197 label=_('Mode'),

1198 choices=InterfaceModeChoices,

1199 required=False,

1200 help_text=_('IEEE 802.1Q operational mode (for L2 interfaces)'),

1201 )

1202 vlan_group = CSVModelChoiceField(

1203 label=_('VLAN group'),

1204 queryset=VLANGroup.objects.all(),

1205 required=False,

1206 to_field_name='name',

1207 help_text=_('Filter VLANs available for assignment by group'),

1208 )

1209 untagged_vlan = CSVModelChoiceField(

1210 label=_('Untagged VLAN'),

1211 queryset=VLAN.objects.all(),

1212 required=False,

1213 to_field_name='vid',

1214 help_text=_('Assigned untagged VLAN ID (filtered by VLAN group)'),

1215 )

1216 tagged_vlans = CSVModelMultipleChoiceField(

1217 label=_('Tagged VLANs'),

1218 queryset=VLAN.objects.all(),

1219 required=False,

1220 to_field_name='vid',

1221 help_text=mark_safe(

1222 _(

1223 'Assigned tagged VLAN IDs separated by commas, encased with double quotes '

1224 '(filtered by VLAN group). Example:'

1225 )

1226 + ' <code>"100,200,300"</code>'

1227 ),

1228 )

1229 qinq_svlan = CSVModelChoiceField(

1230 label=_('Q-in-Q Service VLAN'),

1231 queryset=VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),

1232 required=False,

1233 to_field_name='vid',

1234 help_text=_('Assigned Q-in-Q Service VLAN ID (filtered by VLAN group)'),

1235 )

1236 vrf = CSVModelChoiceField(

1237 label=_('VRF'),

1238 queryset=VRF.objects.all(),

1239 required=False,

1240 to_field_name='rd',

1241 help_text=_('Assigned VRF')

1242 )

1243 rf_role = CSVChoiceField(

1244 label=_('Rf role'),

1245 choices=WirelessRoleChoices,

1246 required=False,

1247 help_text=_('Wireless role (AP/station)')

1248 )

1249

1250 class Meta:

1251 model = Interface

1252 fields = (

1253 'device', 'name', 'label', 'parent', 'bridge', 'lag', 'type', 'channels', 'channel_id', 'speed', 'duplex',

1254 'enabled', 'mark_connected', 'wwn', 'vdcs', 'mtu', 'mgmt_only', 'description', 'poe_mode', 'poe_type',

1255 'mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vrf', 'rf_role', 'rf_channel',

1256 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'owner', 'tags'

1257 )

1258

1259 def __init__(self, data=None, *args, **kwargs):

1260 super().__init__(data, *args, **kwargs)

1261

1262 if data:

1263 # Limit choices for parent, bridge, and LAG interfaces to the assigned device

1264 if device := data.get('device'):

1265 params = {

1266 f"device__{self.fields['device'].to_field_name}": device

1267 }

1268 self.fields['parent'].queryset = self.fields['parent'].queryset.filter(**params)

1269 self.fields['bridge'].queryset = self.fields['bridge'].queryset.filter(**params)

1270 self.fields['lag'].queryset = self.fields['lag'].queryset.filter(**params)

1271 self.fields['vdcs'].queryset = self.fields['vdcs'].queryset.filter(**params)

1272

1273 # Limit choices for VLANs to the assigned VLAN group

1274 if vlan_group := data.get('vlan_group'):

1275 params = {f"group__{self.fields['vlan_group'].to_field_name}": vlan_group}

1276 self.fields['untagged_vlan'].queryset = self.fields['untagged_vlan'].queryset.filter(**params)

1277 self.fields['tagged_vlans'].queryset = self.fields['tagged_vlans'].queryset.filter(**params)

1278 self.fields['qinq_svlan'].queryset = self.fields['qinq_svlan'].queryset.filter(**params)

1279

1280 def clean_enabled(self):

1281 # Make sure enabled is True when it's not included in the uploaded data

1282 if 'enabled' not in self.data:

1283 return True

1284 return self.cleaned_data['enabled']

1285

1286 def clean_vdcs(self):

1287 for vdc in self.cleaned_data['vdcs']:

1288 if vdc.device != self.cleaned_data['device']:

1289 raise forms.ValidationError(

1290 _("VDC {vdc} is not assigned to device {device}").format(

1291 vdc=vdc, device=self.cleaned_data['device']

1292 )

1293 )

1294 return self.cleaned_data['vdcs']

1295

1296

1297class FrontPortImportForm(OwnerCSVMixin, NetBoxModelImportForm):

1298 device = CSVModelChoiceField(

1299 label=_('Device'),

1300 queryset=Device.objects.all(),

1301 to_field_name='name'

1302 )

1303 type = CSVChoiceField(

1304 label=_('Type'),

1305 choices=PortTypeChoices,

1306 help_text=_('Physical medium classification')

1307 )

1308 rear_port = CSVModelChoiceField(

1309 label=_('Rear port'),

1310 queryset=RearPort.objects.all(),

1311 to_field_name='name',

1312 help_text=_('Corresponding rear port (mapped to the front port\'s first position)')

1313 )

1314 rear_port_position = forms.IntegerField(

1315 label=_('Rear port position'),

1316 required=False,

1317 help_text=_('Mapped position on the corresponding rear port (defaults to 1)')

1318 )

1319

1320 class Meta:

1321 model = FrontPort

1322 fields = (

1323 'device', 'name', 'label', 'type', 'color', 'mark_connected', 'positions', 'rear_port',

1324 'rear_port_position', 'description', 'owner', 'tags'

1325 )

1326

1327 def __init__(self, *args, **kwargs):

1328 super().__init__(*args, **kwargs)

1329

1330 # Limit RearPort choices to those belonging to this device (or VC master)

1331 if self.is_bound and 'device' in self.data:

1332 try:

1333 device = self.fields['device'].to_python(self.data['device'])

1334 except forms.ValidationError:

1335 device = None

1336 else:

1337 try:

1338 device = self.instance.device

1339 except Device.DoesNotExist:

1340 device = None

1341

1342 if device:

1343 self.fields['rear_port'].queryset = RearPort.objects.filter(

1344 device__in=[device, device.get_vc_master()]

1345 )

1346 else:

1347 self.fields['rear_port'].queryset = RearPort.objects.none()

1348

1349 def clean(self):

1350 super().clean()

1351

1352 rear_port = self.cleaned_data.get('rear_port')

1353 rear_port_position = self.cleaned_data.get('rear_port_position') or 1

1354 if not rear_port:

1355 return

1356

1357 # Validate the rear port position against the selected rear port

1358 if rear_port_position > rear_port.positions:

1359 raise forms.ValidationError({

1360 'rear_port_position': _(

1361 "Invalid rear port position ({rear_port_position}): Rear port {name} has only {positions} "

1362 "positions."

1363 ).format(

1364 rear_port_position=rear_port_position,

1365 name=rear_port.name,

1366 positions=rear_port.positions

1367 )

1368 })

1369

1370 # Ensure the target rear port position isn't already occupied. reconcile_port_mappings() creates the

1371 # mapping via create() (bypassing validate_unique()), so without this check a collision would surface

1372 # as an uncaught IntegrityError (HTTP 500) rather than a row-level validation error.

1373 occupied = PortMapping.objects.filter(

1374 rear_port=rear_port, rear_port_position=rear_port_position

1375 ).exclude(front_port=self.instance.pk)

1376 if occupied.exists():

1377 raise forms.ValidationError({

1378 'rear_port_position': _(

1379 "Rear port {name} position {rear_port_position} is already occupied."

1380 ).format(

1381 name=rear_port.name,

1382 rear_port_position=rear_port_position

1383 )

1384 })

1385

1386 def _save_m2m(self):

1387 super()._save_m2m()

1388

1389 # Map the front port's first position to the specified rear port & position

1390 if rear_port := self.cleaned_data.get('rear_port'):

1391 reconcile_port_mappings(

1392 PortMapping,

1393 parent_field='front_port',

1394 parent=self.instance,

1395 desired=[{

1396 'front_port_position': 1,

1397 'rear_port_id': rear_port.pk,

1398 'rear_port_position': self.cleaned_data.get('rear_port_position') or 1,

1399 }],

1400 )

1401

1402

1403class RearPortImportForm(OwnerCSVMixin, NetBoxModelImportForm):

1404 device = CSVModelChoiceField(

1405 label=_('Device'),

1406 queryset=Device.objects.all(),

1407 to_field_name='name'

1408 )

1409 type = CSVChoiceField(

1410 label=_('Type'),

1411 help_text=_('Physical medium classification'),

1412 choices=PortTypeChoices,

1413 )

1414

1415 class Meta:

1416 model = RearPort

1417 fields = (

1418 'device', 'name', 'label', 'type', 'color', 'mark_connected', 'positions', 'description', 'owner', 'tags',

1419 )

1420

1421

1422class ModuleBayImportForm(OwnerCSVMixin, NetBoxModelImportForm):

1423 device = CSVModelChoiceField(

1424 label=_('Device'),

1425 queryset=Device.objects.all(),

1426 to_field_name='name'

1427 )

1428

1429 class Meta:

1430 model = ModuleBay

1431 fields = ('device', 'name', 'label', 'position', 'enabled', 'description', 'owner', 'tags')

1432

1433 def clean_enabled(self):

1434 # Make sure enabled is True when it's not included in the uploaded data

1435 if 'enabled' not in self.data:

1436 return True

1437 return self.cleaned_data['enabled']

1438

1439

1440class DeviceBayImportForm(OwnerCSVMixin, NetBoxModelImportForm):

1441 device = CSVModelChoiceField(

1442 label=_('Device'),

1443 queryset=Device.objects.all(),

1444 to_field_name='name'

1445 )

1446 installed_device = CSVModelChoiceField(

1447 label=_('Installed device'),

1448 queryset=Device.objects.all(),

1449 required=False,

1450 to_field_name='name',

1451 help_text=_('Child device installed within this bay'),

1452 error_messages={

1453 'invalid_choice': _('Child device not found.'),

1454 }

1455 )

1456

1457 class Meta:

1458 model = DeviceBay

1459 fields = ('device', 'name', 'label', 'enabled', 'installed_device', 'description', 'owner', 'tags')

1460

1461 def __init__(self, *args, **kwargs):

1462 super().__init__(*args, **kwargs)

1463

1464 # Limit installed device choices to devices of the correct type and location

1465 if self.is_bound and 'device' in self.data:

1466 try:

1467 device = self.fields['device'].to_python(self.data['device'])

1468 except forms.ValidationError:

1469 device = None

1470 else:

1471 try:

1472 device = self.instance.device

1473 except Device.DoesNotExist:

1474 device = None

1475

1476 if device:

1477 self.fields['installed_device'].queryset = Device.objects.filter(

1478 site=device.site,

1479 rack=device.rack,

1480 parent_bay__isnull=True,

1481 device_type__u_height=0,

1482 device_type__subdevice_role=SubdeviceRoleChoices.ROLE_CHILD

1483 ).exclude(pk=device.pk)

1484 else:

1485 self.fields['installed_device'].queryset = Device.objects.none()

1486

1487 def clean_enabled(self):

1488 # Make sure enabled is True when it's not included in the uploaded data

1489 if 'enabled' not in self.data:

1490 return True

1491 return self.cleaned_data['enabled']

1492

1493

1494class InventoryItemImportForm(OwnerCSVMixin, NetBoxModelImportForm):

1495 device = CSVModelChoiceField(

1496 label=_('Device'),

1497 queryset=Device.objects.all(),

1498 to_field_name='name'

1499 )

1500 role = CSVModelChoiceField(

1501 label=_('Role'),

1502 queryset=InventoryItemRole.objects.all(),

1503 to_field_name='name',

1504 required=False

1505 )

1506 manufacturer = CSVModelChoiceField(

1507 label=_('Manufacturer'),

1508 queryset=Manufacturer.objects.all(),

1509 to_field_name='name',

1510 required=False

1511 )

1512 parent = CSVModelChoiceField(

1513 label=_('Parent'),

1514 queryset=Device.objects.all(),

1515 to_field_name='name',

1516 required=False,

1517 help_text=_('Parent inventory item')

1518 )

1519 component_type = CSVContentTypeField(

1520 label=_('Component type'),

1521 queryset=ContentType.objects.all(),

1522 limit_choices_to=MODULAR_COMPONENT_MODELS,

1523 required=False,

1524 help_text=_('Component Type')

1525 )

1526 component_name = forms.CharField(

1527 label=_('Component name'),

1528 required=False,

1529 help_text=_('Component Name')

1530 )

1531 status = CSVChoiceField(

1532 label=_('Status'),

1533 choices=InventoryItemStatusChoices,

1534 help_text=_('Operational status')

1535 )

1536

1537 class Meta:

1538 model = InventoryItem

1539 fields = (

1540 'device', 'name', 'label', 'status', 'role', 'manufacturer', 'parent', 'part_id', 'serial', 'asset_tag',

1541 'discovered', 'description', 'owner', 'tags', 'component_type', 'component_name',

1542 )

1543

1544 def __init__(self, *args, **kwargs):

1545 super().__init__(*args, **kwargs)

1546

1547 # Limit parent choices to inventory items belonging to this device

1548 device = None

1549 if self.is_bound and 'device' in self.data:

1550 try:

1551 device = self.fields['device'].to_python(self.data['device'])

1552 except forms.ValidationError:

1553 pass

1554 if device:

1555 self.fields['parent'].queryset = InventoryItem.objects.filter(device=device)

1556 else:

1557 self.fields['parent'].queryset = InventoryItem.objects.none()

1558

1559 def clean(self):

1560 super().clean()

1561 cleaned_data = self.cleaned_data

1562 component_type = cleaned_data.get('component_type')

1563 component_name = cleaned_data.get('component_name')

1564 device = self.cleaned_data.get("device")

1565

1566 if component_type:

1567 if device is None:

1568 cleaned_data.pop('component_type', None)

1569 if component_name is None:

1570 cleaned_data.pop('component_type', None)

1571 raise forms.ValidationError(

1572 _("Component name must be specified when component type is specified")

1573 )

1574 if all([device, component_name]):

1575 try:

1576 model = component_type.model_class()

1577 self.instance.component = model.objects.get(device=device, name=component_name)

1578 except ObjectDoesNotExist:

1579 cleaned_data.pop('component_type', None)

1580 cleaned_data.pop('component_name', None)

1581 raise forms.ValidationError(

1582 _("Component not found: {device} - {component_name}").format(

1583 device=device, component_name=component_name

1584 )

1585 )

1586 else:

1587 cleaned_data.pop('component_type', None)

1588 if not component_name:

1589 raise forms.ValidationError(

1590 _("Component name must be specified when component type is specified")

1591 )

1592 else:

1593 if component_name:

1594 raise forms.ValidationError(

1595 _("Component type must be specified when component name is specified")

1596 )

1597 return cleaned_data

1598

1599

1600#

1601# Device component roles

1602#

1603

1604class InventoryItemRoleImportForm(OrganizationalModelImportForm):

1605 slug = SlugField()

1606

1607 class Meta:

1608 model = InventoryItemRole

1609 fields = ('name', 'slug', 'color', 'description', 'owner', 'comments')

1613# Addressing

1616class MACAddressImportForm(PrimaryModelImportForm):

1617 device = CSVModelChoiceField(

1618 label=_('Device'),

1619 queryset=Device.objects.all(),

1620 required=False,

1621 to_field_name='name',

1622 help_text=_('Parent device of assigned interface (if any)')

1623 )

1624 virtual_machine = CSVModelChoiceField(

1625 label=_('Virtual machine'),

1626 queryset=VirtualMachine.objects.all(),

1627 required=False,

1628 to_field_name='name',

1629 help_text=_('Parent VM of assigned interface (if any)')

1630 )

1631 interface = CSVModelChoiceField(

1632 label=_('Interface'),

1633 queryset=Interface.objects.none(), # Can also refer to VMInterface

1634 required=False,

1635 to_field_name='name',

1636 help_text=_('Assigned interface')

1637 )

1638 is_primary = forms.BooleanField(

1639 label=_('Is primary'),

1640 help_text=_('Make this the primary MAC address for the assigned interface'),

1641 required=False

1642 )

1643

1644 class Meta:

1645 model = MACAddress

1646 fields = [

1647 'mac_address', 'device', 'virtual_machine', 'interface', 'is_primary', 'description', 'owner', 'comments',

1648 'tags',

1649 ]

1650

1651 def __init__(self, data=None, *args, **kwargs):

1652 super().__init__(data, *args, **kwargs)

1653

1654 if data:

1655

1656 # Limit interface queryset by assigned device

1657 if data.get('device'):

1658 self.fields['interface'].queryset = Interface.objects.filter(

1659 **{f"device__{self.fields['device'].to_field_name}": data['device']}

1660 )

1661

1662 # Limit interface queryset by assigned device

1663 elif data.get('virtual_machine'):

1664 self.fields['interface'].queryset = VMInterface.objects.filter(

1665 **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']}

1666 )

1667

1668 def clean(self):

1669 super().clean()

1670

1671 device = self.cleaned_data.get('device')

1672 virtual_machine = self.cleaned_data.get('virtual_machine')

1673 interface = self.cleaned_data.get('interface')

1674

1675 # Validate interface assignment

1676 if interface and not device and not virtual_machine:

1677 raise forms.ValidationError({

1678 "interface": _("Must specify the parent device or VM when assigning an interface")

1679 })

1680

1681 def save(self, *args, **kwargs):

1682

1683 # Set interface assignment

1684 if interface := self.cleaned_data.get('interface'):

1685 self.instance.assigned_object = interface

1686

1687 instance = super().save(*args, **kwargs)

1688

1689 # Assign the MAC address as primary for its interface, if designated as such

1690 if interface and self.cleaned_data['is_primary'] and self.instance.pk:

1691 interface.snapshot()

1692 interface.primary_mac_address = self.instance

1693 interface.save()

1695 return instance

1699# Cables

1702class CableBundleImportForm(PrimaryModelImportForm):

1703 class Meta:

1704 model = CableBundle

1705 fields = ('name', 'description', 'owner', 'comments', 'tags')

1706

1707

1708class CableImportForm(PrimaryModelImportForm):

1709 # Cable.clean() reports termination errors (e.g. cable profile violations) against the model's

1710 # a_terminations/b_terminations attributes, which have no corresponding fields on this form.

1711 # Map them onto the columns which define each side's terminations.

1712 TERMINATION_ERROR_FIELDS = {

1713 'a_terminations': 'side_a_name',

1714 'b_terminations': 'side_b_name',

1715 }

1716

1717 # Columns which take effect only by resolving a side's terminations, and are therefore

1718 # meaningless without that side's name column.

1719 TERMINATION_DEPENDENT_COLUMNS = ('site', 'device', 'power_panel', 'type')

1720

1721 # Termination A

1722 side_a_site = CSVModelChoiceField(

1723 label=_('Side A site'),

1724 queryset=Site.objects.all(),

1725 required=False,

1726 to_field_name='name',

1727 help_text=_('Site of parent device A (if any). Restricts the devices & power panels which may be matched.'),

1728 )

1729 side_a_device = CSVModelMultipleChoiceField(

1730 label=_('Side A device'),

1731 queryset=Device.objects.all(),

1732 required=False,

1733 to_field_name='name',

1734 help_text=format_html_lazy(

1735 '{} <code>{}</code>',

1736 _('Device name(s) for device component terminations. Separate multiple values with commas, '

1737 'encased with double quotes. Example:'),

1738 '"device1,device2"'

1739 )

1740 )

1741 side_a_power_panel = CSVModelMultipleChoiceField(

1742 label=_('Side A power panel'),

1743 queryset=PowerPanel.objects.all(),

1744 required=False,

1745 to_field_name='name',

1746 help_text=format_html_lazy(

1747 '{} <code>{}</code>',

1748 _('Power panel name(s) for power feed terminations. Separate multiple values with commas, '

1749 'encased with double quotes. Example:'),

1750 '"panel1,panel2"'

1751 )

1752 )

1753 side_a_type = CSVContentTypeField(

1754 label=_('Side A type'),

1755 queryset=ContentType.objects.all(),

1756 limit_choices_to=CABLE_TERMINATION_MODELS,

1757 help_text=_('Termination type')

1758 )

1759 side_a_name = forms.CharField(

1760 label=_('Side A name'),

1761 help_text=format_html_lazy(

1762 '{} <code>{}</code>',

1763 _('Termination name(s). Separate multiple values with commas, encased with double quotes. '

1764 'Example:'),

1765 '"eth0,eth1"'

1766 )

1767 )

1768

1769 # Termination B

1770 side_b_site = CSVModelChoiceField(

1771 label=_('Side B site'),

1772 queryset=Site.objects.all(),

1773 required=False,

1774 to_field_name='name',

1775 help_text=_('Site of parent device B (if any). Restricts the devices & power panels which may be matched.'),

1776 )

1777 side_b_device = CSVModelMultipleChoiceField(

1778 label=_('Side B device'),

1779 queryset=Device.objects.all(),

1780 required=False,

1781 to_field_name='name',

1782 help_text=format_html_lazy(

1783 '{} <code>{}</code>',

1784 _('Device name(s) for device component terminations. Separate multiple values with commas, '

1785 'encased with double quotes. Example:'),

1786 '"device1,device2"'

1787 )

1788 )

1789 side_b_power_panel = CSVModelMultipleChoiceField(

1790 label=_('Side B power panel'),

1791 queryset=PowerPanel.objects.all(),

1792 required=False,

1793 to_field_name='name',

1794 help_text=format_html_lazy(

1795 '{} <code>{}</code>',

1796 _('Power panel name(s) for power feed terminations. Separate multiple values with commas, '

1797 'encased with double quotes. Example:'),

1798 '"panel1,panel2"'

1799 )

1800 )

1801 side_b_type = CSVContentTypeField(

1802 label=_('Side B type'),

1803 queryset=ContentType.objects.all(),

1804 limit_choices_to=CABLE_TERMINATION_MODELS,

1805 help_text=_('Termination type')

1806 )

1807 side_b_name = forms.CharField(

1808 label=_('Side B name'),

1809 help_text=format_html_lazy(

1810 '{} <code>{}</code>',

1811 _('Termination name(s). Separate multiple values with commas, encased with double quotes. '

1812 'Example:'),

1813 '"eth0,eth1"'

1814 )

1815 )

1816

1817 # Cable attributes

1818 status = CSVChoiceField(

1819 label=_('Status'),

1820 choices=LinkStatusChoices,

1821 required=False,

1822 help_text=_('Connection status')

1823 )

1824 profile = CSVChoiceField(

1825 label=_('Profile'),

1826 choices=CableProfileChoices,

1827 required=False,

1828 help_text=_('Cable connection profile')

1829 )

1830 type = CSVChoiceField(

1831 label=_('Type'),

1832 choices=CableTypeChoices,

1833 required=False,

1834 help_text=_('Physical medium classification')

1835 )

1836 tenant = CSVModelChoiceField(

1837 label=_('Tenant'),

1838 queryset=Tenant.objects.all(),

1839 required=False,

1840 to_field_name='name',

1841 help_text=_('Assigned tenant')

1842 )

1843 bundle = CSVModelChoiceField(

1844 label=_('Bundle'),

1845 queryset=CableBundle.objects.all(),

1846 required=False,

1847 to_field_name='name',

1848 help_text=_('Cable bundle name'),

1849 )

1850 length_unit = CSVChoiceField(

1851 label=_('Length unit'),

1852 choices=CableLengthUnitChoices,

1853 required=False,

1854 help_text=_('Length unit')

1855 )

1856 color = forms.CharField(

1857 label=_('Color'),

1858 required=False,

1859 max_length=16,

1860 help_text=_('Color name (e.g. "Red") or hex code (e.g. "f44336")')

1861 )

1862

1863 class Meta:

1864 model = Cable

1865 fields = [

1866 'side_a_site', 'side_a_device', 'side_a_power_panel', 'side_a_type', 'side_a_name',

1867 'side_b_site', 'side_b_device', 'side_b_power_panel', 'side_b_type', 'side_b_name',

1868 'type', 'status', 'profile', 'tenant', 'bundle', 'label', 'color', 'length', 'length_unit',

1869 'description', 'owner', 'comments', 'tags',

1870 ]

1871

1872 def __init__(self, data=None, *args, **kwargs):

1873 super().__init__(data, *args, **kwargs)

1874

1875 if data:

1876 # Limit choices for side_a_device to the assigned side_a_site

1877 if side_a_site := data.get('side_a_site'):

1878 side_a_parent_params = {f'site__{self.fields['side_a_site'].to_field_name}': side_a_site}

1879 self.fields['side_a_device'].queryset = self.fields['side_a_device'].queryset.filter(

1880 **side_a_parent_params

1881 )

1882 self.fields['side_a_power_panel'].queryset = self.fields['side_a_power_panel'].queryset.filter(

1883 **side_a_parent_params

1884 )

1885

1886 # Limit choices for side_b_device to the assigned side_b_site

1887 if side_b_site := data.get('side_b_site'):

1888 side_b_parent_params = {f'site__{self.fields['side_b_site'].to_field_name}': side_b_site}

1889 self.fields['side_b_device'].queryset = self.fields['side_b_device'].queryset.filter(

1890 **side_b_parent_params

1891 )

1892 self.fields['side_b_power_panel'].queryset = self.fields['side_b_power_panel'].queryset.filter(

1893 **side_b_parent_params

1894 )

1895

1896 def _update_errors(self, errors):

1897 # Remap any termination errors raised by Cable.clean() onto the relevant import column.

1898 # Without this, the base form reports them as prefixed non-field errors, as neither

1899 # a_terminations nor b_terminations exists as a field on this form. Errors are dispatched

1900 # per key so that both sides can fall back to a non-field error without colliding.

1901 if hasattr(errors, 'error_dict'):

1902 remapped = {}

1903 for name, error_list in errors.error_dict.items():

1904 mapped_name = self._map_termination_field(name) or NON_FIELD_ERRORS

1905 remapped.setdefault(mapped_name, []).extend(error_list)

1906 errors = forms.ValidationError(remapped)

1907

1908 super()._update_errors(errors)

1909

1910 def _map_termination_field(self, field):

1911 """

1912 Return the import column against which a model-level termination error should be reported,

1913 or None (i.e. a non-field error) if that column is not present on the form. Columns absent

1914 from an update record are removed from the form by BulkImportView, so a cable profile

1915 violation can be reported against a side whose terminations were not being modified.

1916 """

1917 if field not in self.TERMINATION_ERROR_FIELDS:

1918 return field

1919 mapped_field = self.TERMINATION_ERROR_FIELDS[field]

1920 return mapped_field if mapped_field in self.fields else None

1921

1922 @staticmethod

1923 def _split_side_values(value):

1924 """

1925 Split a side_* cell into an ordered list of values, preserving duplicates and empty

1926 entries. Accepts a comma-separated string (CSV) or a native list (JSON/YAML).

1927 """

1928 if value in (None, ''):

1929 return []

1930 if not isinstance(value, (list, tuple)):

1931 value = str(value).split(',')

1932 return ['' if item is None else str(item).strip() for item in value]

1933

1934 def _check_companion_column(self, side, field_name):

1935 """

1936 Verify that a column needed to resolve a side's terminations is present on the form.

1937

1938 When updating an existing object, BulkImportView removes every field which does not appear

1939 in the record. A record which redefines a side's termination names must therefore also

1940 include the columns identifying their type and parent, otherwise the names cannot be

1941 resolved and the update would appear to succeed while changing nothing.

1942 """

1943 if field_name not in self.fields:

1944 raise forms.ValidationError(

1945 _(

1946 "Side {side_upper}: The {column} column must be included when modifying terminations"

1947 ).format(side_upper=side.upper(), column=field_name)

1948 )

1949

1950 def _resolve_side_parent_objects(self, field_name):

1951 """

1952 Resolve a side's parent objects from the raw submitted values, preserving their order.

1953 CSVModelMultipleChoiceField cleans to an unordered queryset, which cannot be used to pair

1954 each parent with its corresponding termination name by position. Resolution errors are

1955 reported on the parent field itself.

1956 """

1957 if field_name not in self.cleaned_data:

1958 # The parent field has already raised its own validation error

1959 return None

1960 field = self.fields[field_name]

1961 to_field_name = field.to_field_name or 'pk'

1962

1963 parents = []

1964 for value in self._split_side_values(self.data.get(field_name)):

1965 try:

1966 parents.append(field.queryset.get(**{to_field_name: value}))

1967 except ObjectDoesNotExist:

1968 self.add_error(field_name, _("Object not found: {value}").format(value=value))

1969 return None

1970 except MultipleObjectsReturned:

1971 self.add_error(

1972 field_name,

1973 _('"{value}" is not a unique value for this field; multiple objects were found').format(

1974 value=value

1975 )

1976 )

1977 return None

1978 return parents

1979

1980 @staticmethod

1981 def _get_device_component_termination(model, device, name):

1982 """

1983 Resolve a device component by its device and name. If the device is a virtual chassis

1984 master and the component is not found on it, search all virtual chassis members.

1985 """

1986 queryset = model.objects.filter(device=device, name=name)

1987 if (

1988 device.virtual_chassis and

1989 device.virtual_chassis.master == device and

1990 not queryset.exists()

1991 ):

1992 queryset = model.objects.filter(device__in=device.virtual_chassis.members.all(), name=name)

1993 return queryset.get()

1994

1995 def _clean_side(self, side):

1996 """

1997 Derive a Cable's A/B termination objects.

1998

1999 :param side: 'a' or 'b'

2000 """

2001 if side not in ('a', 'b'):

2002 raise ValueError(_("Invalid side designation: {side}").format(side=side))

2003

2004 content_type = self.cleaned_data.get(f'side_{side}_type')

2005 # Native list values (JSON/YAML) bypass the CharField; strings use its cleaned value

2006 names = self.data.get(f'side_{side}_name')

2007 if not isinstance(names, (list, tuple)):

2008 names = self.cleaned_data.get(f'side_{side}_name')

2009 names = self._split_side_values(names)

2010 if not names:

2011 return None

2012

2013 if not content_type:

2014 # BulkImportView removes any field absent from an update record, so a missing termination

2015 # type here means the column was omitted rather than left blank. Reject it: silently

2016 # ignoring the submitted names would report a successful update which changed nothing.

2017 self._check_companion_column(side, f'side_{side}_type')

2018 return None

2019

2020 if '' in names:

2021 raise forms.ValidationError(

2022 _("Side {side_upper}: Empty termination names are not permitted").format(side_upper=side.upper())

2023 )

2024

2025 model = content_type.model_class()

2026

2027 # Identify the parent field for the termination type. PowerFeed terminations reference a

2028 # PowerPanel; all other supported types reference a Device.

2029 if content_type.model == 'powerfeed':

2030 parent_field_name = f'side_{side}_power_panel'

2031 parent_label = _('power panel')

2032 elif any(field.name == 'device' for field in model._meta.fields):

2033 parent_field_name = f'side_{side}_device'

2034 parent_label = _('device')

2035 else:

2036 raise forms.ValidationError(

2037 _("Bulk import does not support {type} terminations").format(type=content_type)

2038 )

2039

2040 self._check_companion_column(side, parent_field_name)

2041

2042 parents = self._resolve_side_parent_objects(parent_field_name)

2043 if parents is None:

2044 # The parent field has already raised its own validation error

2045 return None

2046 if not parents:

2047 raise forms.ValidationError(

2048 _("Side {side_upper}: Must specify a {parent} for the selected termination type").format(

2049 side_upper=side.upper(), parent=parent_label

2050 )

2051 )

2052 if len(parents) == 1:

2053 parents = parents * len(names)

2054 elif len(parents) != len(names):

2055 raise forms.ValidationError(

2056 _(

2057 "Side {side_upper}: Must specify either one {parent} for all terminations or one {parent} "

2058 "per termination name"

2059 ).format(side_upper=side.upper(), parent=parent_label)

2060 )

2061

2062 terminations = []

2063 for parent, name in zip(parents, names):

2064 try:

2065 if content_type.model == 'powerfeed':

2066 termination_object = model.objects.get(power_panel=parent, name=name)

2067 else:

2068 termination_object = self._get_device_component_termination(model, parent, name)

2069 except ObjectDoesNotExist:

2070 raise forms.ValidationError(

2071 _("{side_upper} side termination not found: {parent} {name}").format(

2072 side_upper=side.upper(), parent=parent, name=name

2073 )

2074 )

2075 except MultipleObjectsReturned:

2076 raise forms.ValidationError(

2077 _("{side_upper} side termination not unique: {parent} {name}").format(

2078 side_upper=side.upper(), parent=parent, name=name

2079 )

2080 )

2081 if termination_object.cable is not None and termination_object.cable != self.instance:

2082 raise forms.ValidationError(

2083 _("Side {side_upper}: {parent} {termination_object} is already connected").format(

2084 side_upper=side.upper(), parent=parent, termination_object=termination_object

2085 )

2086 )

2087 terminations.append(termination_object)

2088

2089 if len({termination.pk for termination in terminations}) != len(terminations):

2090 raise forms.ValidationError(

2091 _("Side {side_upper}: Duplicate termination specified").format(side_upper=side.upper())

2092 )

2093

2094 setattr(self.instance, f'{side}_terminations', terminations)

2095 return terminations

2096

2097 def _clean_color(self, color):

2098 """

2099 Derive a colors hex code

2100

2101 :param color: color as hex or color name

2102 """

2103 color_parsed = color.strip().lower()

2104

2105 for hex_code, label in ColorChoices.CHOICES:

2106 if color.lower() == label.lower():

2107 color_parsed = hex_code

2108

2109 if len(color_parsed) > 6:

2110 raise forms.ValidationError(

2111 _(f"{color} did not match any used color name and was longer than six characters: invalid hex.")

2112 )

2113 return color_parsed

2114

2115 def clean(self):

2116 cleaned_data = super().clean()

2117

2118 # Termination resolution is driven by clean_side_<x>_name(), which Django never calls for a

2119 # field BulkImportView has removed. An update record which supplies a side's supporting

2120 # columns but omits its name column would therefore have those columns silently discarded

2121 # and report an update which changed nothing; reject it instead. This cannot trigger on

2122 # creation, where the name columns are always present (and required).

2123 for side in ('a', 'b'):

2124 if f'side_{side}_name' in self.fields:

2125 continue

2126 if supplied := [

2127 column for column in self.TERMINATION_DEPENDENT_COLUMNS

2128 if f'side_{side}_{column}' in self.fields

2129 ]:

2130 self.add_error(None, _(

2131 "Side {side_upper}: The side_{side}_name column must be included when modifying "

2132 "terminations (found {columns})"

2133 ).format(

2134 side_upper=side.upper(),

2135 side=side,

2136 columns=', '.join(f'side_{side}_{column}' for column in supplied),

2137 ))

2138

2139 return cleaned_data

2140

2141 def clean_side_a_name(self):

2142 return self._clean_side('a')

2143

2144 def clean_side_b_name(self):

2145 return self._clean_side('b')

2146

2147 def clean_length_unit(self):

2148 # Avoid trying to save as NULL

2149 length_unit = self.cleaned_data.get('length_unit', None)

2150 return length_unit if length_unit is not None else ''

2151

2152 def clean_color(self):

2153 color = self.cleaned_data.get('color', None)

2154 return self._clean_color(color) if color is not None else ''

2155#

2156# Virtual chassis

2157#

2158

2159

2160class VirtualChassisImportForm(PrimaryModelImportForm):

2161 master = CSVModelChoiceField(

2162 label=_('Master'),

2163 queryset=Device.objects.all(),

2164 to_field_name='name',

2165 required=False,

2166 help_text=_('Master device')

2167 )

2168

2169 class Meta:

2170 model = VirtualChassis

2171 fields = ('name', 'domain', 'master', 'description', 'owner', 'comments', 'tags')

2175# Power

2178class PowerPanelImportForm(PrimaryModelImportForm):

2179 site = CSVModelChoiceField(

2180 label=_('Site'),

2181 queryset=Site.objects.all(),

2182 to_field_name='name',

2183 help_text=_('Name of parent site')

2184 )

2185 location = CSVModelChoiceField(

2186 label=_('Location'),

2187 queryset=Location.objects.all(),

2188 required=False,

2189 to_field_name='name'

2190 )

2191

2192 class Meta:

2193 model = PowerPanel

2194 fields = ('site', 'location', 'name', 'description', 'owner', 'comments', 'tags')

2195

2196 def __init__(self, data=None, *args, **kwargs):

2197 super().__init__(data, *args, **kwargs)

2198

2199 if data:

2200

2201 # Limit group queryset by assigned site

2202 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

2203 self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)

2204

2205

2206class PowerFeedImportForm(PrimaryModelImportForm):

2207 site = CSVModelChoiceField(

2208 label=_('Site'),

2209 queryset=Site.objects.all(),

2210 to_field_name='name',

2211 help_text=_('Assigned site')

2212 )

2213 power_panel = CSVModelChoiceField(

2214 label=_('Power panel'),

2215 queryset=PowerPanel.objects.all(),

2216 to_field_name='name',

2217 help_text=_('Upstream power panel')

2218 )

2219 location = CSVModelChoiceField(

2220 label=_('Location'),

2221 queryset=Location.objects.all(),

2222 to_field_name='name',

2223 required=False,

2224 help_text=_("Rack's location (if any)")

2225 )

2226 rack = CSVModelChoiceField(

2227 label=_('Rack'),

2228 queryset=Rack.objects.all(),

2229 to_field_name='name',

2230 required=False,

2231 help_text=_('Rack')

2232 )

2233 tenant = CSVModelChoiceField(

2234 queryset=Tenant.objects.all(),

2235 to_field_name='name',

2236 required=False,

2237 help_text=_('Assigned tenant')

2238 )

2239 status = CSVChoiceField(

2240 label=_('Status'),

2241 choices=PowerFeedStatusChoices,

2242 help_text=_('Operational status')

2243 )

2244 type = CSVChoiceField(

2245 label=_('Type'),

2246 choices=PowerFeedTypeChoices,

2247 help_text=_('Primary or redundant')

2248 )

2249 supply = CSVChoiceField(

2250 label=_('Supply'),

2251 choices=PowerFeedSupplyChoices,

2252 help_text=_('Supply type (AC/DC)')

2253 )

2254 phase = CSVChoiceField(

2255 label=_('Phase'),

2256 choices=PowerFeedPhaseChoices,

2257 help_text=_('Single or three-phase')

2258 )

2259

2260 class Meta:

2261 model = PowerFeed

2262 fields = (

2263 'site', 'power_panel', 'location', 'rack', 'name', 'status', 'type', 'mark_connected', 'supply', 'phase',

2264 'voltage', 'amperage', 'max_utilization', 'tenant', 'description', 'owner', 'comments', 'tags',

2265 )

2266

2267 def __init__(self, data=None, *args, **kwargs):

2268 super().__init__(data, *args, **kwargs)

2269

2270 if data:

2271

2272 # Limit power_panel queryset by site

2273 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

2274 self.fields['power_panel'].queryset = self.fields['power_panel'].queryset.filter(**params)

2275

2276 # Limit location queryset by site

2277 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

2278 self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)

2279

2280 # Limit rack queryset by site and group

2281 params = {

2282 f"site__{self.fields['site'].to_field_name}": data.get('site'),

2283 f"location__{self.fields['location'].to_field_name}": data.get('location'),

2284 }

2285 self.fields['rack'].queryset = self.fields['rack'].queryset.filter(**params)

2286

2287

2288class CoolingSourceImportForm(PrimaryModelImportForm):

2289 site = CSVModelChoiceField(

2290 label=_('Site'),

2291 queryset=Site.objects.all(),

2292 to_field_name='name',

2293 help_text=_('Name of parent site')

2294 )

2295 location = CSVModelChoiceField(

2296 label=_('Location'),

2297 queryset=Location.objects.all(),

2298 required=False,

2299 to_field_name='name'

2300 )

2301 type = CSVChoiceField(

2302 label=_('Type'),

2303 choices=CoolingSourceTypeChoices,

2304 help_text=_('Cooling source type')

2305 )

2306 status = CSVChoiceField(

2307 label=_('Status'),

2308 choices=CoolingSourceStatusChoices,

2309 help_text=_('Operational status')

2310 )

2311 fluid_type = CSVChoiceField(

2312 label=_('Fluid type'),

2313 choices=FluidTypeChoices,

2314 required=False,

2315 help_text=_('Coolant fluid type')

2316 )

2317

2318 class Meta:

2319 model = CoolingSource

2320 fields = (

2321 'site', 'location', 'name', 'type', 'status', 'fluid_type', 'cooling_capacity', 'description', 'owner',

2322 'comments', 'tags',

2323 )

2324

2325 def __init__(self, data=None, *args, **kwargs):

2326 super().__init__(data, *args, **kwargs)

2327

2328 if data:

2329

2330 # Limit location queryset by assigned site

2331 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

2332 self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)

2333

2334

2335class CoolingFeedImportForm(PrimaryModelImportForm):

2336 site = CSVModelChoiceField(

2337 label=_('Site'),

2338 queryset=Site.objects.all(),

2339 to_field_name='name',

2340 help_text=_('Assigned site')

2341 )

2342 cooling_source = CSVModelChoiceField(

2343 label=_('Cooling source'),

2344 queryset=CoolingSource.objects.all(),

2345 to_field_name='name',

2346 help_text=_('Upstream cooling source')

2347 )

2348 location = CSVModelChoiceField(

2349 label=_('Location'),

2350 queryset=Location.objects.all(),

2351 to_field_name='name',

2352 required=False,

2353 help_text=_("Rack's location (if any)")

2354 )

2355 rack = CSVModelChoiceField(

2356 label=_('Rack'),

2357 queryset=Rack.objects.all(),

2358 to_field_name='name',

2359 required=False,

2360 help_text=_('Rack')

2361 )

2362 tenant = CSVModelChoiceField(

2363 queryset=Tenant.objects.all(),

2364 to_field_name='name',

2365 required=False,

2366 help_text=_('Assigned tenant')

2367 )

2368 status = CSVChoiceField(

2369 label=_('Status'),

2370 choices=CoolingFeedStatusChoices,

2371 help_text=_('Operational status')

2372 )

2373 max_flow_unit = CSVChoiceField(

2374 label=_('Max flow unit'),

2375 choices=FlowRateUnitChoices,

2376 required=False,

2377 help_text=_('Unit for maximum flow')

2378 )

2379

2380 class Meta:

2381 model = CoolingFeed

2382 fields = (

2383 'site', 'cooling_source', 'location', 'rack', 'name', 'status',

2384 'cooling_capacity', 'max_flow', 'max_flow_unit', 'tenant', 'description', 'owner',

2385 'comments', 'tags',

2386 )

2387

2388 def __init__(self, data=None, *args, **kwargs):

2389 super().__init__(data, *args, **kwargs)

2390

2391 if data:

2392

2393 # Limit cooling_source queryset by site

2394 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

2395 self.fields['cooling_source'].queryset = self.fields['cooling_source'].queryset.filter(**params)

2396

2397 # Limit location queryset by site

2398 params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}

2399 self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)

2400

2401 # Limit rack queryset by site and group

2402 params = {

2403 f"site__{self.fields['site'].to_field_name}": data.get('site'),

2404 f"location__{self.fields['location'].to_field_name}": data.get('location'),

2405 }

2406 self.fields['rack'].queryset = self.fields['rack'].queryset.filter(**params)

2407

2408

2409class VirtualDeviceContextImportForm(PrimaryModelImportForm):

2410 device = CSVModelChoiceField(

2411 label=_('Device'),

2412 queryset=Device.objects.all(),

2413 to_field_name='name',

2414 help_text=_('Assigned role')

2415 )

2416 tenant = CSVModelChoiceField(

2417 label=_('Tenant'),

2418 queryset=Tenant.objects.all(),

2419 required=False,

2420 to_field_name='name',

2421 help_text=_('Assigned tenant')

2422 )

2423 status = CSVChoiceField(

2424 label=_('Status'),

2425 choices=VirtualDeviceContextStatusChoices,

2426 )

2427 primary_ip4 = CSVModelChoiceField(

2428 label=_('Primary IPv4'),

2429 queryset=IPAddress.objects.all(),

2430 required=False,

2431 to_field_name='address',

2432 help_text=_('IPv4 address with mask, e.g. 1.2.3.4/24')

2433 )

2434 primary_ip6 = CSVModelChoiceField(

2435 label=_('Primary IPv6'),

2436 queryset=IPAddress.objects.all(),

2437 required=False,

2438 to_field_name='address',

2439 help_text=_('IPv6 address with prefix length, e.g. 2001:db8::1/64')

2440 )

2441

2442 class Meta:

2443 fields = [

2444 'name', 'device', 'status', 'tenant', 'identifier', 'owner', 'comments', 'primary_ip4', 'primary_ip6',

2445 ]

2446 model = VirtualDeviceContext

2447

2448 def __init__(self, data=None, *args, **kwargs):

2449 super().__init__(data, *args, **kwargs)

2450

2451 if data:

2452

2453 # Limit primary_ip4/ip6 querysets by assigned device

2454 params = {f"interface__device__{self.fields['device'].to_field_name}": data.get('device')}

2455 self.fields['primary_ip4'].queryset = self.fields['primary_ip4'].queryset.filter(**params)

2456 self.fields['primary_ip6'].queryset = self.fields['primary_ip6'].queryset.filter(**params)

Coverage for dcim/forms/bulk_import.py: 43%

790 statements