Code development platform for open source projects from the European Union institutions

Skip to content
Snippets Groups Projects
Commit baab45fb authored by Laurent VENIER's avatar Laurent VENIER :speech_balloon:
Browse files

:twisted_rightwards_arrows: Merge branch 'new_rps_mapping_properties' into 'main'

:sparkles: Add new Mapping properties

See merge request !43
parents b5578f7b 5fc9979f
No related branches found
No related tags found
1 merge request!43✨ Add new Mapping properties
Pipeline #98241 passed
Showing
with 595 additions and 12 deletions
......@@ -10,7 +10,7 @@ class NetBoxRpsConfig(PluginConfig):
name = "netbox_rps_plugin"
verbose_name = "NetBox RPS"
description = "A Netbox plugin to add RPS resources"
version = "0.10.1"
version = "0.11.0"
author = "Vincent Simonin"
author_email = "vincent.simonin@ext.ec.europa.eu"
base_url = "rps"
......
......@@ -117,6 +117,12 @@ class MappingSerializer(NetBoxModelSerializer):
"testingpage",
"webdav",
"Comment",
"gzip_proxied",
"keepalive_requests",
"keepalive_timeout",
"proxy_cache",
"proxy_read_timeout",
"client_max_body_size",
"custom_fields",
"created",
"last_updated",
......
......@@ -18,6 +18,12 @@ class MappingFilterSet(NetBoxModelFilterSet):
"Comment",
"webdav",
"testingpage",
"gzip_proxied",
"keepalive_requests",
"keepalive_timeout",
"proxy_cache",
"proxy_read_timeout",
"client_max_body_size",
)
# pylint: disable=W0613
......
......@@ -29,13 +29,28 @@ class MappingForm(NetBoxModelForm):
"authentication",
"webdav",
"testingpage",
"gzip_proxied",
"keepalive_requests",
"keepalive_timeout",
"proxy_cache",
"proxy_read_timeout",
"client_max_body_size",
"Comment",
"tags",
)
help_texts = {"target": "URL-friendly unique shorthand"}
help_texts = {
"target": "URL-friendly unique shorthand",
"keepalive_requests": "Min value 100 requests, max 5000 requests, default 1000 requests.",
"keepalive_timeout": "In seconds. Min value 1, max 300, default 75.",
"proxy_read_timeout": "In seconds. Min value 1, max 300, default 60.",
"client_max_body_size": "In Mega Bytes. Min 1, max 255, default 1."
}
labels = {
"source": "Source",
"target": "Target",
"keepalive_timeout": "Keepalive timeout (s)",
"proxy_read_timeout": " Proxy read timeout (s)",
"client_max_body_size": "Client max body size (MB)"
}
......@@ -62,6 +77,16 @@ class MappingFilterForm(NetBoxModelFilterSetForm):
Comment = forms.CharField(
max_length=500, min_length=1, required=False, label="Comment"
)
gzip_proxied = webdav = forms.BooleanField(
required=False, widget=forms.Select(choices=BOOLEAN_WITH_BLANK_CHOICES)
)
keepalive_requests = forms.IntegerField(min_value=100, max_value=5000, required=False)
keepalive_timeout = forms.IntegerField(min_value=1, max_value=300, required=False)
proxy_cache = forms.BooleanField(
required=False, widget=forms.Select(choices=BOOLEAN_WITH_BLANK_CHOICES)
)
proxy_read_timeout = forms.IntegerField(min_value=1, max_value=300, required=False)
client_max_body_size = forms.IntegerField(min_value=1, max_value=255, required=False)
tag = TagFilterField(model)
......
"""Migration File"""
# pylint: disable=C0103
from django.db import migrations, models
class Migration(migrations.Migration):
"""Migration Class"""
dependencies = [("netbox_rps_plugin", "0007_saml_config")]
operations = [
migrations.AddField(
model_name="mapping",
name="gzip_proxied",
field=models.BooleanField(default=False)
),
migrations.AddField(
model_name="mapping",
name="keepalive_requests",
field=models.IntegerField(default=1000)
),
migrations.AddField(
model_name="mapping",
name="keepalive_timeout",
field=models.IntegerField(default=75)
),
migrations.AddField(
model_name="mapping",
name="proxy_cache",
field=models.BooleanField(default=False)
),
migrations.AddField(
model_name="mapping",
name="proxy_read_timeout",
field=models.IntegerField(default=60)
),
migrations.AddField(
model_name="mapping",
name="client_max_body_size",
field=models.IntegerField(default=1)
),
]
......@@ -2,7 +2,7 @@
from django.db import models
from django.urls import reverse
from django.core.validators import URLValidator
from django.core.validators import URLValidator, MaxValueValidator, MinValueValidator
from netbox.models import NetBoxModel
from utilities.choices import ChoiceSet
......@@ -69,6 +69,20 @@ class Mapping(NetBoxModel):
default=False,
)
Comment = models.CharField(max_length=500, blank=True)
gzip_proxied = models.BooleanField(default=False)
keepalive_requests = models.IntegerField(
default=1000, validators=[MinValueValidator(100), MaxValueValidator(5000)]
)
keepalive_timeout = models.IntegerField(
default=75, validators=[MinValueValidator(1), MaxValueValidator(300)]
)
proxy_cache = models.BooleanField(default=False)
proxy_read_timeout = models.IntegerField(
default=60, validators=[MinValueValidator(1), MaxValueValidator(300)]
)
client_max_body_size = models.IntegerField(
default=1, validators=[MinValueValidator(1), MaxValueValidator(255)]
)
class Meta:
ordering = ("source", "target")
......
......@@ -28,6 +28,12 @@ class MappingTable(NetBoxTable):
"testingpage",
"webdav",
"Comment",
"gzip_proxied",
"keepalive_requests",
"keepalive_timeout",
"proxy_cache",
"proxy_read_timeout",
"client_max_body_size",
"httpheader_count",
"tags",
"created",
......@@ -38,6 +44,12 @@ class MappingTable(NetBoxTable):
"target",
"authentication",
"webdav",
"gzip_proxied",
"keepalive_requests",
"keepalive_timeout",
"proxy_cache",
"proxy_read_timeout",
"client_max_body_size",
"httpheader_count",
)
......
......@@ -64,7 +64,7 @@
</tr>
<tr>
<th scope="row">Webdav</th>
<td>{{ object.webdav }}</td>
<td>{{ object.webdav|yesno }}</td>
</tr>
<tr>
<th scope="row">Comment</th>
......@@ -76,6 +76,30 @@
{% endif %}
</td>
</tr>
<tr>
<th scope="row">gzip proxied</th>
<td>{{ object.gzip_proxied|yesno }}</td>
</tr>
<tr>
<th scope="row">Keepalive requests</th>
<td>{{ object.keepalive_requests }}</td>
</tr>
<tr>
<th scope="row">Keepalive timeout</th>
<td>{{ object.keepalive_timeout }}&nbsp;s</td>
</tr>
<tr>
<th scope="row">Proxy cache</th>
<td>{{ object.proxy_cache|yesno }}</td>
</tr>
<tr>
<th scope="row">Proxy read timeout</th>
<td>{{ object.proxy_read_timeout }}&nbsp;s</td>
</tr>
<tr>
<th scope="row">Client max body size</th>
<td>{{ object.client_max_body_size }}&nbsp;MB</td>
</tr>
</table>
</div>
</div>
......
......@@ -2,7 +2,7 @@ from setuptools import find_packages, setup
setup(
name='netbox_rps_plugin',
version='0.10.1',
version='0.11.0',
description='A Netbox plugin to add RPS resources',
install_requires=[],
packages=find_packages(),
......
......@@ -17,9 +17,10 @@ class Base(unittest.TestCase):
def tearDown(self) -> None:
"""Teardown function"""
requests.delete(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json=[{"id": self.mapping_id}],
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
if self.mapping_id is not None:
requests.delete(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json=[{"id": self.mapping_id}],
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
"""Test case for Mapping creation"""
import unittest
import json
import os
import requests
from .base import Base
HOST = os.getenv("HOST", default="localhost")
PORT = os.getenv("PORT", default="8080")
API_KEY = os.getenv("API_KEY", "only4testingpurpose")
class TestMappingPropertiesValidation(Base):
"""Test case for Mapping properties validation domain class"""
def test_that_gzip_proxied_is_false_by_default(self) -> None:
"""Test that gzip proxied property is false by default"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["gzip_proxied"], False)
self.mapping_id = json.loads(response.content)["id"]
def test_that_gzip_proxied_is_boolean(self) -> None:
"""Test that gzip proxied property is a boolean"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"gzip_proxied": True,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["gzip_proxied"], True)
self.mapping_id = json.loads(response.content)["id"]
def test_that_proxy_cache_is_false_by_default(self) -> None:
"""Test that proxy cache property is false by default"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["proxy_cache"], False)
self.mapping_id = json.loads(response.content)["id"]
def test_that_proxy_cache_is_boolean(self) -> None:
"""Test that proxy cache property is a boolean"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"proxy_cache": True,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["proxy_cache"], True)
self.mapping_id = json.loads(response.content)["id"]
def test_that_keepalive_requests_have_limit(self) -> None:
"""Test that keepalive request property is between 100 and 5000"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"keepalive_requests": 99,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content,
b'{"keepalive_requests":["Ensure this value is greater than or equal to 100."]}',
)
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"keepalive_requests": 5001,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content,
b'{"keepalive_requests":["Ensure this value is less than or equal to 5000."]}',
)
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"keepalive_requests": 1500,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["keepalive_requests"], 1500)
self.mapping_id = json.loads(response.content)["id"]
def test_that_keepalive_requests_default_value_is_1000(self) -> None:
"""Test that keepalive request property default value is 1000"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["keepalive_requests"], 1000)
self.mapping_id = json.loads(response.content)["id"]
def test_that_keepalive_requests_default_type_is_integer(self) -> None:
"""Test that keepalive request property type is integer"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"keepalive_requests": "toto",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content, b'{"keepalive_requests":["A valid integer is required."]}'
)
def test_that_keepalive_timeout_have_limit(self) -> None:
"""Test that keepalive timeout property is between 1 and 300"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"keepalive_timeout": 0,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content,
b'{"keepalive_timeout":["Ensure this value is greater than or equal to 1."]}',
)
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"keepalive_timeout": 301,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content,
b'{"keepalive_timeout":["Ensure this value is less than or equal to 300."]}',
)
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"keepalive_timeout": 200,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["keepalive_timeout"], 200)
self.mapping_id = json.loads(response.content)["id"]
def test_that_keepalive_timeout_default_value_is_75(self) -> None:
"""Test that keepalive timeout property default value is 75"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["keepalive_timeout"], 75)
self.mapping_id = json.loads(response.content)["id"]
def test_that_keepalive_timeout_default_type_is_integer(self) -> None:
"""Test that keepalive timeout property type is integer"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"keepalive_timeout": "toto",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content, b'{"keepalive_timeout":["A valid integer is required."]}'
)
def test_that_proxy_read_timeout_have_limit(self) -> None:
"""Test that proxy_read_timeout property is between 1 and 300"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"proxy_read_timeout": 0,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content,
b'{"proxy_read_timeout":["Ensure this value is greater than or equal to 1."]}',
)
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"proxy_read_timeout": 301,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content,
b'{"proxy_read_timeout":["Ensure this value is less than or equal to 300."]}',
)
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"proxy_read_timeout": 200,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["proxy_read_timeout"], 200)
self.mapping_id = json.loads(response.content)["id"]
def test_that_proxy_read_timeout_default_value_is_60(self) -> None:
"""Test that proxy_read_timeout property default value is 60"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["proxy_read_timeout"], 60)
self.mapping_id = json.loads(response.content)["id"]
def test_that_proxy_read_timeout_default_type_is_integer(self) -> None:
"""Test that proxy_read_timeout property type is integer"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"proxy_read_timeout": "toto",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content, b'{"proxy_read_timeout":["A valid integer is required."]}'
)
def test_that_client_max_body_size_have_limit(self) -> None:
"""Test that client_max_body_size property is between 1 and 255"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"client_max_body_size": 0,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content,
b'{"client_max_body_size":["Ensure this value is greater than or equal to 1."]}',
)
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"client_max_body_size": 256,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content,
b'{"client_max_body_size":["Ensure this value is less than or equal to 255."]}',
)
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"client_max_body_size": 200,
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["client_max_body_size"], 200)
self.mapping_id = json.loads(response.content)["id"]
def test_that_client_max_body_size_default_value_is_60(self) -> None:
"""Test that client_max_body_size property default value is 1"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.json()["client_max_body_size"], 1)
self.mapping_id = json.loads(response.content)["id"]
def test_that_client_max_body_size_default_type_is_integer(self) -> None:
"""Test that client_max_body_size property type is integer"""
response = requests.post(
url=f"http://{HOST}:{PORT}/api/plugins/rps/mapping/",
json={
"source": "https://truc00.com/api",
"target": "http://10.10.10.10:1800/api",
"proxy_read_timeout": "toto",
},
headers={"Authorization": f"Token {API_KEY}"},
timeout=5,
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.content, b'{"proxy_read_timeout":["A valid integer is required."]}'
)
if __name__ == "__main__":
unittest.main()
......@@ -431,7 +431,7 @@ ignore-imports = true
ignore-signatures = true
# Minimum lines number of a similarity.
min-similarity-lines = 10
min-similarity-lines = 15
[tool.pylint.spelling]
# Limits count of emitted suggestions for spelling mistakes.
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment