Service connections

[cdot65]

User description

Checklist for This Pull Request

🚨Please adhere to the guidelines for contributing to this repository.

• Ensure you are submitting your pull request to a branch dedicated to a specific topic/feature/bugfix. Avoid using the master branch for pull requests.
• Target your pull request to the main development branch in this repository.
• Ensure your commit messages follow the project's preferred format.
• Check that your code additions do not fail any linting checks or unit tests.

Pull Request Description

Adding support for service connections.

What does this pull request accomplish?

• Feature addition

Are there any breaking changes included?

• Yes
• No

---

PR Type

Enhancement

---

Description

• Add Service Connection management functionality

• Implement ServiceConnection models and API methods

• Update documentation for Service Connections

• Enhance error handling and validation

---

Changes walkthrough 📝

	Relevant files
Enhancement	12 files client.py Add service_connection mapping to SCM client                          +4/-0      __init__.py Import and expose ServiceConnection class                                +3/-0      service_connections.py Implement ServiceConnection class with CRUD operations      +345/-0  __init__.py Update network config imports and exports                                +10/-2    __init__.py Reorganize object imports and add __all__                                +21/-6    __init__.py Add __all__ for security configuration exports                      +10/-0    __init__.py Import and expose ServiceConnection models                              +27/-0    service_connections.py Implement ServiceConnection Pydantic models                            +124/-0  __init__.py Reorganize network model imports and add __all__                  +45/-27  __init__.py Update object model imports and add __all__                            +80/-24  __init__.py Add __all__ for operations model exports                                  +10/-0    __init__.py Add __all__ for security model exports                                      +26/-0
Tests	5 files factories.py Add ServiceConnection factory classes for testing                +159/-3  test_remote_networks.py Remove commented-out tests in remote networks                        +0/-27    test_service_connections.py Add comprehensive tests for ServiceConnection class            +439/-0  test_service_connections_models.py Add tests for ServiceConnection Pydantic models                    +385/-0  test_schedules_models.py Enhance schedule model tests with date validation                +237/-10
Documentation	6 files README.md Add service_connection to unified client services                +1/-0      index.md Add Service Connections to deployment config docs                +1/-0      service_connections.md Add comprehensive Service Connections documentation            +435/-0  index.md Add Service Connections to deployment models docs                +7/-1      service_connections_models.md Add Service Connection models documentation                            +280/-0  mkdocs.yml Update navigation to include Service Connections                  +2/-0
Additional files	1 files schedules.py +78/-27

---

Need help?

Type /help how to ... in the comments thread for any questions about PR-Agent usage.

Check out the documentation for more information.

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Error Handling The error handling in the fetch method could be improved. It currently raises an InvalidObjectError for multiple different scenarios, which may make it difficult to distinguish between different types of errors. raise InvalidObjectError( message="Invalid response format: expected dictionary", error_code="E003", http_status_code=500, details={"error": "Response is not a dictionary"}, ) # If we receive a data list, we need to extract the first matching item if "data" in response and isinstance(response["data"], list): if len(response["data"]) == 0: raise InvalidObjectError( message=f"No service connection found with name: {name}", error_code="E004", http_status_code=404, details={"error": "Service connection not found"} ) for item in response["data"]: if item.get("name") == name: return ServiceConnectionResponseModel(**item) # If we get here, no exact match was found raise InvalidObjectError( message=f"No exact match found for service connection with name: {name}", error_code="E004", http_status_code=404, details={"error": "Service connection not found"} ) # Direct response with ID field (single resource response) if "id" in response: return ServiceConnectionResponseModel(**response) raise InvalidObjectError( message="Invalid response format", error_code="E003", http_status_code=500, details={"error": "Response format not recognized"} ) Test Coverage While the test coverage is generally good, there could be more edge cases tested, particularly around error handling and pagination in the list method. """Unit tests for the ServiceConnection class.""" import uuid from unittest.mock import MagicMock import pytest from scm.config.deployment import ServiceConnection from scm.exceptions import InvalidObjectError, MissingQueryParameterError from scm.models.deployment import ( ServiceConnectionCreateModel, ServiceConnectionResponseModel, ServiceConnectionUpdateModel, ) @pytest.fixture def sample_service_connection_dict(): """Return a sample service connection dictionary.""" return { "id": str(uuid.uuid4()), "name": "test-connection", "folder": "Service Connections", "ipsec_tunnel": "test-tunnel", "region": "us-east-1", "onboarding_type": "classic", "subnets": ["10.0.0.0/24", "192.168.1.0/24"], "bgp_peer": { "local_ip_address": "192.168.1.1", "peer_ip_address": "192.168.1.2", }, "protocol": { "bgp": { "enable": True, "peer_as": "65000", } }, "source_nat": True, } @pytest.fixture def sample_service_connection_response(sample_service_connection_dict): """Return a sample ServiceConnectionResponseModel.""" return ServiceConnectionResponseModel(**sample_service_connection_dict) @pytest.mark.usefixtures("load_env") class TestServiceConnectionBase: """Base class for ServiceConnection tests.""" @pytest.fixture(autouse=True) def setup_method(self, mock_scm): """Setup method that runs before each test.""" self.mock_scm = mock_scm self.mock_scm.get = MagicMock() self.mock_scm.post = MagicMock() self.mock_scm.put = MagicMock() self.mock_scm.delete = MagicMock() self.client = ServiceConnection(self.mock_scm, max_limit=1000) class TestServiceConnection(TestServiceConnectionBase): """Test suite for ServiceConnection class.""" def test_init(self): """Test initialization of ServiceConnection class.""" service_connection = ServiceConnection(self.mock_scm) assert service_connection.api_client == self.mock_scm assert service_connection.ENDPOINT == "/config/deployment/v1/service-connections" assert service_connection.max_limit == service_connection.DEFAULT_MAX_LIMIT def test_logger_initialization(self): """Test that the logger is properly initialized.""" import logging # Create the ServiceConnection instance service_connection = ServiceConnection(self.mock_scm) # Verify the logger attribute assert hasattr(service_connection, "logger") assert isinstance(service_connection.logger, logging.Logger) # Check that the logger has the correct name # The logger name should be the module's __name__, which is 'scm.config.deployment.service_connections' assert service_connection.logger.name.endswith('service_connections') # Check that the logger exists in the logging system assert service_connection.logger.name in logging.root.manager.loggerDict def test_init_with_custom_max_limit(self): """Test initialization with custom max_limit.""" service_connection = ServiceConnection(self.mock_scm, max_limit=500) assert service_connection.max_limit == 500 def test_init_with_invalid_max_limit(self): """Test initialization with invalid max_limit.""" with pytest.raises(InvalidObjectError) as excinfo: ServiceConnection(self.mock_scm, max_limit="invalid") assert "Invalid max_limit type" in str(excinfo.value) def test_init_with_negative_max_limit(self): """Test initialization with negative max_limit.""" with pytest.raises(InvalidObjectError) as excinfo: ServiceConnection(self.mock_scm, max_limit=-1) assert "Invalid max_limit value" in str(excinfo.value) def test_init_with_excessive_max_limit(self): """Test initialization with excessive max_limit.""" with pytest.raises(InvalidObjectError) as excinfo: service_connection = ServiceConnection(self.mock_scm) service_connection.max_limit = 2000 assert "max_limit exceeds maximum allowed value" in str(excinfo.value) def test_max_limit_property_setter(self): """Test the max_limit property setter.""" # Create service connection with default max_limit service_connection = ServiceConnection(self.mock_scm) assert service_connection.max_limit == service_connection.DEFAULT_MAX_LIMIT # Change max_limit value service_connection.max_limit = 500 assert service_connection.max_limit == 500 # Try invalid values with pytest.raises(InvalidObjectError) as excinfo: service_connection.max_limit = "invalid" assert "Invalid max_limit type" in str(excinfo.value) with pytest.raises(InvalidObjectError) as excinfo: service_connection.max_limit = -10 assert "Invalid max_limit value" in str(excinfo.value) with pytest.raises(InvalidObjectError) as excinfo: service_connection.max_limit = 1500 assert "max_limit exceeds maximum allowed value" in str(excinfo.value) # Confirm original value is unchanged after failed attempts assert service_connection.max_limit == 500 def test_create(self, sample_service_connection_dict): """Test create method.""" self.mock_scm.post.return_value = sample_service_connection_dict # Create a copy without the ID for create operation create_data = sample_service_connection_dict.copy() create_data.pop("id") result = self.client.create(create_data) # Check that correct API call was made self.mock_scm.post.assert_called_once() call_args = self.mock_scm.post.call_args assert call_args[0][0] == self.client.ENDPOINT # Check payload validation payload = call_args[1]["json"] # Should be deserialized from a ServiceConnectionCreateModel ServiceConnectionCreateModel(**payload) # Check result assert isinstance(result, ServiceConnectionResponseModel) assert result.name == sample_service_connection_dict["name"] assert result.folder == sample_service_connection_dict["folder"] def test_get(self, sample_service_connection_dict): """Test get method.""" self.mock_scm.get.return_value = sample_service_connection_dict object_id = sample_service_connection_dict["id"] result = self.client.get(object_id) # Check that correct API call was made expected_endpoint = f"{self.client.ENDPOINT}/{object_id}" self.mock_scm.get.assert_called_once_with(expected_endpoint) # Check result assert isinstance(result, ServiceConnectionResponseModel) assert result.id == uuid.UUID(object_id) assert result.name == sample_service_connection_dict["name"] def test_update(self, sample_service_connection_dict): """Test update method.""" self.mock_scm.put.return_value = sample_service_connection_dict object_id = sample_service_connection_dict["id"] # Create update model update_model = ServiceConnectionUpdateModel(**sample_service_connection_dict) result = self.client.update(update_model) # Check that correct API call was made expected_endpoint = f"{self.client.ENDPOINT}/{object_id}" self.mock_scm.put.assert_called_once() call_args = self.mock_scm.put.call_args assert call_args[0][0] == expected_endpoint # ID should not be in the payload since it's in the URL assert "id" not in call_args[1]["json"] # Check result assert isinstance(result, ServiceConnectionResponseModel) assert result.id == uuid.UUID(object_id) assert result.name == sample_service_connection_dict["name"] def test_delete(self, sample_service_connection_dict): """Test delete method.""" object_id = sample_service_connection_dict["id"] self.client.delete(object_id) # Check that correct API call was made expected_endpoint = f"{self.client.ENDPOINT}/{object_id}" self.mock_scm.delete.assert_called_once_with(expected_endpoint) def test_list(self, sample_service_connection_dict): """Test list method.""" self.mock_scm.get.return_value = { "data": [sample_service_connection_dict], "limit": 20, "offset": 0, "total": 1 } result = self.client.list() # Check that correct API call was made self.mock_scm.get.assert_called_once() call_args = self.mock_scm.get.call_args assert call_args[0][0] == self.client.ENDPOINT assert "folder" in call_args[1]["params"] assert call_args[1]["params"]["folder"] == "Service Connections" # Check result assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], ServiceConnectionResponseModel) assert result[0].name == sample_service_connection_dict["name"] def test_list_response_errors(self): """Test list method error handling for invalid responses.""" # Test non-dictionary response self.mock_scm.get.return_value = ["not", "a", "dictionary"] with pytest.raises(InvalidObjectError) as excinfo: self.client.list() assert "Response is not a dictionary" in str(excinfo.value) # Test missing data field self.mock_scm.get.return_value = {"no_data": "field"} with pytest.raises(InvalidObjectError) as excinfo: self.client.list() assert "\"data\" field missing in the response" in str(excinfo.value) # Test data field not a list self.mock_scm.get.return_value = {"data": "not a list"} with pytest.raises(InvalidObjectError) as excinfo: self.client.list() assert "\"data\" field must be a list" in str(excinfo.value) def test_list_pagination(self, sample_service_connection_dict): """Test list method pagination.""" # Create multiple pages of data conn1 = sample_service_connection_dict.copy() conn1["id"] = str(uuid.uuid4()) conn1["name"] = "connection1" conn2 = sample_service_connection_dict.copy() conn2["id"] = str(uuid.uuid4()) conn2["name"] = "connection2" # Mock responses for pagination self.mock_scm.get.side_effect = [ # First page { "data": [conn1], "limit": 1, "offset": 0, "total": 2 }, # Second page { "data": [conn2], "limit": 1, "offset": 1, "total": 2 }, # Empty page (to end pagination) { "data": [], "limit": 1, "offset": 2, "total": 2 } ] # Set a small limit to force pagination self.client.max_limit = 1 result = self.client.list() # Should have made 3 calls (2 pages + 1 empty page to end pagination) assert self.mock_scm.get.call_count == 3 # We should get both connections in the result assert len(result) == 2 conn_names = [conn.name for conn in result] assert "connection1" in conn_names assert "connection2" in conn_names def test_list_with_name_filter(self, sample_service_connection_dict): """Test list method with name filter.""" self.mock_scm.get.return_value = { "data": [sample_service_connection_dict], "limit": 20, "offset": 0, "total": 1 } result = self.client.list(name="test-connection") # Check that correct API call was made self.mock_scm.get.assert_called_once() call_args = self.mock_scm.get.call_args assert call_args[0][0] == self.client.ENDPOINT assert call_args[1]["params"]["name"] == "test-connection" # Check result assert isinstance(result, list) assert len(result) == 1 assert result[0].name == "test-connection" def test_list_folder_override(self, sample_service_connection_dict): """Test list method folder override.""" self.mock_scm.get.return_value = { "data": [sample_service_connection_dict], "limit": 20, "offset": 0, "total": 1 } # Attempt to use a different folder (should be ignored/overridden) result = self.client.list(folder="Other Folder") # Check that the API call was made with the correct folder call_args = self.mock_scm.get.call_args assert call_args[1]["params"]["folder"] == "Service Connections" # Check result assert isinstance(result, list) assert len(result) == 1 def test_fetch(self, sample_service_connection_dict): """Test fetch method with direct response.""" self.mock_scm.get.return_value = sample_service_connection_dict result = self.client.fetch(name="test-connection") # Check that correct API call was made self.mock_scm.get.assert_called_once() call_args = self.mock_scm.get.call_args assert call_args[0][0] == self.client.ENDPOINT assert call_args[1]["params"]["name"] == "test-connection" assert call_args[1]["params"]["folder"] == "Service Connections" # Check result assert isinstance(result, ServiceConnectionResponseModel) assert result.name == sample_service_connection_dict["name"] def test_fetch_with_data_list(self, sample_service_connection_dict): """Test fetch method with data list response.""" # Create a list response with one matching item list_response = { "data": [ sample_service_connection_dict, # Match { "id": str(uuid.uuid4()), "name": "other-connection", "folder": "Service Connections", "ipsec_tunnel": "test-tunnel", "region": "us-east-1", } ] } self.mock_scm.get.return_value = list_response result = self.client.fetch(name="test-connection") # Check result is the matching connection assert isinstance(result, ServiceConnectionResponseModel) assert result.name == "test-connection" def test_fetch_with_empty_name(self): """Test fetch method with empty name parameter.""" with pytest.raises(MissingQueryParameterError) as excinfo: self.client.fetch(name="") assert '"name" is not allowed to be empty' in str(excinfo.value) def test_fetch_with_no_results(self): """Test fetch method with no matching results.""" # Empty data list self.mock_scm.get.return_value = {"data": []} with pytest.raises(InvalidObjectError) as excinfo: self.client.fetch(name="non-existent") assert "Service connection not found" in str(excinfo.value) def test_fetch_with_non_dict_response(self): """Test fetch method with non-dictionary response.""" # Mock a non-dictionary response to hit line 294 self.mock_scm.get.return_value = ["not", "a", "dictionary"] with pytest.raises(InvalidObjectError) as excinfo: self.client.fetch(name="test-connection") assert "Response is not a dictionary" in str(excinfo.value) def test_fetch_with_no_exact_match(self, sample_service_connection_dict): """Test fetch method with no exact match in results.""" # List with no exact match modified_dict = sample_service_connection_dict.copy() modified_dict["name"] = "similar-connection" self.mock_scm.get.return_value = {"data": [modified_dict]} with pytest.raises(InvalidObjectError) as excinfo: self.client.fetch(name="test-connection") assert "Service connection not found" in str(excinfo.value) def test_fetch_invalid_response(self): """Test fetch method with invalid response format.""" # Response with neither id nor data self.mock_scm.get.return_value = {"error": "some error"} with pytest.raises(InvalidObjectError) as excinfo: self.client.fetch(name="test-connection") assert "Response format not recognized" in str(excinfo.value)

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

✅ All tests successful. No failed tests found.

Files with missing lines	Coverage Δ
scm/client.py	100.00% <ø> (ø)
scm/config/deployment/__init__.py	100.00% <100.00%> (ø)
scm/config/deployment/service_connections.py	100.00% <100.00%> (ø)
scm/config/network/__init__.py	100.00% <100.00%> (ø)
scm/config/objects/__init__.py	100.00% <100.00%> (ø)
scm/config/security/__init__.py	100.00% <100.00%> (ø)
scm/models/deployment/__init__.py	100.00% <100.00%> (ø)
scm/models/deployment/service_connections.py	100.00% <100.00%> (ø)
scm/models/network/__init__.py	100.00% <100.00%> (ø)
scm/models/objects/__init__.py	100.00% <100.00%> (ø)
... and 3 more

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Validate name filter input Consider adding input validation for the 'name' parameter. Check if it's a non-empty string and possibly enforce a maximum length to prevent potential issues with overly long names. scm/config/deployment/service_connections.py [180-206] def list( self, name: Optional[str] = None, **filters, ) -> List[ServiceConnectionResponseModel]: """ Lists service connection objects with optional filtering. Args: name: Optional name filter **filters: Additional query parameters to pass to the API Returns: List[ServiceConnectionResponseModel]: A list of service connection objects """ # Pagination logic limit = self._max_limit offset = 0 all_objects = [] # Folder is required for service-connections API (from OpenAPI spec) params = dict(filters) params["folder"] = "Service Connections" # Always set folder parameter # Add name filter if provided if name: - params["name"] = name + if not isinstance(name, str) or not name.strip(): + raise ValueError("Name filter must be a non-empty string") + if len(name) > 255: # Adjust max length as needed + raise ValueError("Name filter exceeds maximum length of 255 characters") + params["name"] = name.strip() Suggestion importance[1-10]: 8 __ Why: Adding input validation for the 'name' parameter enhances the robustness of the function. It prevents potential issues with empty strings or overly long names, which could cause problems in API calls or database operations.	Medium
	Add error handling for retrieval Add error handling for potential exceptions when fetching or retrieving service connections. docs/sdk/config/deployment/service_connections.md [179-184] -# Fetch by name -service_connection = client.service_connection.fetch(name="aws-service-connection") -print(f"Found service connection: {service_connection.name}") +from scm.exceptions import ObjectNotPresentError -# Get by ID -service_connection_by_id = client.service_connection.get(service_connection.id) -print(f"Retrieved service connection: {service_connection_by_id.name}") +try: + # Fetch by name + service_connection = client.service_connection.fetch(name="aws-service-connection") + print(f"Found service connection: {service_connection.name}") + + # Get by ID + service_connection_by_id = client.service_connection.get(service_connection.id) + print(f"Retrieved service connection: {service_connection_by_id.name}") +except ObjectNotPresentError as e: + print(f"Service connection not found: {e}") Suggestion importance[1-10]: 8 __ Why: The suggestion adds important error handling for potential exceptions when fetching or retrieving service connections, which improves the code's robustness and user experience.	Medium
	Use regex for datetime validation Consider using a regular expression to validate the datetime range format. This would make the validation more concise and potentially more robust. scm/models/objects/schedules.py [153-160] +import re + if not v: raise ValueError("Non-recurring schedule must contain at least one datetime range") - + +datetime_pattern = r'^\d{4}/(?:0[1-9]|1[0-2])/(?:0[1-9]|[12]\d|3[01])@(?:[01]\d|2[0-3]):[0-5]\d-\d{4}/(?:0[1-9]|1[0-2])/(?:0[1-9]|[12]\d|3[01])@(?:[01]\d|2[0-3]):[0-5]\d$' for dt_range in v: - # Check for the format YYYY/M/D@HH:MM - detecting missing leading zeros - dt_parts = dt_range.split("-") - if len(dt_parts) != 2: - raise ValueError("Invalid datetime range format - must contain a single hyphen") + if not re.match(datetime_pattern, dt_range): + raise ValueError("Invalid datetime range format. Expected YYYY/MM/DD@HH:MM-YYYY/MM/DD@HH:MM") Suggestion importance[1-10]: 7 __ Why: Using a regular expression for datetime validation is a more concise and potentially more robust approach. It can catch format errors more efficiently and reduce the amount of code needed for validation.	Medium
	Use parameterized tests Consider using parameterized tests to reduce code duplication and improve test maintainability. This approach allows you to test multiple scenarios with a single test method. tests/scm/config/deployment/test_service_connections.py [241-258] -def test_list_response_errors(self): +@pytest.mark.parametrize("response, expected_error", [ + (["not", "a", "dictionary"], "Response is not a dictionary"), + ({"no_data": "field"}, "\"data\" field missing in the response"), + ({"data": "not a list"}, "\"data\" field must be a list"), +]) +def test_list_response_errors(self, response, expected_error): """Test list method error handling for invalid responses.""" - # Test non-dictionary response - self.mock_scm.get.return_value = ["not", "a", "dictionary"] + self.mock_scm.get.return_value = response with pytest.raises(InvalidObjectError) as excinfo: self.client.list() - assert "Response is not a dictionary" in str(excinfo.value) - - # Test missing data field - self.mock_scm.get.return_value = {"no_data": "field"} - with pytest.raises(InvalidObjectError) as excinfo: - self.client.list() - assert "\"data\" field missing in the response" in str(excinfo.value) - - # Test data field not a list - self.mock_scm.get.return_value = {"data": "not a list"} - with pytest.raises(InvalidObjectError) as excinfo: - self.client.list() - assert "\"data\" field must be a list" in str(excinfo.value) + assert expected_error in str(excinfo.value) Suggestion importance[1-10]: 7 __ Why: Using parameterized tests can significantly improve code maintainability and reduce duplication. It allows testing multiple scenarios more efficiently and makes the test suite more scalable.	Medium
	Add error handling for deletion Include error handling and a confirmation message for the deletion operation. docs/sdk/config/deployment/service_connections.md [293-295] -# Delete by ID -service_connection_id = "123e4567-e89b-12d3-a456-426655440000" -client.service_connection.delete(service_connection_id) +from scm.exceptions import ObjectNotPresentError +try: + # Delete by ID + service_connection_id = "123e4567-e89b-12d3-a456-426655440000" + client.service_connection.delete(service_connection_id) + print(f"Service connection {service_connection_id} deleted successfully") +except ObjectNotPresentError as e: + print(f"Failed to delete service connection: {e}") + Suggestion importance[1-10]: 7 __ Why: The suggestion enhances the deletion operation by adding error handling and a confirmation message, which improves code reliability and provides better feedback to the user.	Medium
	Move import to file top Consider moving the logging import to the top of the file to follow Python's import conventions. This improves code readability and organization. tests/scm/config/deployment/test_service_connections.py [73-89] +import logging + def test_logger_initialization(self): """Test that the logger is properly initialized.""" - import logging - # Create the ServiceConnection instance service_connection = ServiceConnection(self.mock_scm) # Verify the logger attribute assert hasattr(service_connection, "logger") assert isinstance(service_connection.logger, logging.Logger) # Check that the logger has the correct name # The logger name should be the module's __name__, which is 'scm.config.deployment.service_connections' assert service_connection.logger.name.endswith('service_connections') # Check that the logger exists in the logging system assert service_connection.logger.name in logging.root.manager.loggerDict Suggestion importance[1-10]: 3 __ Why: Moving the import statement to the top of the file is a good practice for code organization, but it's a minor improvement that doesn't significantly impact functionality or readability in this context.	Low