diff --git a/tests/__init__.py b/old_tests/__init__.py similarity index 100% rename from tests/__init__.py rename to old_tests/__init__.py diff --git a/tests/test_base_url.py b/old_tests/test_base_url.py similarity index 100% rename from tests/test_base_url.py rename to old_tests/test_base_url.py diff --git a/tests/test_client_create_method.py b/old_tests/test_client_create_method.py similarity index 100% rename from tests/test_client_create_method.py rename to old_tests/test_client_create_method.py diff --git a/tests/test_client_delete_method.py b/old_tests/test_client_delete_method.py similarity index 100% rename from tests/test_client_delete_method.py rename to old_tests/test_client_delete_method.py diff --git a/tests/test_client_description.py b/old_tests/test_client_description.py similarity index 100% rename from tests/test_client_description.py rename to old_tests/test_client_description.py diff --git a/tests/test_client_endpoints.py b/old_tests/test_client_endpoints.py similarity index 100% rename from tests/test_client_endpoints.py rename to old_tests/test_client_endpoints.py diff --git a/tests/test_client_http_username_password.py b/old_tests/test_client_http_username_password.py similarity index 100% rename from tests/test_client_http_username_password.py rename to old_tests/test_client_http_username_password.py diff --git a/tests/test_client_list_method.py b/old_tests/test_client_list_method.py similarity index 100% rename from tests/test_client_list_method.py rename to old_tests/test_client_list_method.py diff --git a/tests/test_client_name.py b/old_tests/test_client_name.py similarity index 100% rename from tests/test_client_name.py rename to old_tests/test_client_name.py diff --git a/tests/test_client_notifications.py b/old_tests/test_client_notifications.py similarity index 100% rename from tests/test_client_notifications.py rename to old_tests/test_client_notifications.py diff --git a/tests/test_client_qos_metrics.py b/old_tests/test_client_qos_metrics.py similarity index 100% rename from tests/test_client_qos_metrics.py rename to old_tests/test_client_qos_metrics.py diff --git a/tests/test_client_scheduling.py b/old_tests/test_client_scheduling.py similarity index 100% rename from tests/test_client_scheduling.py rename to old_tests/test_client_scheduling.py diff --git a/tests/test_client_update_method.py b/old_tests/test_client_update_method.py similarity index 100% rename from tests/test_client_update_method.py rename to old_tests/test_client_update_method.py diff --git a/tests/test_client_utility_methods.py b/old_tests/test_client_utility_methods.py similarity index 100% rename from tests/test_client_utility_methods.py rename to old_tests/test_client_utility_methods.py diff --git a/old_tests/test_config.py b/old_tests/test_config.py new file mode 100644 index 0000000..b6f4baf --- /dev/null +++ b/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, + ) diff --git a/tests/test_link.py b/old_tests/test_link.py similarity index 100% rename from tests/test_link.py rename to old_tests/test_link.py diff --git a/tests/test_location.py b/old_tests/test_location.py similarity index 100% rename from tests/test_location.py rename to old_tests/test_location.py diff --git a/tests/test_node.py b/old_tests/test_node.py similarity index 100% rename from tests/test_node.py rename to old_tests/test_node.py diff --git a/tests/test_ports.py b/old_tests/test_ports.py similarity index 100% rename from tests/test_ports.py rename to old_tests/test_ports.py diff --git a/old_tests/test_response.py b/old_tests/test_response.py new file mode 100644 index 0000000..550c834 --- /dev/null +++ b/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() diff --git a/tests/test_response_utility_methods.py b/old_tests/test_response_utility_methods.py similarity index 100% rename from tests/test_response_utility_methods.py rename to old_tests/test_response_utility_methods.py diff --git a/tests/test_topology.py b/old_tests/test_topology.py similarity index 100% rename from tests/test_topology.py rename to old_tests/test_topology.py diff --git a/tests/test_topology_response.py b/old_tests/test_topology_response.py similarity index 100% rename from tests/test_topology_response.py rename to old_tests/test_topology_response.py diff --git a/tests/topology_test_config.py b/old_tests/topology_test_config.py similarity index 100% rename from tests/topology_test_config.py rename to old_tests/topology_test_config.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..fc14c22 --- /dev/null +++ b/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") diff --git a/tests/test_config.py b/tests/test_config.py index b6f4baf..6d4a8d2 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,182 +1,5 @@ -from sdxlib.sdx_client import SDXClient +from sdxlib.config import BASE_URL, VERSION -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, - ) +def test_config_values(): + assert BASE_URL == "https://sdxapi.atlanticwave-sdx.ai/production" + assert VERSION == "1.0" diff --git a/tests/test_exception.py b/tests/test_exception.py new file mode 100644 index 0000000..291b621 --- /dev/null +++ b/tests/test_exception.py @@ -0,0 +1,12 @@ +from sdxlib.exception import SDXException + +def test_sdx_exception_basic(): + exc = SDXException(status_code=404, message="Not found") + assert exc.status_code == 404 + assert str(exc) == "SDXException: Not found (status_code=404)" + +def test_sdx_exception_with_details(): + exc = SDXException(400, message="Bad request", error_details={"field": "endpoints"}) + assert "Bad request" in str(exc) + assert exc.error_details == {"field": "endpoints"} + \ No newline at end of file diff --git a/tests/test_l2vpn.py b/tests/test_l2vpn.py new file mode 100644 index 0000000..b7dddc3 --- /dev/null +++ b/tests/test_l2vpn.py @@ -0,0 +1,55 @@ +from sdxlib.l2vpn import create_l2vpn, get_l2vpn +from sdxlib.config import BASE_URL, VERSION + +BASE = f"{BASE_URL}/l2vpn/{VERSION}" + +def test_create_l2vpn_success(mock_requests, mock_response): + mock_response.status_code = 201 + mock_response.headers["Content-Type"] = "application/json" + mock_response.json.return_value = {"id": "svc-abc123", "status": "created"} + mock_requests.return_value = mock_response + + result = create_l2vpn( + token="tok123", + name="test-vpn", + endpoints=[{"port_id": "urn:sdx:port:1", "vlan": "100"}] + ) + + assert result["data"] == {"id": "svc-abc123", "status": "created"} + assert result.get("error") is None + mock_requests.assert_called_once_with( + "POST", f"{BASE}", json=mock_requests.call_args[1]["json"], headers=mock_requests.call_args[1]["headers"], timeout=60 + ) + +def test_create_l2vpn_bad_request(mock_requests, mock_response): + mock_response.status_code = 400 + mock_response.json.return_value = {"error": "Missing endpoint"} + mock_requests.return_value = mock_response + + result = create_l2vpn("tok", "bad", []) + + assert result["status_code"] == 400 + assert "error" in result + +def test_get_l2vpn_not_found(mock_requests, mock_response): + mock_response.status_code = 404 + mock_response.headers["Content-Type"] = "application/json" + mock_response.json.return_value = {"error": "Service not found", "detail": "No L2VPN with id svc-999"} + mock_requests.return_value = mock_response + + result = get_l2vpn("tok", "svc-999") + + assert result["status_code"] == 404 + assert "not found" in result.get("error", "").lower() + assert result.get("error") == "Service not found" + +def test_create_l2vpn_validation_error(mock_requests, mock_response): + mock_response.status_code = 400 + mock_response.headers["Content-Type"] = "application/json" + mock_response.json.return_value = {"error": "Invalid endpoints", "fields": ["endpoints"]} + mock_requests.return_value = mock_response + + result = create_l2vpn("tok", "test", []) # empty endpoints - should fails + + assert result["status_code"] == 400 + assert "invalid" in result.get("error", "").lower() diff --git a/tests/test_networkutils.py b/tests/test_networkutils.py new file mode 100644 index 0000000..4596bfe --- /dev/null +++ b/tests/test_networkutils.py @@ -0,0 +1,52 @@ +from unittest.mock import patch, MagicMock +from sdxlib.networkutils import ping_host, check_connection +import subprocess +import socket + +@patch("subprocess.run") +def test_ping_host_success(mock_run): + mock_run.return_value.stdout = "PING success..." + with patch("sys.stdout", new_callable=MagicMock) as mock_stdout: + ping_host("example.com") + mock_stdout.write.assert_called() + +@patch("subprocess.run") +def test_ping_host_fail(mock_run): + mock_run.side_effect = subprocess.CalledProcessError( + returncode=2, + cmd=["ping", "-c", "4", "invalid.host"], + output=b"", + stderr=b"ping: invalid.host: Temporary failure in name resolution\n" + ) + + with patch("sys.stdout", new_callable=MagicMock) as mock_stdout: + ping_host("invalid.host") + + calls = [call[0][0].lower() for call in mock_stdout.write.call_args_list] + output = " ".join(calls) + + assert "error" in output or "failed" in output or "failure" in output or "resolution" in output + assert mock_run.called + +@patch("subprocess.run") +def test_ping_host_success(mock_run): + mock_run.return_value = MagicMock(stdout="PING success 64 bytes...", returncode=0) + with patch("sys.stdout", new_callable=MagicMock) as mock_stdout: + ping_host("example.com") + assert mock_run.called + assert "success" in "".join([c[0][0] for c in mock_stdout.write.call_args_list]).lower() + +@patch("socket.create_connection") +def test_check_connection_success(mock_conn): + mock_conn.return_value = MagicMock() + with patch("sys.stdout", new_callable=MagicMock): + check_connection("example.com", 80) + assert mock_conn.called + +@patch("socket.create_connection") +def test_check_connection_fail(mock_conn): + mock_conn.side_effect = socket.timeout("timeout") + with patch("sys.stdout", new_callable=MagicMock) as mock_stdout: + check_connection("example.com", 9999) + output = "".join([c[0][0] for c in mock_stdout.write.call_args_list]).lower() + assert "error" in output or "timeout" in output diff --git a/tests/test_request.py b/tests/test_request.py new file mode 100644 index 0000000..d370db2 --- /dev/null +++ b/tests/test_request.py @@ -0,0 +1,69 @@ +import json +import requests +from sdxlib.request import _make_request + +def test_make_request_success_json_dict(mock_requests, mock_response): + mock_response.status_code = 200 + mock_response.headers["Content-Type"] = "application/json" + mock_response.json.return_value = {"key": "value"} + mock_requests.return_value = mock_response + + result, status = _make_request("GET", "http://test", timeout=10) + assert status == 200 + assert result == {"key": "value"} + +def test_make_request_success_json_list(mock_requests, mock_response): + mock_response.status_code = 200 + mock_response.headers["Content-Type"] = "application/json" + mock_response.json.return_value = [{"item": 1}] + mock_requests.return_value = mock_response + + result, status = _make_request("GET", "http://test") + assert status == 200 + assert result == {"data": [{"item": 1}]} + +def test_make_request_timeout(mock_requests): + mock_requests.side_effect = requests.Timeout() + + result, status = _make_request("GET", "http://test") + assert status == 408 + assert "timeout" in result.get("error", "").lower() + +def test_make_request_connection_error(mock_requests): + mock_requests.side_effect = requests.ConnectionError("connection failed") + + result, status = _make_request("GET", "http://test") + assert status == 0 + assert "connection failed" in result.get("error", "") + +def test_make_request_json_decode_error(mock_requests, mock_response): + mock_response.status_code = 200 + mock_response.headers["Content-Type"] = "application/json" + mock_response.json.side_effect = json.JSONDecodeError("bad json", "", 0) + mock_response.text = "not json" + mock_requests.return_value = mock_response + + result, status = _make_request("GET", "http://test") + assert status == 200 + assert "JSON parse error" in result.get("error", "") + +def test_make_request_non_json_success(mock_requests, mock_response): + mock_response.status_code = 200 + mock_response.headers["Content-Type"] = "text/plain" + mock_response.text = "plain text response" + mock_requests.return_value = mock_response + + result, status = _make_request("GET", "http://test") + assert status == 200 + assert "warning" in result + assert "Non-JSON success" in result["warning"] + +def test_make_request_error_non_json(mock_requests, mock_response): + mock_response.status_code = 404 + mock_response.headers["Content-Type"] = "text/html" + mock_response.text = "Not Found" + mock_requests.return_value = mock_response + + result, status = _make_request("GET", "http://test") + assert status == 404 + assert "Non-JSON error response" in result.get("error", "") diff --git a/tests/test_response.py b/tests/test_response.py index 550c834..9199971 100644 --- a/tests/test_response.py +++ b/tests/test_response.py @@ -1,111 +1,17 @@ -import json -import unittest -from sdxlib.sdx_response import SDXResponse -from test_config import * +from sdxlib.response import normalize_result +def test_normalize_result_success(): + payload = {"result": "ok", "id": 123} + normalized = normalize_result(payload, 200) -class SDXResponseTest(unittest.TestCase): - def setUp(self): - """Initialize a valid SDXResponse object for testing.""" - self.response = SDXResponse(MOCK_RESPONSE[list(MOCK_RESPONSE.keys())[0]]) + assert normalized["status_code"] == 200 + assert normalized["data"] == payload + assert normalized.get("error") is None - 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_normalize_result_error(): + payload = {"error": "Invalid parameters", "detail": "..."} + normalized = normalize_result(payload, 400) - 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() + assert normalized["status_code"] == 400 + assert normalized["data"] == payload + assert "error" in normalized diff --git a/tests/test_topology_utils.py b/tests/test_topology_utils.py new file mode 100644 index 0000000..88e73d1 --- /dev/null +++ b/tests/test_topology_utils.py @@ -0,0 +1,158 @@ +from sdxlib.topology_utils import ( + _parse_vlan_value, + _collapse_ints_to_ranges, + _iter_advertised_vlan_ints, + get_topology, + _get_vlans_in_use, + get_available_ports, +) + +def test_parse_vlan_value(): + assert _parse_vlan_value(200) == [200] + assert _parse_vlan_value("777") == [777] + assert _parse_vlan_value("100-103") == [100, 101, 102, 103] + assert _parse_vlan_value("500:503") == [500, 501, 502, 503] + assert _parse_vlan_value("10,20-22,30") == [10, 20, 21, 22, 30] + assert _parse_vlan_value("invalid") == [] + assert _parse_vlan_value(None) == [] + +def test_collapse_ints_to_ranges(): + assert _collapse_ints_to_ranges([]) == "None" + assert _collapse_ints_to_ranges([42]) == "42" + assert _collapse_ints_to_ranges([100, 101, 102, 104]) == "100-102, 104" + assert _collapse_ints_to_ranges([5, 6, 7, 8, 9]) == "5-9" + +def test_iter_advertised_vlan_ints(): + assert list(_iter_advertised_vlan_ints(["100-102"])) == [100, 101, 102] + assert list(_iter_advertised_vlan_ints([[200, 203]])) == [200, 201, 202, 203] + assert list(_iter_advertised_vlan_ints([])) == [] + +def test_get_topology_success(mocker, mock_requests, mock_response): + mock_response.status_code = 200 + mock_response.json.return_value = {"nodes": [], "links": []} + mock_requests.return_value = mock_response + + result = get_topology("tok123") + + assert isinstance(result, dict) + assert "status_code" in result + assert result["status_code"] == 200 + +def test_get_vlans_in_use(mocker): + fake_l2vpns = [ + {"endpoints": [{"port_id": "p1", "vlan": "100-102"}, {"port_id": "p2", "vlan": "500"}]}, + {"endpoints": [{"port_id": "p1", "vlan": "200"}]}, + ] + mocker.patch("sdxlib.topology_utils.get_all_l2vpns", return_value=fake_l2vpns) + + usage = _get_vlans_in_use("tok") + assert usage["p1"] == [100, 101, 102, 200] + assert usage["p2"] == [500] + +def test_get_available_ports_with_used_vlans( mocker): + mock_topology = { + "nodes": [ + { + "id": "urn:sdx:node:example:swA", + "name": "Switch A", + "ports": [ + { + "id": "urn:sdx:port:example:swA:eth1", + "name": "Eth1 Customer", + "status": "up", + "nni": False, + "entities": ["customer-group-1"], + "services": { + "l2vpn_ptp": { + "vlan_range": ["100-200"] + } + } + } + ] + } + ] + } + + mocker.patch( + "sdxlib.topology_utils.get_topology", + return_value=mock_topology + ) + + mocker.patch( + "sdxlib.topology_utils.get_all_l2vpns", + return_value=[ + { + "service_id": "svc-123", + "name": "Test VPN", + "endpoints": [ + {"port_id": "urn:sdx:port:example:swA:eth1", "vlan": "150"} + ] + } + ] + ) + + mocker.patch( + "sdxlib.topology_utils._get_vlans_in_use", + return_value={ + "urn:sdx:port:example:swA:eth1": [150] + } + ) + + ports = get_available_ports("tok") + + assert isinstance(ports, list) + assert len(ports) == 1, f"Ports - {len(ports)}: {ports}" + + if ports: + port_info = ports[0] + assert port_info["Port ID"] == "urn:sdx:port:example:swA:eth1" + assert port_info["Status"] == "up" + assert "150" in port_info["VLANs in Use"] + assert "100-149" in port_info["VLANs Available"] or "100-149, 151-200" == port_info["VLANs Available"] + assert "151-200" in port_info["VLANs Available"] or "100-149, 151-200" == port_info["VLANs Available"] + +def test_get_available_ports_no_l2vpns(mocker): + mock_topology_data = { + "nodes": [ + { + "id": "urn:sdx:node:example:nodeA", + "name": "Node A", + "ports": [ + { + "id": "urn:sdx:port:example:sw1:eth1", + "name": "Customer Eth1", + "status": "up", + "nni": False, + "entities": ["customer-entity-1"], + "services": { + "l2vpn_ptp": { + "vlan_range": ["100-200"] + } + } + } + ] + } + ] + } + + mocker.patch( + "sdxlib.topology_utils.get_topology", + return_value=mock_topology_data + ) + + mocker.patch("sdxlib.topology_utils.get_all_l2vpns", return_value=[]) + mocker.patch("sdxlib.topology_utils._get_vlans_in_use", return_value={}) + + ports = get_available_ports("tok") + + assert isinstance(ports, list) + assert len(ports) >= 1, f"No ports: {ports}" + + if ports: + p = ports[0] + assert p["Port ID"] == "urn:sdx:port:example:sw1:eth1" + assert p["Domain"] == "example" + assert p["Device"] == "sw1" + assert p["Port"] == "eth1" + assert p["VLANs Available"] == "100-200" + assert p["VLANs in Use"] == "None" diff --git a/tests/test_validator.py b/tests/test_validator.py new file mode 100644 index 0000000..9a4e7b5 --- /dev/null +++ b/tests/test_validator.py @@ -0,0 +1,106 @@ +import pytest +from sdxlib.validator import ( + validate_required_url, + validate_name, + is_valid_email, + validate_notifications, + map_http_error +) +from unittest.mock import MagicMock + +def test_validate_required_url_ok(): + assert validate_required_url("https://example.com") == "https://example.com" + +def test_validate_required_url_fail_empty(): + with pytest.raises(ValueError, match="BASE_URL must be a non-empty string"): + validate_required_url(" ") + +def test_validate_required_url_fail_not_str(): + with pytest.raises(ValueError): + validate_required_url(123) + +def test_validate_name_ok(): + assert validate_name("my-service-name") == "my-service-name" + +def test_validate_name_empty(): + with pytest.raises(ValueError, match="name must be a non-empty string"): + validate_name("") + +def test_validate_name_too_long(): + long_name = "a" * 51 + with pytest.raises(ValueError, match="name must be at most 50 characters"): + validate_name(long_name) + +def test_is_valid_email(): + assert is_valid_email("user@example.com") is True + assert is_valid_email("user.name@sub.domain.org") is True + assert is_valid_email("invalid") is False + assert is_valid_email("no@") is False + assert is_valid_email("@domain") is False + assert is_valid_email("no@at") is True + assert is_valid_email(None) is False + +def test_validate_notifications_none(): + assert validate_notifications(None) is None + +def test_validate_notifications_valid(): + input_list = [{"email": "a@example.com"}, {"email": "b@domain.org"}] + result = validate_notifications(input_list) + assert result == input_list + +def test_validate_notifications_too_many(): + too_many = [{"email": f"user{i}@test.com"} for i in range(11)] + with pytest.raises(ValueError, match="notifications can contain at most 10 entries"): + validate_notifications(too_many) + +def test_validate_notifications_invalid_email(): + invalid = [{"email": "bad-email"}] + with pytest.raises(ValueError, match="invalid email format"): + validate_notifications(invalid) + +def test_validate_notifications_not_dict(): + bad = [{"email": "ok@example.com"}, "not-a-dict"] + with pytest.raises(ValueError, match="each notification must be a dict"): + validate_notifications(bad) + +def test_map_http_error_json(): + mock_response = MagicMock() + mock_response.status_code = 400 + mock_response.headers = {"Content-Type": "application/json"} + mock_response.json.return_value = {"error": "bad request"} + + mock_exc = MagicMock() + mock_exc.response = mock_response + + result = map_http_error(None, mock_exc, "create") + assert result["status_code"] == 400 + assert result["message"] == "Invalid JSON or incomplete/incorrect body" + assert result["details"] == {"error": "bad request"} + +def test_map_http_error_text(): + mock_response = MagicMock() + mock_response.status_code = 500 + mock_response.headers = {"Content-Type": "text/plain"} + mock_response.text = "Server error" + + mock_exc = MagicMock() + mock_exc.response = mock_response + + result = map_http_error(None, mock_exc, "update") + assert result["status_code"] == 500 + assert result["message"] == "Unknown error" + assert result["details"] == "Server error" + +def test_map_http_error_text(): + mock_response = MagicMock() + mock_response.status_code = 500 + mock_response.headers = {"Content-Type": "text/plain"} + mock_response.text = "Server error" + + mock_exc = MagicMock() + mock_exc.response = mock_response + + result = map_http_error(None, mock_exc, "update") + assert result["status_code"] == 500 + assert result["message"] == "Unknown error" + assert result["details"] == "Server error"