RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into feature

This commit is contained in:
Jeremy Stretch 2026-04-23 11:10:58 -04:00
parent c71635510c 7941303d4b
commit df02abbbdf
54 changed files with 1649 additions and 565 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.github/ISSUE_TEMPLATE/03-performance.yaml vendored
View File

@ -35,9 +35,17 @@ body:
required: true
- type: textarea
attributes:
label: Details
label: Observations
description: >
Describe in detail the operations being performed and the indications of a performance issue.
Include any relevant testing parameters, benchmarks, and expected results.
Describe in detail the operations being performed and the indications of a performance issue. Include any
relevant testing parameters, benchmarks, and expected results.
validations:
required: true
- type: textarea
attributes:
label: Proposed Changes
description: >
What specific changes do you propose to improve application performance? (If you're not sure about this,
consider starting a [discussion](https://github.com/netbox-community/netbox/discussions/new/choose) instead.)
validations:
required: true

16
CONTRIBUTING.md
View File

@ -20,7 +20,7 @@ In her book [Working in Public](https://www.amazon.com/Working-Public-Making-Mai
> Stadiums are projects with low contributor growth and high user growth. While they may receive casual contributions, their regular contributor base does not grow proportionately to their users. As a result, they tend to be powered by one or a few developers.
The bulk of NetBox's development is carried out by a handful of core maintainers, with occasional contributions from collaborators in the community. We find the stadium analogy very useful in conveying the roles and obligations of both contributors and users.
The bulk of NetBox's development is carried out by a handful of core maintainers at [NetBox Labs](https://netboxlabs.com), with occasional contributions from collaborators in the community. We find the stadium analogy very useful in conveying the roles and obligations of both contributors and users.
If you're a contributor, actively working on the center stage, you have an obligation to produce quality content that will benefit the project as a whole. Conversely, if you're in the audience consuming the work being produced, you have the option of making requests and suggestions, but must also recognize that contributors are under no obligation to act on them.
@ -34,6 +34,9 @@ NetBox users are welcome to participate in either role, on stage or in the crowd
* Please avoid pinging members with `@` unless they've previously expressed interest or involvement with that particular issue.
* Familiarize yourself with [this list of discussion anti-patterns](https://github.com/bradfitz/issue-tracker-behaviors) and make every effort to avoid them.
> [!CAUTION]
> We do not currently accept issues submitted via GitHub's API: All issues must be submitted using one of the [provided templates](https://github.com/netbox-community/netbox/issues/new/choose). The templates not only help ensure high-quality submissions, but they also automatically assign issue types and labels for categorization. This does not happen when issues are submitted via the API.
## :bug: Reporting Bugs
:warning: Bug reports are used to call attention to some unintended or unexpected behavior in NetBox, such as when an error occurs or when the result of taking some action is inconsistent with the documentation. **Bug reports may not be used to suggest new functionality**; please see "feature requests" below if that is your goal.
@ -58,7 +61,7 @@ intake policy](https://github.com/netbox-community/netbox/wiki/Issue-Intake-Poli
* First, check the GitHub [issues list](https://github.com/netbox-community/netbox/issues?q=is%3Aissue) to see if the feature you have in mind has already been proposed. If you happen to find an open feature request that matches your idea, click "add a reaction" in the top right corner of the issue and add a thumbs up ( :thumbsup: ). This ensures that the issue has a better chance of receiving attention. Also feel free to add a comment with any additional justification for the feature.
* Please don't submit duplicate issues! Sometimes we reject feature requests, for various reasons. Even if you disagree with those reasons, please **do not** submit a duplicate feature request. It is very disrepectful of the maintainers' time, and you may be barred from opening future issues.
* Please don't submit duplicate issues! Sometimes we reject feature requests, for various reasons. Even if you disagree with those reasons, please **do not** submit a duplicate feature request. It is very disrespectful of the maintainers' time, and you may be barred from opening future issues.
* If you have a rough idea that's not quite ready for formal submission yet, start a [GitHub discussion](https://github.com/netbox-community/netbox/discussions) instead. This is a great way to test the viability and narrow down the scope of a new feature prior to submitting a formal proposal, and can serve to generate interest in your idea from other community members.
@ -84,7 +87,7 @@ intake policy](https://github.com/netbox-community/netbox/wiki/Issue-Intake-Poli
* It's very important that you not submit a pull request until a relevant issue has been opened **and** assigned to you. Otherwise, you risk wasting time on work that may ultimately not be needed.
* Community members are limited to a maximum of **three open PRs** at any time. This is to avoid the accumulation of too much parallel work and maintain focus on already PRs under review. If you already have three NetBox PRs open, please wait for at least one of them to be merged (or closed) before opening another.
* Community members are limited to a maximum of **three open PRs** at any time. This is to avoid the accumulation of too much parallel work and maintain focus on PRs already under review. If you already have three NetBox PRs open, please wait for at least one of them to be merged (or closed) before opening another.
* New pull requests should generally be based off of the `main` branch. This branch, in keeping with the [trunk-based development](https://trunkbaseddevelopment.com/) approach, is used for ongoing development and bug fixes and always represents the newest stable code, from which releases are periodically branched. (If you're developing for an upcoming minor release, use `feature` instead.)
@ -93,12 +96,11 @@ intake policy](https://github.com/netbox-community/netbox/wiki/Issue-Intake-Poli
* All code submissions must meet the following criteria (CI will enforce these checks where feasible):
* Consist entirely of original work
* Python syntax is valid
* All tests pass when run with `./manage.py test`
* PEP 8 compliance is enforced, with the exception that lines may be
greater than 80 characters in length
* All tests pass when run with `NETBOX_CONFIGURATION=netbox.configuration_testing ./manage.py test`
* `ruff check` successfully validates style compliance
> [!CAUTION]
> Any contributions which include solely AI-generated or reproduced content will be rejected. All PRs must be submitted by a human.
> Any contributions which include solely AI-generated content will be rejected. All PRs must be submitted by a human.
* Some other tips to keep in mind:
* If you'd like to volunteer for someone else's issue, please post a comment on that issue letting us know. (GitHub allows only people who have commented on an issue to be assigned as its owner.)

4
docs/best-practices/performance-handbook.md
View File

@ -34,7 +34,7 @@ NetBox ships with a reasonable default configuration for most environments, but
#### Reduce the Maximum Page Size
NetBox paginates large result sets to reduce the overall response size. The [`MAX_PAGE_SIZE`](../configuration/miscellaneous.md#max_page_size) parameter specifies the maximum number of results per page that a client can request. This is set to 1,000 by default. Consider lowering this number if you find that API clients are frequently requesting very large result sets.
NetBox paginates large result sets to reduce the overall response size. The [`MAX_PAGE_SIZE`](../configuration/miscellaneous.md#max_page_size) parameter specifies the maximum number of results per page that a client can request. This is set to 1,000 by default. Consider lowering this number if you find that API clients are frequently requesting very large result sets. `MAX_PAGE_SIZE` applies to both the REST API (`?limit=`) and the GraphQL API (`pagination: {limit: …}`), so lowering it reduces the maximum size of responses from either API.
#### Limit GraphQL Aliases
@ -185,3 +185,5 @@ Like the REST API, the GraphQL API supports pagination. Queries which return a l
}
}
```
The requested `limit` is capped by [`MAX_PAGE_SIZE`](../configuration/miscellaneous.md#max_page_size).

6
docs/configuration/miscellaneous.md
View File

@ -45,7 +45,7 @@ Sets content for the top banner in the user interface.
!!! tip
If you'd like the top and bottom banners to match, set the following:
```python
BANNER_TOP = 'Your banner text'
BANNER_BOTTOM = BANNER_TOP
@ -188,7 +188,9 @@ This specifies the URL to use when presenting a map of a physical location by st
Default: `1000`
A web user or API consumer can request an arbitrary number of objects by appending the "limit" parameter to the URL (e.g. `?limit=1000`). This parameter defines the maximum acceptable limit. Setting this to `0` or `None` will allow a client to retrieve _all_ matching objects at once with no limit by specifying `?limit=0`.
Defines the maximum number of objects that may be returned in a single page across the web UI, REST API, and GraphQL API. Setting `MAX_PAGE_SIZE` to `0` or `None` removes the limit.
See the [REST API](../integrations/rest-api.md#pagination) and [GraphQL API](../integrations/graphql-api.md#pagination) pagination documentation for details.
---

33
docs/features/configuration-rendering.md
View File

@ -75,6 +75,39 @@ The configuration can be rendered as JSON or as plaintext by setting the `Accept
* `Accept: application/json`
* `Accept: text/plain`
### Overriding the Config Template
To render a specific config template against a device's context data - rather than the template resolved via the fallback chain above — include `config_template_id` in the request body:
```no-highlight
curl -X POST \
-H "Authorization: Token $TOKEN" \
-H "Content-Type: application/json" \
-H "Accept: application/json; indent=4" \
http://netbox:8000/api/dcim/devices/123/render-config/ \
--data '{
"config_template_id": 42
}'
```
This is useful for rendering partial or alternative templates against a device's assembled context without changing any stored assignments. Any additional keys in the request body are passed into the template as context variables alongside the device's own config context data, as with standard rendering:
```no-highlight
--data '{
"config_template_id": 42,
"environment": "staging"
}'
```
!!! note "Permissions"
Overriding the config template requires the requesting user to have `view` permission for the "Extras > Config Template" object type in addition to the `render_config` permission on the device.
The same override is available in the UI by appending `config_template_id` as a query parameter to the device's render config URL:
```no-highlight
/dcim/devices/123/render-config/?config_template_id=42
```
### General Purpose Use
NetBox config templates can also be rendered without being tied to any specific device, using a separate general purpose REST API endpoint. Any data included with a POST request to this endpoint will be passed as context data for the template.

8
docs/integrations/graphql-api.md
View File

@ -141,7 +141,13 @@ To ensure consistent ordering, objects will always be ordered by their primary k
!!! note "Cursor-based pagination was introduced in NetBox v4.5.2."
Both pagination strategies support passing an optional `limit` parameter. In both approaches, this specifies the maximum number of objects to include in the response. If no limit is specified, a default value of 100 is used.
Both pagination strategies support an optional `limit` parameter specifying the maximum number of objects to include in the response. The [`MAX_PAGE_SIZE`](../configuration/miscellaneous.md#max_page_size) configuration parameter (default `1000`) sets a hard ceiling on this value; if no limit is specified, up to `MAX_PAGE_SIZE` records are returned.
When `MAX_PAGE_SIZE` is set to `0` or `None`:
* Omitting the `pagination` argument entirely returns all matching records.
* Supplying `pagination` without a `limit` returns up to Strawberry Django's default of 100 records.
* Supplying `pagination: {limit: 0}` returns _zero_ records — the opposite of the REST API's `?limit=0` semantics.
### Offset Pagination

1
netbox/core/tests/test_views.py
View File

@ -373,6 +373,7 @@ class SystemTestCase(TestCase):
self.assertIn('plugins', data)
self.assertIn('config', data)
self.assertIn('objects', data)
self.assertIn('db_schema', data)
def test_system_view_with_config_revision(self):
ConfigRevision.objects.create()

1
netbox/core/urls.py
View File

@ -50,6 +50,7 @@ urlpatterns = (
path('config-revisions/<int:pk>/', include(get_model_urls('core', 'configrevision'))),
path('system/', views.SystemView.as_view(), name='system'),
path('system/db-schema/', views.SystemDBSchemaView.as_view(), name='system_db_schema'),
path('plugins/', views.PluginListView.as_view(), name='plugin_list'),
path('plugins/<str:name>/', views.PluginView.as_view(), name='plugin'),

49
netbox/core/utils.py
View File

@ -1,3 +1,4 @@
from django.db import DatabaseError, connection
from django.http import Http404
from django.utils.translation import gettext_lazy as _
from django_rq.queues import get_queue, get_queue_by_index, get_redis_connection
@ -18,6 +19,7 @@ from rq.registry import (
__all__ = (
'delete_rq_job',
'enqueue_rq_job',
'get_db_schema',
'get_rq_jobs',
'get_rq_jobs_from_status',
'requeue_rq_job',
@ -154,3 +156,50 @@ def stop_rq_job(job_id):
queue = get_queue_by_index(queue_index)
return stop_jobs(queue, job_id)[0]
def get_db_schema():
"""
Query the current PostgreSQL schema and return a list of tables, each with its columns and
indexes. Returns an empty list if the database is not accessible.
"""
db_schema = []
try:
with connection.cursor() as cursor:
cursor.execute("""
SELECT table_name, column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_schema = current_schema()
ORDER BY table_name, ordinal_position
""")
columns_by_table = {}
for table_name, column_name, data_type, is_nullable, column_default in cursor.fetchall():
columns_by_table.setdefault(table_name, []).append({
'name': column_name,
'type': data_type,
'nullable': is_nullable == 'YES',
'default': column_default,
})
cursor.execute("""
SELECT tablename, indexname, indexdef
FROM pg_indexes
WHERE schemaname = current_schema()
ORDER BY tablename, indexname
""")
indexes_by_table = {}
for table_name, index_name, index_def in cursor.fetchall():
indexes_by_table.setdefault(table_name, []).append({
'name': index_name,
'definition': index_def,
})
for table_name in sorted(columns_by_table.keys()):
db_schema.append({
'name': table_name,
'columns': columns_by_table[table_name],
'indexes': indexes_by_table.get(table_name, []),
})
except DatabaseError:
pass
return db_schema

119
netbox/core/views.py
View File

@ -3,11 +3,12 @@ import platform
from copy import deepcopy
from django import __version__ as django_version
from django.apps import apps as django_apps_registry
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.mixins import UserPassesTestMixin
from django.core.cache import cache
from django.db import ProgrammingError, connection
from django.db import DatabaseError, connection
from django.http import Http404, HttpResponse, HttpResponseForbidden
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
@ -22,10 +23,18 @@ from rq.job import JobStatus as RQJobStatus
from rq.worker import Worker
from rq.worker_registration import clean_worker_registry
from core.utils import delete_rq_job, enqueue_rq_job, get_rq_jobs_from_status, requeue_rq_job, stop_rq_job
from core.utils import (
delete_rq_job,
enqueue_rq_job,
get_db_schema,
get_rq_jobs_from_status,
requeue_rq_job,
stop_rq_job,
)
from extras.ui.panels import CustomFieldsPanel, TagsPanel
from netbox.config import PARAMS, get_config
from netbox.object_actions import AddObject, BulkDelete, BulkExport, DeleteObject
from netbox.plugins import PluginConfig
from netbox.plugins.utils import get_installed_plugins
from netbox.ui import layout
from netbox.ui.panels import (
@ -199,7 +208,7 @@ class DataFileView(generic.ObjectView):
layout.Row(
layout.Column(
PluginContentPanel('full_width_page'),
)
),
),
)
@ -260,6 +269,12 @@ class JobLogView(generic.ObjectView):
layout.Row(
layout.Column(
ContextTablePanel('table', title=_('Log Entries')),
PluginContentPanel('left_page'),
),
),
layout.Row(
layout.Column(
PluginContentPanel('full_width_page'),
),
),
)
@ -397,6 +412,12 @@ class ConfigRevisionView(generic.ObjectView):
layout.Column(
TemplatePanel('core/panels/configrevision_data.html'),
TemplatePanel('core/panels/configrevision_comment.html'),
PluginContentPanel('left_page'),
),
),
layout.Row(
layout.Column(
PluginContentPanel('full_width_page'),
),
),
)
@ -660,14 +681,13 @@ class WorkerView(BaseRQView):
# System
#
class SystemView(UserPassesTestMixin, View):
def test_func(self):
return self.request.user.is_superuser
def get(self, request):
# System status
def _get_stats(self):
psql_version = db_name = db_size = None
try:
with connection.cursor() as cursor:
@ -676,11 +696,11 @@ class SystemView(UserPassesTestMixin, View):
psql_version = psql_version.split('(')[0].strip()
cursor.execute("SELECT current_database()")
db_name = cursor.fetchone()[0]
cursor.execute(f"SELECT pg_size_pretty(pg_database_size('{db_name}'))")
cursor.execute("SELECT pg_size_pretty(pg_database_size(current_database()))")
db_size = cursor.fetchone()[0]
except (ProgrammingError, IndexError):
except (DatabaseError, IndexError):
pass
stats = {
return {
'netbox_release': settings.RELEASE,
'django_version': django_version,
'python_version': platform.python_version(),
@ -690,23 +710,23 @@ class SystemView(UserPassesTestMixin, View):
'rq_worker_count': Worker.count(get_connection('default')),
}
# Django apps
django_apps = get_installed_apps()
# Configuration
config = get_config()
# Plugins
plugins = get_installed_plugins()
# Object counts
def _get_object_counts(self):
objects = {}
for ot in ObjectType.objects.public().order_by('app_label', 'model'):
if model := ot.model_class():
objects[ot] = model.objects.count()
return objects
def get(self, request):
stats = self._get_stats()
django_apps = get_installed_apps()
config = get_config()
plugins = get_installed_plugins()
objects = self._get_object_counts()
# Raw data export
if 'export' in request.GET:
db_schema = get_db_schema()
stats['netbox_release'] = stats['netbox_release'].asdict()
params = [param.name for param in PARAMS]
data = {
@ -719,6 +739,12 @@ class SystemView(UserPassesTestMixin, View):
'objects': {
f'{ot.app_label}.{ot.model}': count for ot, count in objects.items()
},
'db_schema': {
table['name']: {
'columns': table['columns'],
'indexes': table['indexes'],
} for table in db_schema
},
}
response = HttpResponse(json.dumps(data, cls=ConfigJSONEncoder, indent=4), content_type='text/json')
response['Content-Disposition'] = 'attachment; filename="netbox.json"'
@ -738,6 +764,61 @@ class SystemView(UserPassesTestMixin, View):
})
class SystemDBSchemaView(UserPassesTestMixin, View):
def test_func(self):
return self.request.user.is_superuser
@staticmethod
def _get_db_schema_groups(db_schema):
plugin_app_labels = {
app_config.label
for app_config in django_apps_registry.get_app_configs()
if isinstance(app_config, PluginConfig)
}
# Sort longest-first so "netbox_branching" matches before "netbox"
sorted_plugin_labels = sorted(plugin_app_labels, key=len, reverse=True)
groups = {}
for table in db_schema:
matched_plugin = next(
(label for label in sorted_plugin_labels if table['name'].startswith(label + '_')),
None,
)
if matched_plugin:
prefix = matched_plugin
elif '_' in table['name']:
prefix = table['name'].split('_')[0]
else:
prefix = 'other'
groups.setdefault(prefix, []).append(table)
return sorted(
[
{
'name': name,
'tables': tables,
'index_count': sum(len(t['indexes']) for t in tables),
'is_plugin': name in plugin_app_labels,
}
for name, tables in groups.items()
],
key=lambda g: (g['is_plugin'], g['name']),
)
def get(self, request):
db_schema = get_db_schema()
db_schema_groups = self._get_db_schema_groups(db_schema)
db_schema_stats = {
'total_tables': len(db_schema),
'total_columns': sum(len(t['columns']) for t in db_schema),
'total_indexes': sum(len(t['indexes']) for t in db_schema),
}
return render(request, 'core/htmx/system_db_schema.html', {
'db_schema': db_schema,
'db_schema_groups': db_schema_groups,
'db_schema_stats': db_schema_stats,
})
#
# Plugins
#

13
netbox/dcim/filtersets.py
View File

@ -1592,6 +1592,19 @@ class VirtualDeviceContextFilterSet(PrimaryModelFilterSet, TenancyFilterSet, Pri
@register_filterset
class ModuleFilterSet(PrimaryModelFilterSet):
profile_id = django_filters.ModelMultipleChoiceFilter(
field_name='module_type__profile',
queryset=ModuleTypeProfile.objects.all(),
distinct=False,
label=_('Profile (ID)'),
)
profile = django_filters.ModelMultipleChoiceFilter(
field_name='module_type__profile__name',
queryset=ModuleTypeProfile.objects.all(),
distinct=False,
to_field_name='name',
label=_('Profile (name)'),
)
manufacturer_id = django_filters.ModelMultipleChoiceFilter(
field_name='module_type__manufacturer',
queryset=Manufacturer.objects.all(),

14
netbox/dcim/forms/filtersets.py
View File

@ -719,6 +719,7 @@ class ModuleTypeFilterForm(PrimaryModelFilterSetForm):
profile_id = DynamicModelMultipleChoiceField(
queryset=ModuleTypeProfile.objects.all(),
required=False,
null_option='None',
label=_('Profile')
)
manufacturer_id = DynamicModelMultipleChoiceField(
@ -1072,9 +1073,13 @@ class ModuleFilterForm(LocalConfigContextFilterForm, TenancyFilterForm, PrimaryM
fieldsets = (
FieldSet('q', 'filter_id', 'tag'),
FieldSet('region_id', 'site_group_id', 'site_id', 'location_id', 'rack_id', 'device_id', name=_('Location')),
FieldSet('manufacturer_id', 'module_type_id', 'status', 'serial', 'asset_tag', name=_('Hardware')),
FieldSet(
'profile_id', 'manufacturer_id', 'module_type_id', 'status', 'serial', 'asset_tag',
name=_('Hardware')
),
FieldSet('owner_group_id', 'owner_id', name=_('Ownership')),
)
selector_fields = ('filter_id', 'q', 'manufacturer_id', 'profile_id')
device_id = DynamicModelMultipleChoiceField(
queryset=Device.objects.all(),
required=False,
@ -1127,10 +1132,17 @@ class ModuleFilterForm(LocalConfigContextFilterForm, TenancyFilterForm, PrimaryM
required=False,
label=_('Manufacturer')
)
profile_id = DynamicModelMultipleChoiceField(
queryset=ModuleTypeProfile.objects.all(),
required=False,
null_option='None',
label=_('Profile')
)
module_type_id = DynamicModelMultipleChoiceField(
queryset=ModuleType.objects.all(),
required=False,
query_params={
'profile_id': '$profile_id',
'manufacturer_id': '$manufacturer_id'
},
label=_('Type')

2
netbox/dcim/forms/model_forms.py
View File

@ -2011,7 +2011,7 @@ class MACAddressForm(PrimaryModelForm):
class Meta:
model = MACAddress
fields = [
'mac_address', 'interface', 'vminterface', 'description', 'owner', 'tags',
'mac_address', 'interface', 'vminterface', 'description', 'owner', 'comments', 'tags',
]
def __init__(self, *args, **kwargs):

18
netbox/dcim/forms/object_import.py
View File

@ -150,9 +150,25 @@ class PortTemplateMappingImportForm(forms.ModelForm):
class Meta:
model = PortTemplateMapping
fields = [
'front_port', 'front_port_position', 'rear_port', 'rear_port_position',
'device_type', 'module_type', 'front_port', 'front_port_position', 'rear_port', 'rear_port_position',
]
def clean_device_type(self):
if device_type := self.cleaned_data['device_type']:
front_port = self.fields['front_port']
rear_port = self.fields['rear_port']
front_port.queryset = front_port.queryset.filter(device_type=device_type)
rear_port.queryset = rear_port.queryset.filter(device_type=device_type)
return device_type
def clean_module_type(self):
if module_type := self.cleaned_data['module_type']:
front_port = self.fields['front_port']
rear_port = self.fields['rear_port']
front_port.queryset = front_port.queryset.filter(module_type=module_type)
rear_port.queryset = rear_port.queryset.filter(module_type=module_type)
return module_type
class ModuleBayTemplateImportForm(forms.ModelForm):

8
netbox/dcim/tables/modules.py
View File

@ -93,6 +93,11 @@ class ModuleTable(PrimaryModelTable):
accessor=tables.A('module_type__manufacturer'),
linkify=True
)
profile = tables.Column(
verbose_name=_('Profile'),
accessor=tables.A('module_type__profile'),
linkify=True,
)
module_type = tables.Column(
verbose_name=_('Module Type'),
linkify=True
@ -107,7 +112,8 @@ class ModuleTable(PrimaryModelTable):
class Meta(PrimaryModelTable.Meta):
model = Module
fields = (
'pk', 'id', 'device', 'module_bay', 'manufacturer', 'module_type', 'status', 'serial', 'asset_tag',
'pk', 'id', 'device', 'module_bay', 'manufacturer', 'profile', 'module_type', 'status',
'serial', 'asset_tag',
'description', 'comments', 'tags', 'created', 'last_updated',
)
default_columns = (

79
netbox/dcim/tests/test_api.py
View File

@ -1,5 +1,6 @@
import json
from django.conf import settings
from django.test import override_settings, tag
from django.urls import reverse
from django.utils.translation import gettext as _
@ -1983,6 +1984,41 @@ class DeviceTest(APIViewTestCases.APIViewTestCase):
self.assertEqual(response.data['oob_ip']['nat_inside']['address'], str(real_ip.address))
self.assertEqual(response.data['oob_ip']['nat_outside'], [])
def test_render_config_with_config_template_id(self):
default_template = ConfigTemplate.objects.create(
name='Default Template',
template_code='Default config for {{ device.name }}'
)
override_template = ConfigTemplate.objects.create(
name='Override Template',
template_code='Override config for {{ device.name }}'
)
device = Device.objects.first()
device.config_template = default_template
device.save()
self.add_permissions('dcim.render_config_device', 'dcim.view_device', 'extras.view_configtemplate')
url = reverse('dcim-api:device-render-config', kwargs={'pk': device.pk})
# Render with override template
response = self.client.post(url, {'config_template_id': override_template.pk}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(response.data['content'], f'Override config for {device.name}')
# Render with nonexistent config_template_id
response = self.client.post(url, {'config_template_id': 999999}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
# Render with non-integer config_template_id
response = self.client.post(url, {'config_template_id': 'abc'}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
# Without view_configtemplate permission, override template should not be accessible
self.remove_permissions('extras.view_configtemplate')
response = self.client.post(url, {'config_template_id': override_template.pk}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
class ModuleTest(APIViewTestCases.APIViewTestCase):
model = Module
@ -1990,16 +2026,23 @@ class ModuleTest(APIViewTestCases.APIViewTestCase):
bulk_update_data = {
'serial': '1234ABCD',
}
user_permissions = ('dcim.view_modulebay', 'dcim.view_moduletype', 'dcim.view_device')
user_permissions = (
'dcim.view_modulebay', 'dcim.view_moduletype', 'dcim.view_moduletypeprofile', 'dcim.view_device'
)
@classmethod
def setUpTestData(cls):
manufacturer = Manufacturer.objects.create(name='Generic', slug='generic')
profiles = (
ModuleTypeProfile(name='Test CPU'),
ModuleTypeProfile(name='Test Hard disk'),
)
ModuleTypeProfile.objects.bulk_create(profiles)
device = create_test_device('Test Device 1')
module_types = (
ModuleType(manufacturer=manufacturer, model='Module Type 1'),
ModuleType(manufacturer=manufacturer, model='Module Type 2'),
ModuleType(manufacturer=manufacturer, model='Module Type 1', profile=profiles[0]),
ModuleType(manufacturer=manufacturer, model='Module Type 2', profile=profiles[1]),
ModuleType(manufacturer=manufacturer, model='Module Type 3'),
)
ModuleType.objects.bulk_create(module_types)
@ -2281,6 +2324,36 @@ class ModuleTest(APIViewTestCases.APIViewTestCase):
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
def test_list_objects_by_profile_id(self):
profiles = ModuleTypeProfile.objects.filter(name__startswith='Test').order_by('name')
self.add_permissions('dcim.view_module')
response = self.client.get(self._get_list_url(), {'profile_id': [profiles[0].pk]}, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data['results']), 1)
response = self.client.get(self._get_list_url(), {'profile_id': [profiles[1].pk]}, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data['results']), 1)
response = self.client.get(
self._get_list_url(),
{'profile_id': [settings.FILTERS_NULL_CHOICE_VALUE]},
**self.header,
)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data['results']), 1)
def test_list_objects_by_profile(self):
profiles = ModuleTypeProfile.objects.filter(name__startswith='Test').order_by('name')
self.add_permissions('dcim.view_module')
response = self.client.get(self._get_list_url(), {'profile': [profiles[0].name]}, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data['results']), 1)
response = self.client.get(self._get_list_url(), {'profile': [profiles[1].name]}, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data['results']), 1)
class ConsolePortTest(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTestCase):
model = ConsolePort

23
netbox/dcim/tests/test_filtersets.py
View File

@ -1,3 +1,4 @@
from django.conf import settings
from django.test import TestCase
from circuits.models import Circuit, CircuitTermination, CircuitType, Provider
@ -3183,6 +3184,11 @@ class ModuleTestCase(TestCase, ChangeLoggedFilterSetTests):
Manufacturer(name='Manufacturer 3', slug='manufacturer-3'),
)
Manufacturer.objects.bulk_create(manufacturers)
module_type_profiles = (
ModuleTypeProfile(name='Test CPU'),
ModuleTypeProfile(name='Test Hard disk'),
)
ModuleTypeProfile.objects.bulk_create(module_type_profiles)
device_types = (
DeviceType(manufacturer=manufacturers[0], model='Device Type 1', slug='device-type-1'),
@ -3246,8 +3252,8 @@ class ModuleTestCase(TestCase, ChangeLoggedFilterSetTests):
Device.objects.bulk_create(devices)
module_types = (
ModuleType(manufacturer=manufacturers[0], model='Module Type 1'),
ModuleType(manufacturer=manufacturers[1], model='Module Type 2'),
ModuleType(manufacturer=manufacturers[0], model='Module Type 1', profile=module_type_profiles[0]),
ModuleType(manufacturer=manufacturers[1], model='Module Type 2', profile=module_type_profiles[1]),
ModuleType(manufacturer=manufacturers[2], model='Module Type 3'),
)
ModuleType.objects.bulk_create(module_types)
@ -3363,6 +3369,19 @@ class ModuleTestCase(TestCase, ChangeLoggedFilterSetTests):
params = {'module_type': [module_types[0].model, module_types[1].model]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6)
def test_profile(self):
profiles = ModuleTypeProfile.objects.filter(name__startswith='Test').order_by('name')
params = {'profile_id': [profiles[0].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
params = {'profile': [profiles[0].name]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
params = {'profile_id': [profiles[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
params = {'profile': [profiles[1].name]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
params = {'profile_id': [settings.FILTERS_NULL_CHOICE_VALUE]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
def test_description(self):
params = {'description': ['foobar1', 'foobar2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)

3
netbox/dcim/tests/test_tables.py
View File

@ -70,6 +70,9 @@ class ModuleTypeTableTest(TableTestCases.StandardTableTestCase):
class ModuleTableTest(TableTestCases.StandardTableTestCase):
table = ModuleTable
def test_profile_column_available(self):
self.assertIn('profile', self.table.base_columns)
#
# Devices

68
netbox/dcim/tests/test_views.py
View File

@ -196,6 +196,28 @@ class SiteTestCase(ViewTestCases.PrimaryObjectViewTestCase):
'description': 'New description',
}
def test_get_object_with_only_site_view_permission_hides_unauthorized_embedded_panels(self):
site = self._get_queryset().first()
obj_perm = ObjectPermission(
name='Test permission',
actions=['view'],
)
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
response = self.client.get(site.get_absolute_url())
self.assertHttpStatus(response, 200)
for panel, url in (
('locations', reverse('dcim:location_list')),
('devices', reverse('dcim:device_list')),
('image attachments', reverse('extras:imageattachment_list')),
):
with self.subTest(panel=panel):
self.assertNotContains(response, url)
class LocationTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
model = Location
@ -2413,6 +2435,43 @@ class DeviceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
self.remove_permissions('dcim.view_device')
self.assertHttpStatus(self.client.get(url), 403)
def test_device_renderconfig_with_config_template_id(self):
default_template = ConfigTemplate.objects.create(
name='Default Template',
template_code='Default config for {{ device.name }}'
)
override_template = ConfigTemplate.objects.create(
name='Override Template',
template_code='Override config for {{ device.name }}'
)
device = Device.objects.first()
device.config_template = default_template
device.save()
self.add_permissions('dcim.view_device', 'dcim.render_config_device', 'extras.view_configtemplate')
url = reverse('dcim:device_render-config', kwargs={'pk': device.pk})
# Render with override config_template_id
response = self.client.get(url, {'config_template_id': override_template.pk})
self.assertHttpStatus(response, 200)
self.assertIn(b'Override config for', response.content)
# Render with nonexistent config_template_id still returns 200 with error message
response = self.client.get(url, {'config_template_id': 999999})
self.assertHttpStatus(response, 200)
self.assertIn(b'Error rendering template', response.content)
# Render with non-integer config_template_id still returns 200 with error message
response = self.client.get(url, {'config_template_id': 'abc'})
self.assertHttpStatus(response, 200)
self.assertIn(b'Error rendering template', response.content)
# Without view_configtemplate permission, override template should not be accessible
self.remove_permissions('extras.view_configtemplate')
response = self.client.get(url, {'config_template_id': override_template.pk})
self.assertHttpStatus(response, 200)
self.assertIn(b'Error rendering template', response.content)
def test_device_role_display_colored(self):
parent_role = DeviceRole.objects.create(name='Parent Role', slug='parent-role', color='111111')
child_role = DeviceRole.objects.create(name='Child Role', slug='child-role', parent=parent_role, color='aa00bb')
@ -2475,13 +2534,14 @@ class ModuleTestCase(
@classmethod
def setUpTestData(cls):
manufacturer = Manufacturer.objects.create(name='Generic', slug='generic')
module_type_profile = ModuleTypeProfile.objects.create(name='Module Type Profile 1')
devices = (
create_test_device('Device 1'),
create_test_device('Device 2'),
)
module_types = (
ModuleType(manufacturer=manufacturer, model='Module Type 1'),
ModuleType(manufacturer=manufacturer, model='Module Type 1', profile=module_type_profile),
ModuleType(manufacturer=manufacturer, model='Module Type 2'),
ModuleType(manufacturer=manufacturer, model='Module Type 3'),
ModuleType(manufacturer=manufacturer, model='Module Type 4'),
@ -2540,6 +2600,12 @@ class ModuleTestCase(
f"{modules[2].pk},offline,Serial 1",
)
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_module_detail_includes_module_type_profile(self):
response = self.client.get(self._get_queryset().first().get_absolute_url())
self.assertContains(response, 'Module Type Profile 1')
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_module_component_replication(self):
self.add_permissions('dcim.add_module')

26
netbox/extras/api/mixins.py
View File

@ -1,8 +1,10 @@
from django.utils.translation import gettext as _
from rest_framework.decorators import action
from rest_framework.renderers import JSONRenderer
from rest_framework.response import Response
from rest_framework.status import HTTP_400_BAD_REQUEST, HTTP_500_INTERNAL_SERVER_ERROR
from extras.models import ConfigTemplate
from netbox.api.authentication import TokenWritePermission
from netbox.api.renderers import TextRenderer
@ -76,7 +78,7 @@ class RenderConfigMixin(ConfigTemplateRenderMixin):
@action(detail=True, methods=['post'], url_path='render-config', renderer_classes=[JSONRenderer, TextRenderer])
def render_config(self, request, pk):
"""
Resolve and render the preferred ConfigTemplate for this Device.
Resolve and render the preferred ConfigTemplate for this Device or Virtual Machine.
"""
# Override restrict() on the default queryset to enforce the render_config & view actions
self.queryset = self.queryset.model.objects.restrict(request.user, 'render_config').restrict(
@ -85,15 +87,25 @@ class RenderConfigMixin(ConfigTemplateRenderMixin):
instance = self.get_object()
object_type = instance._meta.model_name
configtemplate = instance.get_config_template()
if not configtemplate:
return Response({
'error': f'No config template found for this {object_type}.'
}, status=HTTP_400_BAD_REQUEST)
# Check for an optional config_template_id override in the request data
if config_template_id := request.data.get('config_template_id'):
try:
configtemplate = ConfigTemplate.objects.restrict(request.user, 'view').get(pk=config_template_id)
except (ConfigTemplate.DoesNotExist, ValueError):
return Response({
'error': _('Config template with ID {id} not found.').format(id=config_template_id)
}, status=HTTP_400_BAD_REQUEST)
else:
configtemplate = instance.get_config_template()
if not configtemplate:
return Response({
'error': _('No config template found for this {object_type}.').format(object_type=object_type)
}, status=HTTP_400_BAD_REQUEST)
# Compile context data
context_data = instance.get_config_context()
context_data.update(request.data)
context_data.update({k: v for k, v in request.data.items() if k != 'config_template_id'})
context_data.update({object_type: instance})
return self.render_configtemplate(request, configtemplate, context_data)

10
netbox/extras/api/serializers_/scripts.py
View File

@ -9,6 +9,7 @@ from rest_framework import serializers
from core.api.serializers_.jobs import JobSerializer
from core.choices import JobNotificationChoices, ManagedFileRootPathChoices
from extras.models import Script, ScriptModule
from extras.utils import validate_script_content
from netbox.api.serializers import ValidatedModelSerializer
from utilities.datetime import local_now
@ -39,6 +40,15 @@ class ScriptModuleSerializer(ValidatedModelSerializer):
data = super().validate(data)
data.pop('file_root', None)
if file is not None:
# Validate that the uploaded script can be loaded as a Python module
content = file.read()
file.seek(0)
try:
validate_script_content(content, file.name)
except Exception as e:
raise serializers.ValidationError(
_("Error loading script: {error}").format(error=e)
)
data['file'] = file
return data

17
netbox/extras/forms/scripts.py
View File

@ -4,6 +4,7 @@ from django.utils.translation import gettext_lazy as _
from core.choices import JobIntervalChoices, JobNotificationChoices
from core.forms import ManagedFileForm
from extras.utils import validate_script_content
from utilities.datetime import local_now
from utilities.forms.widgets import DateTimePicker, NumberWithOptions
@ -76,6 +77,22 @@ class ScriptFileForm(ManagedFileForm):
"""
ManagedFileForm with a custom save method to use django-storages.
"""
def clean(self):
super().clean()
if upload_file := self.cleaned_data.get('upload_file'):
# Validate that the uploaded script can be loaded as a Python module
content = upload_file.read()
upload_file.seek(0)
try:
validate_script_content(content, upload_file.name)
except Exception as e:
raise forms.ValidationError(
_("Error loading script: {error}").format(error=e)
)
return self.cleaned_data
def save(self, *args, **kwargs):
# If a file was uploaded, save it to disk
if self.cleaned_data['upload_file']:

7
netbox/extras/models/customfields.py
View File

@ -601,7 +601,12 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
# Object
elif self.type == CustomFieldTypeChoices.TYPE_OBJECT:
model = self.related_object_type.model_class()
field_class = CSVModelChoiceField if for_csv_import else DynamicModelChoiceField
if for_csv_import:
field_class = CSVModelChoiceField
elif for_filterset_form:
field_class = DynamicModelMultipleChoiceField
else:
field_class = DynamicModelChoiceField
kwargs = {
'queryset': model.objects.all(),
'required': required,

15
netbox/extras/tests/test_api.py
View File

@ -1450,6 +1450,21 @@ class ScriptModuleTest(APITestCase):
self.assertTrue(ScriptModule.objects.filter(file_path='test_upload.py').exists())
self.assertTrue(Script.objects.filter(module__file_path='test_upload.py', name='TestScript').exists())
def test_upload_faulty_script_module(self):
"""Uploading a script with an import error should return 400 and not create a DB record."""
self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
# 'extras.script' is invalid; the correct module is 'extras.scripts'
script_content = b"from extras.script import Script\nclass TestScript(Script):\n pass\n"
upload_file = SimpleUploadedFile('test_faulty.py', script_content, content_type='text/plain')
response = self.client.post(
self.url,
{'file': upload_file},
format='multipart',
**self.header,
)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
self.assertFalse(ScriptModule.objects.filter(file_path='test_faulty.py').exists())
def test_upload_script_module_without_file_fails(self):
self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
response = self.client.post(self.url, {}, format='json', **self.header)

13
netbox/extras/utils.py
View File

@ -1,4 +1,5 @@
import importlib
import types
from pathlib import Path
from django.core.exceptions import ImproperlyConfigured, SuspiciousFileOperation
@ -21,6 +22,7 @@ __all__ = (
'is_script',
'is_taggable',
'run_validators',
'validate_script_content',
)
@ -134,6 +136,17 @@ def is_script(obj):
return False
def validate_script_content(content, filename):
"""
Validate that the given content can be loaded as a Python module by compiling
and executing it. Raises an exception if the script cannot be loaded.
"""
code = compile(content, filename, 'exec')
module_name = Path(filename).stem
module = types.ModuleType(module_name)
exec(code, module.__dict__)
def is_report(obj):
"""
Returns True if the given object is a Report.

13
netbox/extras/views.py
View File

@ -1280,10 +1280,19 @@ class ObjectRenderConfigView(generic.ObjectView):
context_data = instance.get_config_context()
context_data.update(self.get_extra_context_data(request, instance))
# Check for an optional config_template_id override in the query params
config_template = None
error_message = ''
if config_template_id := request.GET.get('config_template_id'):
try:
config_template = ConfigTemplate.objects.restrict(request.user, 'view').get(pk=config_template_id)
except (ConfigTemplate.DoesNotExist, ValueError):
error_message = _("Config template with ID {id} not found.").format(id=config_template_id)
else:
config_template = instance.get_config_template()
# Render the config template
rendered_config = None
error_message = ''
config_template = instance.get_config_template()
if config_template:
try:
rendered_config = config_template.render(context=context_data)

4
netbox/ipam/api/views.py
View File

@ -392,7 +392,7 @@ class AvailablePrefixesView(AvailableObjectsView):
@extend_schema(
methods=["post"],
responses={201: serializers.PrefixSerializer(many=True)},
request=serializers.PrefixSerializer(many=True),
request=serializers.PrefixLengthSerializer(many=True),
)
def post(self, request, pk):
return super().post(request, pk)
@ -488,7 +488,7 @@ class AvailableVLANsView(AvailableObjectsView):
@extend_schema(
methods=["post"],
responses={201: serializers.VLANSerializer(many=True)},
request=serializers.VLANSerializer(many=True),
request=serializers.CreateAvailableVLANSerializer(many=True),
)
def post(self, request, pk):
return super().post(request, pk)

23
netbox/ipam/tests/test_tables.py
View File

@ -1,4 +1,4 @@
from django.test import RequestFactory, TestCase
from django.test import TestCase
from netaddr import IPNetwork
from ipam.models import FHRPGroupAssignment, IPAddress, IPRange, Prefix
@ -24,7 +24,8 @@ class AnnotatedIPAddressTableTest(TestCase):
cls.ip_range = IPRange.objects.create(
start_address=IPNetwork('10.1.1.2/24'),
end_address=IPNetwork('10.1.1.10/24'),
status='active'
status='active',
mark_populated=True,
)
def test_ipaddress_has_checkbox_iprange_does_not(self):
@ -32,14 +33,18 @@ class AnnotatedIPAddressTableTest(TestCase):
table = AnnotatedIPAddressTable(data, orderable=False)
table.columns.show('pk')
request = RequestFactory().get('/')
html = table.as_html(request)
ipaddress_row = next(
row for row in table.rows
if isinstance(row.record, IPAddress) and row.record.pk == self.ip_address.pk
)
iprange_row = next(
row for row in table.rows
if isinstance(row.record, IPRange) and row.record.pk == self.ip_range.pk
)
ipaddress_checkbox_count = html.count(f'name="pk" value="{self.ip_address.pk}"')
self.assertEqual(ipaddress_checkbox_count, 1)
iprange_checkbox_count = html.count(f'name="pk" value="{self.ip_range.pk}"')
self.assertEqual(iprange_checkbox_count, 0)
self.assertIn('name="pk"', ipaddress_row.get_cell('pk'))
self.assertIn(f'value="{self.ip_address.pk}"', ipaddress_row.get_cell('pk'))
self.assertNotIn('name="pk"', iprange_row.get_cell('pk'))
def test_annotate_ip_space_ipv4_non_pool_excludes_network_and_broadcast(self):
prefix = Prefix.objects.create(

19
netbox/netbox/middleware.py
View File

@ -8,6 +8,7 @@ from django.core.exceptions import ImproperlyConfigured
from django.db import ProgrammingError, connection
from django.db.utils import InternalError
from django.http import Http404, HttpResponseRedirect
from django.middleware.common import CommonMiddleware as DjangoCommonMiddleware
from django_prometheus import middleware
from netbox.config import clear_config, get_config
@ -18,6 +19,7 @@ from utilities.error_handlers import handle_rest_api_exception
from utilities.request import apply_request_processors
__all__ = (
'CommonMiddleware',
'CoreMiddleware',
'MaintenanceModeMiddleware',
'PrometheusAfterMiddleware',
@ -26,6 +28,23 @@ __all__ = (
)
class CommonMiddleware(DjangoCommonMiddleware):
"""
Subclass of Django's CommonMiddleware that suppresses the APPEND_SLASH
redirect for REST API requests using an unsafe HTTP method. Redirecting a
POST/PUT/PATCH/DELETE to a trailing-slash URL would either drop the request
body (clients downgrade to GET on a 302) or raise a RuntimeError when
DEBUG is enabled. Letting the original 404 propagate gives the caller a
clear, actionable error instead.
"""
UNSAFE_METHODS = frozenset(('DELETE', 'PATCH', 'POST', 'PUT'))
def should_redirect_with_slash(self, request):
if request.method in self.UNSAFE_METHODS and is_api_request(request):
return False
return super().should_redirect_with_slash(request)
class CoreMiddleware:
def __init__(self, get_response):

2
netbox/netbox/settings.py
View File

@ -491,7 +491,7 @@ MIDDLEWARE = [
'corsheaders.middleware.CorsMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.locale.LocaleMiddleware',
'django.middleware.common.CommonMiddleware',
'netbox.middleware.CommonMiddleware', # Replaces django.middleware.common.CommonMiddleware
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',

15
netbox/netbox/tables/tables.py
View File

@ -116,16 +116,23 @@ class BaseTable(tables.Table):
self.sequence.remove('actions')
self.sequence.append('actions')
def _apply_prefetching(self):
def _apply_prefetching(self, columns=None):
"""
Dynamically update the table's QuerySet to ensure related fields are pre-fetched
Dynamically update the table's QuerySet to ensure related fields are pre-fetched.
Args:
columns: An optional iterable of column names for which to apply prefetching,
regardless of visibility. If None, only currently visible columns are used.
"""
if not isinstance(self.data, TableQuerysetData):
return
prefetch_fields = []
for column in self.columns:
if not column.visible:
for column in self.columns.iterall():
if columns is not None:
if column.name not in columns:
continue
elif not column.visible:
# Skip hidden columns
continue
model = getattr(self.Meta, 'model') # Must be called *after* resolving columns

9
netbox/netbox/tests/test_base_classes.py
View File

@ -45,6 +45,7 @@ from netbox.graphql.types import (
PrimaryObjectType,
)
from netbox.models import NestedGroupModel, NetBoxModel, OrganizationalModel, PrimaryModel
from netbox.registry import registry
from netbox.tables import (
NestedGroupModelTable,
NetBoxTable,
@ -174,11 +175,10 @@ class FilterSetClassesTestCase(TestCase):
@staticmethod
def get_filterset_for_model(model):
"""
Import and return the filterset class for a given model.
Return the filterset class for a given model from the application registry.
"""
app_label = model._meta.app_label
model_name = model.__name__
return import_string(f'{app_label}.filtersets.{model_name}FilterSet')
label = f'{model._meta.app_label}.{model._meta.model_name}'
return registry['filtersets'].get(label)
@staticmethod
def get_model_filterset_base_class(model):
@ -204,6 +204,7 @@ class FilterSetClassesTestCase(TestCase):
for model in apps.get_models():
if base_class := self.get_model_filterset_base_class(model):
filterset = self.get_filterset_for_model(model)
self.assertIsNotNone(filterset, f"No registered filterset found for model {model}")
self.assertTrue(
issubclass(filterset, base_class),
f"{filterset} does not inherit from {base_class}",

32
netbox/netbox/tests/test_tables.py
View File

@ -47,6 +47,38 @@ class BaseTableTest(TestCase):
prefetch_lookups = table.data.data._prefetch_related_lookups
self.assertEqual(prefetch_lookups, tuple())
def test_prefetch_all_columns_for_export(self):
"""
Verify that related fields for *all* table columns are prefetched when preparing a CSV
export, including columns which are not currently visible in the user's configured view.
"""
request = RequestFactory().get('/')
request.user = self.user
# Configure the table with only local-field columns visible. Related columns like 'site',
# 'rack', and 'region' are hidden in the user's view.
self.user.config.set(
'tables.DeviceTable.columns',
['name', 'status'],
commit=True,
)
table = DeviceTable(Device.objects.all())
table.configure(request)
# With only local-field columns visible, no relations should be prefetched yet.
self.assertEqual(table.data.data._prefetch_related_lookups, tuple())
# Simulate the CSV "All data" export path: re-apply prefetching for every column that
# will be included in the export, regardless of visibility.
export_columns = [
col_name for col_name, _ in table.selected_columns + table.available_columns
]
table._apply_prefetching(columns=export_columns)
prefetch_lookups = table.data.data._prefetch_related_lookups
self.assertIn('rack', prefetch_lookups)
self.assertIn('site__region', prefetch_lookups)
def test_configure_anonymous_user_with_ordering(self):
"""
Verify that table.configure() does not raise an error when an anonymous

59
netbox/netbox/tests/test_ui.py
View File

@ -1,6 +1,6 @@
from types import SimpleNamespace
from django.test import TestCase
from django.test import RequestFactory, TestCase
from circuits.choices import CircuitStatusChoices, VirtualCircuitTerminationRoleChoices
from circuits.models import (
@ -10,9 +10,12 @@ from circuits.models import (
VirtualCircuitTermination,
VirtualCircuitType,
)
from core.models import ObjectType
from dcim.choices import InterfaceTypeChoices
from dcim.models import Interface
from dcim.models import Interface, Site
from netbox.ui import attrs
from netbox.ui.panels import ObjectsTablePanel
from users.models import ObjectPermission, User
from utilities.testing import create_test_device
from vpn.choices import (
AuthenticationAlgorithmChoices,
@ -390,3 +393,55 @@ class DateTimeAttrTest(TestCase):
obj = SimpleNamespace(created='2024-01-01')
context = attr.get_context(obj, 'created', '2024-01-01', {})
self.assertEqual(context['spec'], 'minutes')
class ObjectsTablePanelTest(TestCase):
"""
Verify that ObjectsTablePanel.should_render() hides the panel when
the requesting user lacks view permission for the panel's model.
"""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(username='test_user', password='test_password')
# Grant view permission only for Site
obj_perm = ObjectPermission.objects.create(
name='View sites only',
actions=['view'],
)
obj_perm.object_types.add(ObjectType.objects.get_for_model(Site))
obj_perm.users.add(cls.user)
def setUp(self):
self.factory = RequestFactory()
self.panel = ObjectsTablePanel(model='dcim.site')
self.panel_no_perm = ObjectsTablePanel(model='dcim.location')
def _make_context(self, user=None):
if user is None:
return {}
request = self.factory.get('/')
request.user = user
return {'request': request}
def test_should_render_without_request(self):
"""
Panel should render when no request is present in context.
"""
context = self.panel.get_context({})
self.assertTrue(self.panel.should_render(context))
def test_should_render_with_permission(self):
"""
Panel should render when the user has view permission for the panel's model.
"""
context = self.panel.get_context(self._make_context(self.user))
self.assertTrue(self.panel.should_render(context))
def test_should_not_render_without_permission(self):
"""
Panel should be hidden when the user lacks view permission for the panel's model.
"""
context = self.panel_no_perm.get_context(self._make_context(self.user))
self.assertFalse(self.panel_no_perm.should_render(context))

11
netbox/netbox/ui/panels.py
View File

@ -5,6 +5,7 @@ from django.utils.translation import gettext_lazy as _
from netbox.ui import attrs
from netbox.ui.actions import CopyContent
from utilities.data import resolve_attr_path
from utilities.permissions import get_permission_for_model
from utilities.querydict import dict_to_querydict
from utilities.string import title
from utilities.templatetags.plugins import _get_registered_content
@ -347,6 +348,16 @@ class ObjectsTablePanel(Panel):
'url_params': dict_to_querydict(url_params),
}
def should_render(self, context):
"""
Hide the panel if the user does not have view permission for the panel's model.
"""
request = context.get('request')
if request is None:
return True
return request.user.has_perm(get_permission_for_model(self.model, 'view'))
class TemplatePanel(Panel):
"""

7
netbox/netbox/views/generic/bulk_views.py
View File

@ -93,11 +93,16 @@ class ObjectListView(BaseMultiObjectView, ActionsMixin, TableMixin):
delimiter: The character used to separate columns (a comma is used by default)
"""
exclude_columns = {'pk', 'actions'}
all_columns = [col_name for col_name, _ in table.selected_columns + table.available_columns]
if columns:
all_columns = [col_name for col_name, _ in table.selected_columns + table.available_columns]
exclude_columns.update({
col for col in all_columns if col not in columns
})
# Ensure related objects are prefetched for every column that will be exported, not just
# those currently visible in the configured table view.
table._apply_prefetching(columns=[c for c in all_columns if c not in exclude_columns])
exporter = TableExport(
export_format=TableExport.CSV,
table=table,

2
netbox/project-static/dist/netbox.css vendored
View File

File diff suppressed because one or more lines are too long

12
netbox/project-static/styles/overrides/_tabler.scss
View File

@ -173,6 +173,18 @@ pre code {
height: auto;
}
// Center the accordion chevron inside Tabler's toggle box.
.accordion-button-toggle {
align-items: center;
justify-content: center;
flex: 0 0 auto;
> i {
display: block;
line-height: 1;
}
}
// Theme-based visibility utilities
:root:not(.dummy)[data-bs-theme='light'] .hide-theme-light,
:root:not(.dummy)[data-bs-theme='dark'] .hide-theme-dark {

128
netbox/templates/core/htmx/system_db_schema.html Normal file
View File

@ -0,0 +1,128 @@
{% load i18n %}
{% load humanize %}
{% if db_schema %}
{# Summary boxes #}
<div class="row mb-3">
<div class="col-md-4">
<div class="card text-center">
<div class="card-body">
<div class="display-6 fw-bold">{{ db_schema_stats.total_tables|intcomma }}</div>
<div class="text-muted">{% trans "Tables" %}</div>
</div>
</div>
</div>
<div class="col-md-4">
<div class="card text-center">
<div class="card-body">
<div class="display-6 fw-bold">{{ db_schema_stats.total_columns|intcomma }}</div>
<div class="text-muted">{% trans "Columns" %}</div>
</div>
</div>
</div>
<div class="col-md-4">
<div class="card text-center">
<div class="card-body">
<div class="display-6 fw-bold">{{ db_schema_stats.total_indexes|intcomma }}</div>
<div class="text-muted">{% trans "Indexes" %}</div>
</div>
</div>
</div>
</div>
{# Tables grouped by app prefix #}
{% for group in db_schema_groups %}
<div class="card mb-3">
<h2 class="card-header">
<button class="accordion-button collapsed p-0 w-100" type="button"
data-bs-toggle="collapse" data-bs-target="#db-group-body-{{ group.name }}"
aria-expanded="false" aria-controls="db-group-body-{{ group.name }}">
{{ group.name }}
{% if group.is_plugin %}<span class="badge text-bg-purple ms-1">{% trans "plugin" %}</span>{% endif %}
<span class="badge bg-secondary text-bg-gray ms-1">{{ group.tables|length }} {% trans "tables" %}</span>
<span class="badge bg-secondary text-bg-gray ms-1">{{ group.index_count }} {% trans "indexes" %}</span>
<span class="accordion-button-toggle"><i class="mdi mdi-chevron-down"></i></span>
</button>
</h2>
<div id="db-group-body-{{ group.name }}" class="collapse">
<div class="accordion accordion-flush" id="db-group-{{ group.name }}">
{% for table in group.tables %}
<div class="accordion-item">
<h5 class="accordion-header" id="table-heading-{{ group.name }}-{{ forloop.counter }}">
<button class="accordion-button border-bottom collapsed font-monospace" type="button"
data-bs-toggle="collapse" data-bs-target="#table-collapse-{{ group.name }}-{{ forloop.counter }}"
aria-expanded="false" aria-controls="table-collapse-{{ group.name }}-{{ forloop.counter }}">
{{ table.name }}
<span class="badge bg-secondary text-white ms-2">{{ table.columns|length }} {% trans "columns" %}</span>
{% if table.indexes %}
<span class="badge bg-secondary text-white ms-1">{{ table.indexes|length }} {% trans "indexes" %}</span>
{% endif %}
<span class="accordion-button-toggle"><i class="mdi mdi-chevron-down"></i></span>
</button>
</h5>
<div id="table-collapse-{{ group.name }}-{{ forloop.counter }}" class="accordion-collapse collapse"
aria-labelledby="table-heading-{{ group.name }}-{{ forloop.counter }}">
<div class="accordion-body p-0">
<div class="px-3 py-2">
<strong>{% trans "Columns" %}</strong>
<table class="table table-hover table-sm mb-0 mt-1">
<thead>
<tr>
<th>{% trans "Column" %}</th>
<th>{% trans "Type" %}</th>
<th>{% trans "Nullable" %}</th>
<th>{% trans "Default" %}</th>
</tr>
</thead>
<tbody>
{% for column in table.columns %}
<tr>
<td class="font-monospace">{{ column.name }}</td>
<td class="font-monospace text-muted">{{ column.type }}</td>
<td>
{% if column.nullable %}
<span class="text-success">{% trans "yes" %}</span>
{% else %}
<span class="text-muted">{% trans "no" %}</span>
{% endif %}
</td>
<td class="font-monospace text-muted">{{ column.default|default:"" }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% if table.indexes %}
<div class="px-3 py-2 border-top">
<strong>{% trans "Indexes" %}</strong>
<table class="table table-hover table-sm mb-0 mt-1">
<thead>
<tr>
<th>{% trans "Name" %}</th>
<th>{% trans "Definition" %}</th>
</tr>
</thead>
<tbody>
{% for index in table.indexes %}
<tr>
<td class="font-monospace">{{ index.name }}</td>
<td class="font-monospace text-muted">{{ index.definition }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% endif %}
</div>
</div>
</div>
{% endfor %}
</div>
</div>
</div>
{% endfor %}
{% else %}
<div class="card mb-3">
<div class="card-body text-muted">
{% trans "Schema information unavailable." %}
</div>
</div>
{% endif %}

16
netbox/templates/core/system.html
View File

@ -3,6 +3,7 @@
{% load helpers %}
{% load i18n %}
{% load render_table from django_tables2 %}
{% load humanize %}
{% block title %}{% trans "System" %}{% endblock %}
@ -34,6 +35,11 @@
{% trans "Object Counts" %}
</a>
</li>
<li class="nav-item" role="presentation">
<a class="nav-link" id="database-tab" data-bs-toggle="tab" data-bs-target="#database-panel" type="button" role="tab">
{% trans "Database" %}
</a>
</li>
</ul>
{% endblock tabs %}
@ -173,4 +179,14 @@
</div>
</div>
</div>
{# Database panel #}
<div class="tab-pane" id="database-panel" role="tabpanel" aria-labelledby="database-tab"
hx-get="{% url 'core:system_db_schema' %}"
hx-trigger="show.bs.tab from:#database-tab once"
hx-swap="innerHTML">
<div class="text-center text-muted py-5">
<div class="spinner-border spinner-border-sm me-2" role="status"></div>
{% trans "Loading database schema…" %}
</div>
</div>
{% endblock content %}

4
netbox/templates/dcim/panels/module_type.html
View File

@ -11,6 +11,10 @@
<th scope="row">{% trans "Model" %}</th>
<td>{{ object.module_type|linkify }}</td>
</tr>
<tr>
<th scope="row">{% trans "Profile" %}</th>
<td>{{ object.module_type.profile|linkify|placeholder }}</td>
</tr>
{% for k, v in object.module_type.attributes.items %}
<tr>
<th scope="row">{{ k }}</th>

26
netbox/templates/extras/object_render_config.html
View File

@ -49,13 +49,24 @@
</div>
<div class="row">
<div class="col">
{% if config_template %}
{% if error_message %}
<div class="alert alert-warning">
<h4 class="alert-title mb-1">{% trans "Error rendering template" %}</h4>
{% if config_template.debug %}
<div class="overflow-auto" style="max-height: 30rem;">
<pre class="mb-0">{{ error_message }}</pre>
</div>
{% else %}
<pre class="mb-0 text-warning-emphasis bg-transparent border-0 p-0">{{ error_message }}</pre>
{% endif %}
</div>
{% elif config_template %}
{% if rendered_config %}
<div class="card">
<h2 class="card-header d-flex justify-content-between">
{% trans "Rendered Config" %}
<div class="card-actions">
<a href="?export=True" class="btn btn-sm btn-ghost-primary" role="button">
<a href="?export=True{% if request.GET.config_template_id %}&config_template_id={{ request.GET.config_template_id }}{% endif %}" class="btn btn-sm btn-ghost-primary" role="button">
<i class="mdi mdi-download" aria-hidden="true"></i> {% trans "Download" %}
</a>
{% copy_content "rendered_config" %}
@ -63,17 +74,6 @@
</h2>
<pre class="card-body" id="rendered_config">{{ rendered_config }}</pre>
</div>
{% elif error_message %}
<div class="alert alert-warning">
<h4 class="alert-title mb-1">{% trans "Error rendering template" %}</h4>
{% if config_template.debug %}
<div class="overflow-auto" style="max-height: 30rem;">
<pre class="mb-0">{{ error_message }}</pre>
</div>
{% else %}
<pre class="mb-0 text-warning-emphasis bg-transparent border-0 p-0">{{ error_message }}</pre>
{% endif %}
</div>
{% else %}
<div class="alert alert-warning">
<h4 class="alert-title mb-1">{% trans "Template output is empty" %}</h4>

8
netbox/tenancy/api/views.py
View File

@ -42,13 +42,7 @@ class TenantViewSet(NetBoxModelViewSet):
#
class ContactGroupViewSet(MPTTLockedMixin, NetBoxModelViewSet):
queryset = ContactGroup.objects.add_related_count(
ContactGroup.objects.all(),
Contact,
'groups',
'contact_count',
cumulative=True
)
queryset = ContactGroup.objects.annotate_contacts()
serializer_class = serializers.ContactGroupSerializer
filterset_class = filtersets.ContactGroupFilterSet

26
netbox/tenancy/models/contacts.py
View File

@ -1,12 +1,14 @@
from django.contrib.contenttypes.fields import GenericForeignKey
from django.core.exceptions import ValidationError
from django.db import models
from django.db.models.expressions import RawSQL
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from netbox.models import ChangeLoggedModel, NestedGroupModel, OrganizationalModel, PrimaryModel
from netbox.models.features import CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, has_feature
from tenancy.choices import *
from utilities.mptt import TreeManager
__all__ = (
'Contact',
@ -16,10 +18,34 @@ __all__ = (
)
class ContactGroupManager(TreeManager):
def annotate_contacts(self):
"""
Annotate the total number of Contacts belonging to each ContactGroup.
This returns both direct children and children of child groups. Raw SQL is used here to avoid double-counting
contacts which are assigned to multiple child groups of the parent.
"""
return self.annotate(
contact_count=RawSQL(
"SELECT COUNT(DISTINCT m2m.contact_id)"
" FROM tenancy_contact_groups m2m"
" INNER JOIN tenancy_contactgroup cg ON m2m.contactgroup_id = cg.id"
" WHERE cg.tree_id = tenancy_contactgroup.tree_id"
" AND cg.lft >= tenancy_contactgroup.lft"
" AND cg.lft <= tenancy_contactgroup.rght",
()
)
)
class ContactGroup(NestedGroupModel):
"""
An arbitrary collection of Contacts.
"""
objects = ContactGroupManager()
class Meta:
ordering = ['name']
# Empty tuple triggers Django migration detection for MPTT indexes

72
netbox/tenancy/tests/test_models.py Normal file
View File

@ -0,0 +1,72 @@
from django.test import TestCase
from tenancy.models import Contact, ContactGroup
class ContactGroupTestCase(TestCase):
@classmethod
def setUpTestData(cls):
# Create a tree of contact groups:
# - Group A
# - Group A1
# - Group A2
# - Group B
cls.group_a = ContactGroup.objects.create(name='Group A', slug='group-a')
cls.group_a1 = ContactGroup.objects.create(name='Group A1', slug='group-a1', parent=cls.group_a)
cls.group_a2 = ContactGroup.objects.create(name='Group A2', slug='group-a2', parent=cls.group_a)
cls.group_b = ContactGroup.objects.create(name='Group B', slug='group-b')
# Create contacts
cls.contact1 = Contact.objects.create(name='Contact 1')
cls.contact2 = Contact.objects.create(name='Contact 2')
cls.contact3 = Contact.objects.create(name='Contact 3')
cls.contact4 = Contact.objects.create(name='Contact 4')
def test_annotate_contacts_direct(self):
"""Contacts assigned directly to a group should be counted."""
self.contact1.groups.set([self.group_a])
self.contact2.groups.set([self.group_a])
queryset = ContactGroup.objects.annotate_contacts()
self.assertEqual(queryset.get(pk=self.group_a.pk).contact_count, 2)
def test_annotate_contacts_cumulative(self):
"""Contacts assigned to child groups should be included in the parent's count."""
self.contact1.groups.set([self.group_a1])
self.contact2.groups.set([self.group_a2])
queryset = ContactGroup.objects.annotate_contacts()
self.assertEqual(queryset.get(pk=self.group_a.pk).contact_count, 2)
self.assertEqual(queryset.get(pk=self.group_a1.pk).contact_count, 1)
self.assertEqual(queryset.get(pk=self.group_a2.pk).contact_count, 1)
def test_annotate_contacts_no_double_counting(self):
"""A contact assigned to multiple child groups must be counted only once for the parent."""
self.contact1.groups.set([self.group_a1, self.group_a2])
queryset = ContactGroup.objects.annotate_contacts()
self.assertEqual(queryset.get(pk=self.group_a.pk).contact_count, 1)
def test_annotate_contacts_mixed(self):
"""Test a mix of direct and inherited contacts with overlap."""
self.contact1.groups.set([self.group_a])
self.contact2.groups.set([self.group_a1])
self.contact3.groups.set([self.group_a1, self.group_a2])
self.contact4.groups.set([self.group_b])
queryset = ContactGroup.objects.annotate_contacts()
# Group A: contact1 (direct) + contact2 (via A1) + contact3 (via A1 & A2) = 3
self.assertEqual(queryset.get(pk=self.group_a.pk).contact_count, 3)
# Group A1: contact2 + contact3 = 2
self.assertEqual(queryset.get(pk=self.group_a1.pk).contact_count, 2)
# Group A2: contact3 = 1
self.assertEqual(queryset.get(pk=self.group_a2.pk).contact_count, 1)
# Group B: contact4 = 1
self.assertEqual(queryset.get(pk=self.group_b.pk).contact_count, 1)
def test_annotate_contacts_empty(self):
"""Groups with no contacts should return a count of zero."""
queryset = ContactGroup.objects.annotate_contacts()
self.assertEqual(queryset.get(pk=self.group_a.pk).contact_count, 0)
self.assertEqual(queryset.get(pk=self.group_b.pk).contact_count, 0)

26
netbox/tenancy/views.py
View File

@ -206,13 +206,7 @@ class TenantBulkDeleteView(generic.BulkDeleteView):
@register_model_view(ContactGroup, 'list', path='', detail=False)
class ContactGroupListView(generic.ObjectListView):
queryset = ContactGroup.objects.add_related_count(
ContactGroup.objects.all(),
Contact,
'groups',
'contact_count',
cumulative=True
)
queryset = ContactGroup.objects.annotate_contacts()
filterset = filtersets.ContactGroupFilterSet
filterset_form = forms.ContactGroupFilterForm
table = tables.ContactGroupTable
@ -256,7 +250,7 @@ class ContactGroupView(GetRelatedModelsMixin, generic.ObjectView):
request,
groups,
extra=(
(Contact.objects.restrict(request.user, 'view').filter(groups__in=groups), 'group_id'),
(Contact.objects.restrict(request.user, 'view').filter(groups__in=groups).distinct(), 'group_id'),
),
),
}
@ -282,13 +276,7 @@ class ContactGroupBulkImportView(generic.BulkImportView):
@register_model_view(ContactGroup, 'bulk_edit', path='edit', detail=False)
class ContactGroupBulkEditView(generic.BulkEditView):
queryset = ContactGroup.objects.add_related_count(
ContactGroup.objects.all(),
Contact,
'groups',
'contact_count',
cumulative=True
)
queryset = ContactGroup.objects.annotate_contacts()
filterset = filtersets.ContactGroupFilterSet
table = tables.ContactGroupTable
form = forms.ContactGroupBulkEditForm
@ -302,13 +290,7 @@ class ContactGroupBulkRenameView(generic.BulkRenameView):
@register_model_view(ContactGroup, 'bulk_delete', path='delete', detail=False)
class ContactGroupBulkDeleteView(generic.BulkDeleteView):
queryset = ContactGroup.objects.add_related_count(
ContactGroup.objects.all(),
Contact,
'groups',
'contact_count',
cumulative=True
)
queryset = ContactGroup.objects.annotate_contacts()
filterset = filtersets.ContactGroupFilterSet
table = tables.ContactGroupTable

931
netbox/translations/en/LC_MESSAGES/django.po
View File

File diff suppressed because it is too large Load Diff

2
netbox/utilities/security.py
View File

@ -9,7 +9,7 @@ def validate_peppers(peppers):
"""
Validate the given dictionary of cryptographic peppers for type & sufficient length.
"""
if type(peppers) is not dict:
if not isinstance(peppers, dict):
raise ImproperlyConfigured("API_TOKEN_PEPPERS must be a dictionary.")
for key, pepper in peppers.items():
if type(key) is not int:

22
netbox/utilities/testing/api.py
View File

@ -258,7 +258,8 @@ class APIViewTestCases:
changed_object_id=instance.pk,
action=ObjectChangeActionChoices.ACTION_CREATE,
)
self.assertEqual(objectchange.message, data['changelog_message'])
self.assertObjectChange(objectchange, action=ObjectChangeActionChoices.ACTION_CREATE,
message=data['changelog_message'])
def test_bulk_create_objects(self):
"""
@ -311,7 +312,8 @@ class APIViewTestCases:
)
self.assertEqual(len(objectchanges), len(self.create_data))
for oc in objectchanges:
self.assertEqual(oc.message, changelog_message)
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_CREATE,
message=changelog_message)
class UpdateObjectViewTestCase(APITestCase):
update_data = {}
@ -369,8 +371,8 @@ class APIViewTestCases:
changed_object_type=ContentType.objects.get_for_model(instance),
changed_object_id=instance.pk
)
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_UPDATE)
self.assertEqual(objectchange.message, data['changelog_message'])
self.assertObjectChange(objectchange, action=ObjectChangeActionChoices.ACTION_UPDATE,
message=data['changelog_message'])
def test_update_object_with_etag(self):
"""
@ -459,8 +461,8 @@ class APIViewTestCases:
)
self.assertEqual(len(objectchanges), len(data))
for oc in objectchanges:
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_UPDATE)
self.assertEqual(oc.message, changelog_message)
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_UPDATE,
message=changelog_message)
class DeleteObjectViewTestCase(APITestCase):
@ -507,8 +509,8 @@ class APIViewTestCases:
changed_object_type=ContentType.objects.get_for_model(instance),
changed_object_id=instance.pk
)
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_DELETE)
self.assertEqual(objectchange.message, data['changelog_message'])
self.assertObjectChange(objectchange, action=ObjectChangeActionChoices.ACTION_DELETE,
message=data['changelog_message'])
def test_bulk_delete_objects(self):
"""
@ -548,8 +550,8 @@ class APIViewTestCases:
)
self.assertEqual(len(objectchanges), len(data))
for oc in objectchanges:
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_DELETE)
self.assertEqual(oc.message, changelog_message)
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_DELETE,
message=changelog_message)
class GraphQLTestCase(APITestCase):

25
netbox/utilities/testing/base.py
View File

@ -13,6 +13,7 @@ from django.test import TestCase as _TestCase
from netaddr import IPNetwork
from taggit.managers import TaggableManager
from core.choices import ObjectChangeActionChoices
from core.models import ObjectType
from users.models import ObjectPermission, User
from utilities.data import ranges_to_string
@ -83,6 +84,30 @@ class TestCase(_TestCase):
# Custom assertions
#
def assertObjectChange(self, objectchange, *, action, message=None):
"""
Assert that an ObjectChange record has the expected attributes. If message is provided, it will be
compared against objectchange.message.
"""
# Verify the change action (create, update, delete)
self.assertEqual(objectchange.action, action)
# Verify the changelog message if provided
if message is not None:
self.assertEqual(objectchange.message, message)
# Verify pre/postchange data presence and integrity based on action type
if action == ObjectChangeActionChoices.ACTION_CREATE:
self.assertIsNone(objectchange.prechange_data, "Expected prechange_data to be None for a create")
self.assertIsNotNone(objectchange.postchange_data, "Expected postchange_data to be populated for a create")
elif action == ObjectChangeActionChoices.ACTION_UPDATE:
self.assertIsNotNone(objectchange.prechange_data, "Expected prechange_data to be populated for an update")
self.assertIsNotNone(objectchange.postchange_data, "Expected postchange_data to be populated for an update")
self.assertNotEqual(objectchange.prechange_data, objectchange.postchange_data)
elif action == ObjectChangeActionChoices.ACTION_DELETE:
self.assertIsNotNone(objectchange.prechange_data, "Expected prechange_data to be populated for a delete")
self.assertIsNone(objectchange.postchange_data, "Expected postchange_data to be None for a delete")
def assertHttpStatus(self, response, expected_status):
"""
TestCase method. Provide more detail in the event of an unexpected HTTP response.

23
netbox/utilities/testing/views.py
View File

@ -192,8 +192,8 @@ class ViewTestCases:
changed_object_id=instance.pk
)
self.assertEqual(len(objectchanges), 1)
self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_CREATE)
self.assertEqual(objectchanges[0].message, self.form_data['changelog_message'])
self.assertObjectChange(objectchanges[0], action=ObjectChangeActionChoices.ACTION_CREATE,
message=self.form_data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_create_object_with_constrained_permission(self):
@ -299,8 +299,8 @@ class ViewTestCases:
changed_object_id=instance.pk
)
self.assertEqual(len(objectchanges), 1)
self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_UPDATE)
self.assertEqual(objectchanges[0].message, self.form_data['changelog_message'])
self.assertObjectChange(objectchanges[0], action=ObjectChangeActionChoices.ACTION_UPDATE,
message=self.form_data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_edit_object_with_constrained_permission(self):
@ -394,8 +394,8 @@ class ViewTestCases:
changed_object_id=instance.pk
)
self.assertEqual(len(objectchanges), 1)
self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_DELETE)
self.assertEqual(objectchanges[0].message, form_data['changelog_message'])
self.assertObjectChange(objectchanges[0], action=ObjectChangeActionChoices.ACTION_DELETE,
message=form_data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_delete_object_with_constrained_permission(self):
@ -716,7 +716,8 @@ class ViewTestCases:
self.assertEqual(len(objectchanges), expected_new_objects)
for oc in objectchanges:
self.assertEqual(oc.message, data['changelog_message'])
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_CREATE,
message=data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_bulk_update_objects_with_permission(self):
@ -868,8 +869,8 @@ class ViewTestCases:
)
self.assertEqual(len(objectchanges), len(pk_list))
for oc in objectchanges:
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_UPDATE)
self.assertEqual(oc.message, data['changelog_message'])
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_UPDATE,
message=data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_bulk_edit_objects_with_constrained_permission(self):
@ -964,8 +965,8 @@ class ViewTestCases:
)
self.assertEqual(len(objectchanges), len(pk_list))
for oc in objectchanges:
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_DELETE)
self.assertEqual(oc.message, data['changelog_message'])
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_DELETE,
message=data['changelog_message'])
def test_bulk_delete_objects_with_constrained_permission(self):
pk_list = self._get_queryset().values_list('pk', flat=True)

52
netbox/utilities/tests/test_api.py
View File

@ -474,3 +474,55 @@ class GetPrefetchesForSerializerTestCase(TestCase):
get_prefetches_for_serializer(SiteSerializer),
['region', 'region__parent'],
)
class APITrailingSlashTestCase(APITestCase):
"""
Verify behavior for REST API requests sent to a URL without a trailing slash.
GET requests should continue to be redirected to the trailing-slash URL (Django's default
APPEND_SLASH behavior). Write methods (POST/PUT/PATCH/DELETE) should instead receive a 404
so that the request body is not silently dropped by a redirect.
"""
model = Site
user_permissions = ('dcim.view_site', 'dcim.add_site', 'dcim.change_site', 'dcim.delete_site')
@classmethod
def setUpTestData(cls):
cls.site = Site.objects.create(name='Site 1', slug='site-1')
def _strip_slash(self, url):
return url.rstrip('/')
def test_get_redirects(self):
url = self._strip_slash(reverse('dcim-api:site-list'))
response = self.client.get(url, **self.header)
self.assertIn(response.status_code, (301, 302))
self.assertTrue(response['Location'].endswith('/'))
def test_post_returns_404(self):
url = self._strip_slash(reverse('dcim-api:site-list'))
data = {'name': 'Site 2', 'slug': 'site-2'}
with disable_warnings('django.request'):
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
def test_patch_returns_404(self):
url = self._strip_slash(self._get_detail_url(self.site))
with disable_warnings('django.request'):
response = self.client.patch(url, {'name': 'Renamed'}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
def test_put_returns_404(self):
url = self._strip_slash(self._get_detail_url(self.site))
data = {'name': 'Renamed', 'slug': 'renamed'}
with disable_warnings('django.request'):
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
def test_delete_returns_404(self):
url = self._strip_slash(self._get_detail_url(self.site))
with disable_warnings('django.request'):
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
self.assertTrue(Site.objects.filter(pk=self.site.pk).exists())

38
netbox/virtualization/tests/test_api.py
View File

@ -557,6 +557,44 @@ class VirtualMachineTest(APIViewTestCases.APIViewTestCase):
[str(nat_ip.address)],
)
def test_render_config_with_config_template_id(self):
default_template = ConfigTemplate.objects.create(
name='Default Template',
template_code='Default config for {{ virtualmachine.name }}'
)
override_template = ConfigTemplate.objects.create(
name='Override Template',
template_code='Override config for {{ virtualmachine.name }}'
)
vm = VirtualMachine.objects.first()
vm.config_template = default_template
vm.save()
self.add_permissions(
'virtualization.render_config_virtualmachine', 'virtualization.view_virtualmachine',
'extras.view_configtemplate'
)
url = reverse('virtualization-api:virtualmachine-render-config', kwargs={'pk': vm.pk})
# Render with override template
response = self.client.post(url, {'config_template_id': override_template.pk}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(response.data['content'], f'Override config for {vm.name}')
# Render with nonexistent config_template_id
response = self.client.post(url, {'config_template_id': 999999}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
# Render with non-integer config_template_id
response = self.client.post(url, {'config_template_id': 'abc'}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
# Without view_configtemplate permission, override template should not be accessible
self.remove_permissions('extras.view_configtemplate')
response = self.client.post(url, {'config_template_id': override_template.pk}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
class VMInterfaceTest(APIViewTestCases.APIViewTestCase):
model = VMInterface

40
netbox/virtualization/tests/test_views.py
View File

@ -484,6 +484,46 @@ class VirtualMachineTestCase(ViewTestCases.PrimaryObjectViewTestCase):
self.remove_permissions('virtualization.view_virtualmachine')
self.assertHttpStatus(self.client.get(url), 403)
def test_virtualmachine_renderconfig_with_config_template_id(self):
default_template = ConfigTemplate.objects.create(
name='Default Template',
template_code='Default config for {{ virtualmachine.name }}'
)
override_template = ConfigTemplate.objects.create(
name='Override Template',
template_code='Override config for {{ virtualmachine.name }}'
)
vm = VirtualMachine.objects.first()
vm.config_template = default_template
vm.save()
self.add_permissions(
'virtualization.view_virtualmachine', 'virtualization.render_config_virtualmachine',
'extras.view_configtemplate'
)
url = reverse('virtualization:virtualmachine_render-config', kwargs={'pk': vm.pk})
# Render with override config_template_id
response = self.client.get(url, {'config_template_id': override_template.pk})
self.assertHttpStatus(response, 200)
self.assertIn(b'Override config for', response.content)
# Render with nonexistent config_template_id still returns 200 with error message
response = self.client.get(url, {'config_template_id': 999999})
self.assertHttpStatus(response, 200)
self.assertIn(b'Error rendering template', response.content)
# Render with non-integer config_template_id still returns 200 with error message
response = self.client.get(url, {'config_template_id': 'abc'})
self.assertHttpStatus(response, 200)
self.assertIn(b'Error rendering template', response.content)
# Without view_configtemplate permission, override template should not be accessible
self.remove_permissions('extras.view_configtemplate')
response = self.client.get(url, {'config_template_id': override_template.pk})
self.assertHttpStatus(response, 200)
self.assertIn(b'Error rendering template', response.content)
class VMInterfaceTestCase(ViewTestCases.DeviceComponentViewTestCase):
model = VMInterface