Implement strict VLAN validation in sdx-lib to prevent invalid L2VPNs

old_tests/test_config.py

@@ -0,0 +1,182 @@
+from sdxlib.sdx_client import SDXClient
+
+TEST_SERVICE_ID = "8344657b-2466-4735-9a21-143643073865"
+TEST_URL = "http://aw-sdx-controller.renci.org:8081"
+TEST_USERNAME = "testuser"
+TEST_PASSWORD = "testpassword"
+TEST_NAME = "Test_L2VPN"
+TEST_OWNERSHIP = "user@example.com"
+TEST_CREATION_DATE = "20240522T00:00:00Z"
+TEST_ARCHIVED_DATE = "0"
+TEST_STATUS = "up"
+TEST_STATE = "enabled"
+TEST_COUNTERS_LOCATION = "https://my.aw-sdx.net/l2vpn/7cdf23e8978c"
+TEST_LAST_MODIFIED = "0"
+TEST_CURRENT_PATH = [
+    "urn:sdx:link:tenet.ac.za:LinkToSAX",
+    "urn:sdx:link:tenet.ac.za:LinkToAmpath",
+    "urn:sdx:link:ampath.net:LinkToSAX",
+]
+TEST_OXP_SERVICE_IDS = {
+    "AmLight.net": ["c73da8e1"],
+    "TENET.ac.za": ["5d034620"],
+    "SAX.br": ["7cdf23e8978c"],
+}
+TEST_ENDPOINTS = [
+    {
+        "port_id": "urn:sdx:port:test-oxp_url:test-node_name:test-port_name",
+        "vlan": "100",
+    },
+    {
+        "port_id": "urn:sdx:port:test-oxp_url:test-node_name:test-port_name2",
+        "vlan": "200",
+    },
+]
+VLAN_100 = {
+    "port_id": "urn:sdx:port:test-oxp_url:test-node_name:test-port_name",
+    "vlan": "100",
+}
+VLAN_200 = {
+    "port_id": "urn:sdx:port:test-oxp_url:test-node_name:test-port_name2",
+    "vlan": "200",
+}
+VLAN_ANY = {
+    "port_id": "urn:sdx:port:test-oxp_url:test-node_name:test-port_name",
+    "vlan": "any",
+}
+VLAN_ALL = {
+    "port_id": "urn:sdx:port:test-ox_url:test-node_name:test-port_name",
+    "vlan": "all",
+}
+VLAN_RANGE = {
+    "port_id": "urn:sdx:port:test-oxp_url:test-node_name:test-port_name",
+    "vlan": "100:200",
+}
+VLAN_UNTAGGED = {
+    "port_id": "urn:sdx:port:test-oxp_url:test-node_name:test-port_name2",
+    "vlan": "untagged",
+}
+
+TEST_VALID_RESPONSE = {
+    "service_id": TEST_SERVICE_ID,
+    "name": TEST_NAME,
+    "endpoints": TEST_ENDPOINTS,
+    "ownership": TEST_OWNERSHIP,
+    "creation_date": TEST_CREATION_DATE,
+    "archived_date": TEST_ARCHIVED_DATE,
+    "status": TEST_STATUS,
+    "state": TEST_STATE,
+    "counters_location": TEST_COUNTERS_LOCATION,
+    "last_modified": TEST_LAST_MODIFIED,
+    "current_path": TEST_CURRENT_PATH,
+    "oxp_service_ids": TEST_OXP_SERVICE_IDS,
+}
+
+TEST_MISSING_ATTRIBUTES_RESPONSE = {
+    "service_id": TEST_SERVICE_ID,
+    "name": TEST_NAME,
+    "endpoints": TEST_ENDPOINTS,
+    "ownership": TEST_OWNERSHIP,
+    "creation_date": TEST_CREATION_DATE,
+    "archived_date": TEST_ARCHIVED_DATE,
+    "status": TEST_STATUS,
+    "state": TEST_STATE,
+    "counters_location": TEST_COUNTERS_LOCATION,
+    "last_modified": TEST_LAST_MODIFIED,
+    "current_path": TEST_CURRENT_PATH,
+    "oxp_service_ids": TEST_OXP_SERVICE_IDS,
+    # `scheduling` and `description` are intentionally missing
+}
+
+
+MOCK_RESPONSE = {
+    TEST_SERVICE_ID: {
+        "service_id": TEST_SERVICE_ID,
+        "name": TEST_NAME,
+        "ownership": "user1",
+        "endpoints": TEST_ENDPOINTS,
+        "creation_date": "20240522T00:00:00Z",
+        "archived_date": "0",
+        "status": "up",
+        "state": "enabled",
+        "counters_location": "https://my.aw-sdx.net/l2vpn/7cdf23e8978c",
+        "last_modified": "0",
+        "current_path": ["urn:sdx:link:tenet.ac.za:LinkToAmpath"],
+        "oxp_service_ids": {"ampath.net": ["c73da8e1"], "tenet.ac.za": ["5d034620"],},
+        "qos_metrics": {
+            "max_delay": {"strict": True, "value": 150},
+            "min_bw": {"strict": False, "value": 5},
+        },
+        "scheduling": None,
+        "notifications": [
+            {"email": "user@domain.com"},
+            {"email": "user2@domain2.com"},
+        ],
+    }
+}
+
+# Name error message.
+ERROR_NAME_INVALID = "Name must be a non-empty string with maximum 50 characters."
+
+# Endpoint error messages.
+ERROR_INVALID_ENDPOINTS = "Endpoints must be a list."
+ERROR_EMPTY_ENDPOINTS_LIST = "Endpoints list cannot be empty."
+ERROR_MIN_ENTRIES = "Endpoints must contain at least 2 entries."
+ERROR_LIST_OF_DICTS = "Endpoints must be a list of dictionaries."
+ERROR_NONEMPTY_PORT_ID = "Each endpoint must contain a non-empty 'port_id' key."
+ERROR_INVALID_PORT_ID_FORMAT = "Invalid port_id format: invalid-port_id"
+ERROR_EMPTY_VLAN_VALUE = "Each endpoint must contain a non-empty 'vlan' key."
+ERROR_INVALID_VLAN_TYPE = "VLAN must be a string."
+ERROR_INVALID_VLAN_VALUE = "Invalid VLAN value: '{}'. Must be between 1 and 4095."
+ERROR_VLAN_RANGE_MISMATCH = (
+    "All endpoints must have the same VLAN value if one endpoint is 'all' or a range."
+)
+ERROR_VLAN_INVALID = "Invalid VLAN value: '{}'. Must be 'any', 'all', 'untagged', a string representing an integer between 1 and 4095, or a range."
+ERROR_VLAN_RANGE_VALUE = "Invalid VLAN range format: '{}'. Must be 'VLAN ID1:VLAN ID2'."
+
+# Description error messages
+ERROR_DESCRIPTION_TOO_LONG = "Description attribute must be less than 256 characters."
+
+# Notifications error messages
+ERROR_NOTIFICATIONS_NOT_LIST = "Notifications must be provided as a list."
+ERROR_NOTIFICATION_ITEM_NOT_DICT = "Each notification must be a dictionary."
+ERROR_NOTIFICATION_ITEM_EMAIL_KEY = (
+    "Each notification dictionary must contain a key 'email'."
+)
+ERROR_NOTIFICATION_INVALID_EMAIL_FORMAT = (
+    "Invalid email address or email format: invalid_email"
+)
+ERROR_NOTIFICATION_EXCEEDS_LIMIT = (
+    "Notifications can contain at most 10 email addresses."
+)
+
+# Scheduling error messages
+ERROR_SCHEDULING_FORMAT = (
+    "Invalid 'start_time' format. Use ISO8601 format (YYYY-MM-DDTHH:mm:SSZ)."
+)
+ERROR_SCHEDULING_END_BEFORE_START = "End time must be after start time."
+ERROR_SCHEDULING_NOT_DICT = "Scheduling attribute must be a dictionary."
+
+
+def create_client(
+    base_url=TEST_URL,
+    name=TEST_NAME,
+    endpoints=TEST_ENDPOINTS,
+    description=None,
+    notifications=None,
+    scheduling=None,
+    qos_metrics=None,
+    http_username=None,
+    http_password=None,
+):
+    return SDXClient(
+        base_url=base_url,
+        name=name,
+        endpoints=endpoints,
+        description=description,
+        notifications=notifications,
+        scheduling=scheduling,
+        qos_metrics=qos_metrics,
+        http_username=http_username,
+        http_password=http_password,
+    )

old_tests/test_response.py

@@ -0,0 +1,111 @@
+import json
+import unittest
+from sdxlib.sdx_response import SDXResponse
+from test_config import *
+
+
+class SDXResponseTest(unittest.TestCase):
+    def setUp(self):
+        """Initialize a valid SDXResponse object for testing."""
+        self.response = SDXResponse(MOCK_RESPONSE[list(MOCK_RESPONSE.keys())[0]])
+
+    def test_response_initialization_with_valid_json(self):
+        response_json = TEST_VALID_RESPONSE
+        response = SDXResponse(response_json)
+        self.assertEqual(response.service_id, TEST_SERVICE_ID)
+        self.assertEqual(response.name, TEST_NAME)
+        self.assertEqual(response.endpoints, TEST_ENDPOINTS)
+        self.assertEqual(response.ownership, TEST_OWNERSHIP)
+        self.assertEqual(response.creation_date, TEST_CREATION_DATE)
+        self.assertEqual(response.archived_date, TEST_ARCHIVED_DATE)
+        self.assertEqual(response.status, TEST_STATUS)
+        self.assertEqual(response.state, TEST_STATE)
+        self.assertEqual(response.counters_location, TEST_COUNTERS_LOCATION)
+        self.assertEqual(response.last_modified, TEST_LAST_MODIFIED)
+        self.assertEqual(response.current_path, TEST_CURRENT_PATH)
+        self.assertEqual(response.oxp_service_ids, TEST_OXP_SERVICE_IDS)
+
+    def test_response_initialization_with_missing_attributes(self):
+        response_json = TEST_MISSING_ATTRIBUTES_RESPONSE
+        response = SDXResponse(response_json)
+        self.assertIsNone(response.scheduling)
+        self.assertIsNone(response.description)
+
+    def test_response_initialization_with_invalid_json(self):
+        response_json = "invalid_json"
+        with self.assertRaises(TypeError):
+            SDXResponse(response_json)
+
+    def test_valid_sdx_response(self):
+        """Test that a valid SDXResponse object initializes correctly."""
+        self.assertEqual(
+            self.response.service_id,
+            MOCK_RESPONSE[self.response.service_id]["service_id"],
+        )
+        self.assertEqual(
+            self.response.name, MOCK_RESPONSE[self.response.service_id]["name"]
+        )
+        self.assertEqual(self.response.status, "up")
+
+    def test_missing_required_field(self):
+        """Test that missing required fields raise ValueError."""
+        invalid_response = MOCK_RESPONSE[self.response.service_id].copy()
+        del invalid_response["name"]  # Remove required field
+        with self.assertRaises(ValueError) as context:
+            SDXResponse(invalid_response)
+        self.assertIn("Missing required field: name", str(context.exception))
+
+    def test_invalid_field_type(self):
+        """Test that incorrect field types raise ValueError."""
+        invalid_response = MOCK_RESPONSE[self.response.service_id].copy()
+        invalid_response["endpoints"] = "invalid_type"  # Should be a list
+        with self.assertRaises(ValueError) as context:
+            SDXResponse(invalid_response)
+        expected_message = "Invalid type for endpoints: Expected list, got str"
+        self.assertIn(expected_message, str(context.exception))
+
+    def test_string_representation(self):
+        """Test that __str__ returns valid JSON output."""
+        response_str = str(self.response)
+        try:
+            json.loads(response_str)  # Validate JSON format
+        except json.JSONDecodeError:
+            self.fail("SDXResponse __str__ method does not return valid JSON.")
+
+    def test_equality(self):
+        """Test that SDXResponse objects with the same data are equal."""
+        another_response = SDXResponse(MOCK_RESPONSE[self.response.service_id])
+        self.assertEqual(self.response, another_response)
+
+    def test_sdx_response_with_invalid_state(self):
+        """Test that an invalid state value raises a ValueError."""
+        invalid_response = TEST_VALID_RESPONSE.copy()
+        invalid_response["state"] = "invalid_state"
+        with self.assertRaises(ValueError) as context:
+            SDXResponse(invalid_response)
+        self.assertIn("Invalid state", str(context.exception))
+
+    def test_sdx_response_with_invalid_status(self):
+        """Test that an invalid status value raises a ValueError."""
+        invalid_response = TEST_VALID_RESPONSE.copy()
+        invalid_response["status"] = "invalid_status"
+        with self.assertRaises(ValueError) as context:
+            SDXResponse(invalid_response)
+        self.assertIn("Invalid status", str(context.exception))
+
+    def test_sdx_response_with_invalid_json_format(self):
+        """Test that passing a non-dictionary JSON raises a TypeError."""
+        with self.assertRaises(TypeError):
+            SDXResponse("this is not a dictionary")
+
+    def test_sdx_response_missing_required_key(self):
+        """Test that missing a required key raises a ValueError."""
+        invalid_response = TEST_VALID_RESPONSE.copy()
+        del invalid_response["service_id"]
+        with self.assertRaises(ValueError) as context:
+            SDXResponse(invalid_response)
+        self.assertIn("Missing required field: service_id", str(context.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

sdxlib/l2vpn.py

@@ -3,10 +3,13 @@
 from sdxlib.request import _make_request
 from sdxlib.config import BASE_URL
 from sdxlib.response import normalize_result
+from sdxlib.validator import validate_l2vpn_endpoints
 
 VERSION = "1.0"
 
 def create_l2vpn(token: str, name: str, endpoints: List[Dict[str, str]], **kwargs) -> dict:
+    validate_l2vpn_endpoints(endpoints)
+
     url = f"{BASE_URL}/l2vpn/{VERSION}"
     payload = {"name": name, "endpoints": endpoints, **kwargs}
     response, status = _make_request(

sdxlib/validator.py

@@ -115,3 +115,61 @@ def map_http_error(
         "details": details,
     }
 
+from typing import List, Dict, Any
+from sdxlib.exception import SDXException 
+
+SPECIAL_VLAN_KEYWORDS = {"any", "untagged"}
+
+def validate_l2vpn_endpoints(endpoints: List[Dict[str, Any]]) -> None:
+    """
+    Validates L2VPN endpoints according to issue #496 rules.
+    - Rejects 'all'
+    - Enforces consistency for 'any' or 'untagged'
+    - Prevents mixing special keywords with explicit VLANs
+    """
+    if not endpoints or len(endpoints) < 2:
+        raise ValueError("At least two endpoints are required to create an L2VPN.")
+
+    vlans = []
+    for ep in endpoints:
+        if "vlan" not in ep:
+            raise ValueError("Each endpoint must have the 'vlan' key.")
+
+        vlan = ep["vlan"]
+        if isinstance(vlan, str):
+            vlan = vlan.strip().lower()
+        elif not isinstance(vlan, (int, str)):
+            raise ValueError(f"Invalid VLAN value: {vlan}. Must be string or integer.")
+
+        vlans.append(vlan)
+
+    # Reject 'all'
+    if any(v == "all" for v in vlans):
+        raise ValueError(
+            "The VLAN value 'all' is not supported. Use an explicit VLAN ID (e.g., '123') or 'any'/'untagged' consistently across all endpoints."
+        )
+
+    used_special = {v for v in vlans if v in SPECIAL_VLAN_KEYWORDS}
+
+    if len(used_special) > 1:
+        raise ValueError(
+            "Inconsistent special VLAN keywords. "
+            "All endpoints must use the same keyword ('any' or 'untagged')."
+        )
+
+    if used_special:
+        keyword = next(iter(used_special))
+        if any(v not in SPECIAL_VLAN_KEYWORDS for v in vlans):
+            raise ValueError(
+                f"Cannot mix special keyword '{keyword}' with explicit VLAN IDs. "
+                "All endpoints must use the same special keyword."
+            )
+    else:
+        # Optional: validate numeric VLANs
+        for v in vlans:
+            try:
+                vlan_id = int(v)
+                if not (1 <= vlan_id <= 4094):
+                    raise ValueError(f"VLAN ID out of range: {v}. Must be 1–4094.")
+            except ValueError:
+                raise ValueError(f"Invalid VLAN value: {v}. Must be an integer.")

tests/conftest.py

@@ -0,0 +1,14 @@
+import pytest
+from unittest.mock import MagicMock
+
+@pytest.fixture
+def mock_response():
+    resp = MagicMock()
+    resp.headers = {}
+    resp.json = MagicMock()
+    resp.text = ""
+    return resp
+
+@pytest.fixture
+def mock_requests(mocker):
+    return mocker.patch("requests.request")