Enhance Python type hints, documentation, and test case consistency #473
Conversation
|
/review |
Code Review Analysis
Code Review Feedback💡 General suggestions: The PR makes good strides towards improving code maintainability and readability. It is important to ensure that the changes do not introduce any regressions and that they align with the project's coding standards. It would be beneficial to include tests that cover the changes made, especially since the PR description emphasizes the importance of sustainability and the ability to debug the code in the future. ✨ Usage tips:
|
| @@ -13,7 +13,7 @@ | |||
|
|
|||
| from dataclasses import dataclass | |||
| from pathlib import Path | |||
| from typing import Iterable, List, Optional, Union | |||
| from typing import Iterable, Optional, Union | |||
There was a problem hiding this comment.
Consider using from __future__ import annotations to enable postponed evaluation of type annotations, which can help with forward references and readability. [medium]
| @@ -124,7 +124,7 @@ def remove_file(path: Union[str, Path]) -> bool: | |||
| return False | |||
|
|
|||
|
|
|||
| def create_file_with_contents(path: Union[str, Path], contents: List[str]) -> None: | |||
| def create_file_with_contents(path: Union[str, Path], contents: list[str]) -> None: | |||
There was a problem hiding this comment.
Ensure that the create_file_with_contents function handles the case where contents is an empty list to avoid creating empty files unintentionally. [medium]
| @@ -102,15 +102,15 @@ def __init__(self, broker: DispatcherBroker) -> None: | |||
|
|
|||
| def add(self, parameters: ApplicationAddSourceParameters) -> None: | |||
There was a problem hiding this comment.
Refactor the add method to separate concerns, such as URI verification and GPG key addition, into smaller functions for better readability and maintainability. [important]
| @@ -226,7 +226,7 @@ def remove_attribute(self, xpath) -> bytes: | |||
|
|
|||
| def get_root_elements(self, key: str, attr: str) -> list: | |||
There was a problem hiding this comment.
The get_root_elements method's docstring should be updated to reflect the actual return type, which is list[Element] rather than just list. [medium]
| @@ -318,15 +318,15 @@ def test_success_add_gpg_key_method(self, mock_verify_source): | |||
| patch("dispatcher.source.ubuntu_source_manager.add_gpg_key")): | |||
| command.add(parameters) | |||
|
|
|||
| def test_raises_when_space_check_fails(self): | |||
| def test_raises_when_name_check_fails(self): | |||
There was a problem hiding this comment.
Update the test case test_raises_when_name_check_fails to include a test for the new exception message regarding invalid characters in the file name. [important]
| @@ -164,9 +164,9 @@ def test_do_source_command_remove( | |||
| "dispatcher.source.source_command.create_application_source_manager", | |||
| OsType.Ubuntu, | |||
| ApplicationAddSourceParameters( | |||
There was a problem hiding this comment.
Ensure that the test cases are updated to reflect the new parameter order in ApplicationAddSourceParameters and ApplicationUpdateSourceParameters. [important]
| @@ -138,7 +138,7 @@ def create_file_with_contents(path: Union[str, Path], contents: List[str]) -> No | |||
| raise IOError(f"Error while writing file: {str(e)}") | |||
There was a problem hiding this comment.
Suggestion: Instead of converting exceptions to strings and then wrapping them in IOError, directly raise the original exception to preserve the stack trace. [best practice]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| raise IOError(f"Error while writing file: {str(e)}") | |
| except (PermissionError, IsADirectoryError, OSError) as e: | |
| raise IOError("Error while writing file") from e |
| def create_file_with_contents(path: Union[str, Path], contents: list[str]) -> None: | ||
| """ Create a file and add the contents by line |
There was a problem hiding this comment.
Suggestion: Use a context manager when opening a file to ensure it is properly closed in case of an exception. [best practice]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| def create_file_with_contents(path: Union[str, Path], contents: list[str]) -> None: | |
| """ Create a file and add the contents by line | |
| with open(canonical_path, 'w') as f: | |
| for string in contents: | |
| f.write(f"{string}\n") |
| @@ -233,8 +233,8 @@ def validate_file_type(path: List[str]) -> None: | |||
| """ | |||
| logger.debug("Supported file type prefixes: {0}".format(VALID_MAGIC_FILE_TYPE_PREFIXES)) | |||
|
|
|||
| tarball_list: List[str] = [x for x in path if (not str(x).endswith('.mender') and tarfile.is_tarfile(x))] | |||
| extracted_file_list: List[str] = [] | |||
| tarball_list: list[str] = [x for x in path if (not str(x).endswith('.mender') and tarfile.is_tarfile(x))] | |||
There was a problem hiding this comment.
Suggestion: Use a generator expression instead of a list comprehension for better memory efficiency when you only need to iterate once. [performance]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| tarball_list: list[str] = [x for x in path if (not str(x).endswith('.mender') and tarfile.is_tarfile(x))] | |
| tarball_list: Iterator[str] = (x for x in path if (not str(x).endswith('.mender') and tarfile.is_tarfile(x))) |
| @@ -222,7 +222,7 @@ def safe_extract(tarball: tarfile.TarFile, | |||
| tarball.extractall(path, members, numeric_owner=numeric_owner) | |||
There was a problem hiding this comment.
Suggestion: Avoid using 'str.replace' when removing a single character, use 'str.translate' for better performance. [performance]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| tarball.extractall(path, members, numeric_owner=numeric_owner) | |
| return val.translate({ord('\x00'): None}) |
| @@ -233,8 +233,8 @@ def validate_file_type(path: List[str]) -> None: | |||
| """ | |||
| logger.debug("Supported file type prefixes: {0}".format(VALID_MAGIC_FILE_TYPE_PREFIXES)) | |||
There was a problem hiding this comment.
Suggestion: Replace the string concatenation in the logger.debug call with a more efficient f-string. [enhancement]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| logger.debug("Supported file type prefixes: {0}".format(VALID_MAGIC_FILE_TYPE_PREFIXES)) | |
| logger.debug(f"Supported file type prefixes: {VALID_MAGIC_FILE_TYPE_PREFIXES}") |
| tarball_list: List[str] = [x for x in path if (not str(x).endswith('.mender') and tarfile.is_tarfile(x))] | ||
| extracted_file_list: List[str] = [] | ||
| tarball_list: list[str] = [x for x in path if (not str(x).endswith('.mender') and tarfile.is_tarfile(x))] | ||
| extracted_file_list: list[str] = [] | ||
|
|
||
| for tarball in tarball_list: | ||
| with tarfile.open(tarball) as tar: |
There was a problem hiding this comment.
Suggestion: Consider checking if 'TEMP_EXT_FOLDER' exists before trying to remove it to avoid unnecessary system calls. [enhancement]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| with tarfile.open(tarball) as tar: | |
| if TEMP_EXT_FOLDER.is_dir(): | |
| shutil.rmtree(TEMP_EXT_FOLDER, ignore_errors=True) |
| @@ -124,7 +124,7 @@ def remove_file(path: Union[str, Path]) -> bool: | |||
| return False | |||
|
|
|||
|
|
|||
| def create_file_with_contents(path: Union[str, Path], contents: List[str]) -> None: | |||
| def create_file_with_contents(path: Union[str, Path], contents: list[str]) -> None: | |||
There was a problem hiding this comment.
Suggestion: Use 'pathlib.Path' objects consistently for file system paths to make the code more idiomatic and readable. [maintainability]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| def create_file_with_contents(path: Union[str, Path], contents: list[str]) -> None: | |
| canonical_path = get_canonical_representation_of_path(path) |
| @@ -30,6 +30,7 @@ def do_source_command(parsed_head: XmlHandler, os_type: OsType, dispatcher_broke | |||
|
|
|||
| @param parsed_head: XmlHandler corresponding to the manifest tag | |||
| @param os_type: os type | |||
| @param dispatcher_broker: MQTT broker for Dispatcher agent | |||
| @return Result | |||
| """ | |||
| logger.debug(f"do_source_command: {parsed_head}") | |||
There was a problem hiding this comment.
Suggestion: Remove the redundant logging statement or enhance it to provide more context about the operation being performed. [enhancement]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| logger.debug(f"do_source_command: {parsed_head}") | |
| # If more context is needed, include additional details in the debug message | |
| logger.debug(f"Executing do_source_command with os_type: {os_type} and dispatcher_broker: {dispatcher_broker}") |
| url = parameters.gpg_key_uri | ||
| #URL slicing to remove the last segment (filename) from the URL | ||
| source = url[:-(len(url.split('/')[-1]) + 1)] | ||
| verify_source(source=source, dispatcher_broker=self._dispatcher_broker) |
There was a problem hiding this comment.
Suggestion: Consider validating the URL format before slicing to ensure it's a well-formed URL. This can prevent potential errors when the URL does not conform to standard patterns. [validation]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| url = parameters.gpg_key_uri | |
| #URL slicing to remove the last segment (filename) from the URL | |
| source = url[:-(len(url.split('/')[-1]) + 1)] | |
| verify_source(source=source, dispatcher_broker=self._dispatcher_broker) | |
| from urllib.parse import urlparse | |
| url = parameters.gpg_key_uri | |
| parsed_url = urlparse(url) | |
| if not all([parsed_url.scheme, parsed_url.netloc]): | |
| raise SourceError(f"Invalid GPG key URL: {url}") | |
| # URL slicing to remove the last segment (filename) from the URL | |
| source = url[:-(len(url.split('/')[-1]) + 1)] | |
| verify_source(source=source, dispatcher_broker=self._dispatcher_broker) |
| try: | ||
| url = parameters.gpg_key_uri | ||
| #URL slicing to remove the last segment (filename) from the URL | ||
| source = url[:-(len(url.split('/')[-1]) + 1)] | ||
| verify_source(source=source, dispatcher_broker=self._dispatcher_broker) | ||
| except (DispatcherException, IndexError) as err: | ||
| raise SourceError(f"Source Gpg key URI verification check failed: {err}") | ||
| raise SourceError(f"Source GPG key URI verification check failed: {err}") |
There was a problem hiding this comment.
Suggestion: Instead of catching a generic IndexError, which could be raised by any list indexing operation, check the URL parts length explicitly to provide a clearer error message if the URL is not as expected. [error handling]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| try: | |
| url = parameters.gpg_key_uri | |
| #URL slicing to remove the last segment (filename) from the URL | |
| source = url[:-(len(url.split('/')[-1]) + 1)] | |
| verify_source(source=source, dispatcher_broker=self._dispatcher_broker) | |
| except (DispatcherException, IndexError) as err: | |
| raise SourceError(f"Source Gpg key URI verification check failed: {err}") | |
| raise SourceError(f"Source GPG key URI verification check failed: {err}") | |
| try: | |
| url = parameters.gpg_key_uri | |
| url_parts = url.split('/') | |
| if len(url_parts) < 2: | |
| raise SourceError(f"Source GPG key URI is not properly formatted: {url}") | |
| # URL slicing to remove the last segment (filename) from the URL | |
| source = url[:-(len(url_parts[-1]) + 1)] | |
| verify_source(source=source, dispatcher_broker=self._dispatcher_broker) | |
| except DispatcherException as err: | |
| raise SourceError(f"Source GPG key URI verification check failed: {err}") |
| if parameters.gpg_key_name and parameters.gpg_key_uri: | ||
| try: | ||
| url = parameters.gpg_key_uri | ||
| #URL slicing to remove the last segment (filename) from the URL | ||
| source = url[:-(len(url.split('/')[-1]) + 1)] | ||
| verify_source(source=source, dispatcher_broker=self._dispatcher_broker) | ||
| except (DispatcherException, IndexError) as err: | ||
| raise SourceError(f"Source Gpg key URI verification check failed: {err}") | ||
| raise SourceError(f"Source GPG key URI verification check failed: {err}") | ||
| # Step 2: Add key (Optional) | ||
| add_gpg_key(parameters.gpg_key_uri, parameters.gpg_key_name) |
There was a problem hiding this comment.
Suggestion: Use a context manager when adding a GPG key to ensure that resources are properly managed even if an error occurs. [resource management]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| add_gpg_key(parameters.gpg_key_uri, parameters.gpg_key_name) | |
| # Step 2: Add key (Optional) | |
| with add_gpg_key(parameters.gpg_key_uri, parameters.gpg_key_name) as key_added: | |
| if not key_added: | |
| raise SourceError(f"Failed to add GPG key from: {parameters.gpg_key_uri}") |
| ApplicationAddSourceParameters( | ||
| source_list_file_name="repofilename", | ||
| gpg_key_name="keyname", | ||
| gpg_key_uri="gpguri", | ||
| source_list_file_name="repofilename", | ||
| sources=["sourceA", "sourceB"], |
There was a problem hiding this comment.
Suggestion: Ensure consistent ordering of parameters when creating instances of ApplicationAddSourceParameters for readability and maintainability. [maintainability]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| ApplicationAddSourceParameters( | |
| source_list_file_name="repofilename", | |
| gpg_key_name="keyname", | |
| gpg_key_uri="gpguri", | |
| source_list_file_name="repofilename", | |
| sources=["sourceA", "sourceB"], | |
| ApplicationAddSourceParameters( | |
| source_list_file_name="repofilename", | |
| gpg_key_name="keyname", | |
| gpg_key_uri="gpguri", | |
| sources=["sourceA", "sourceB"], | |
| ), |
| ApplicationUpdateSourceParameters( | ||
| source_list_file_name="filename", sources=["source1", "source2"] | ||
| sources=["source1", "source2"], source_list_file_name="filename" |
There was a problem hiding this comment.
Suggestion: Ensure consistent ordering of parameters when creating instances of ApplicationUpdateSourceParameters for readability and maintainability. [maintainability]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| ApplicationUpdateSourceParameters( | |
| source_list_file_name="filename", sources=["source1", "source2"] | |
| sources=["source1", "source2"], source_list_file_name="filename" | |
| ApplicationUpdateSourceParameters( | |
| source_list_file_name="filename", | |
| sources=["source1", "source2"] | |
| ), |
| parameters = ApplicationRemoveSourceParameters( | ||
| gpg_key_name="example_source.gpg", source_list_file_name="../example_source.list" | ||
| gpg_key_name="example_source.gpg", source_list_file_name="invalid/chars*in:name.list" | ||
| ) |
There was a problem hiding this comment.
Suggestion: Validate the source_list_file_name before using it to prevent potential security risks. [security]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| parameters = ApplicationRemoveSourceParameters( | |
| gpg_key_name="example_source.gpg", source_list_file_name="../example_source.list" | |
| gpg_key_name="example_source.gpg", source_list_file_name="invalid/chars*in:name.list" | |
| ) | |
| if not validate_file_name("invalid/chars*in:name.list"): | |
| raise ValueError("Invalid file name provided") | |
| parameters = ApplicationRemoveSourceParameters( | |
| gpg_key_name="example_source.gpg", source_list_file_name="invalid/chars*in:name.list" | |
| ) |
| with pytest.raises(SourceError) as ex: | ||
| command.add(parameters) |
There was a problem hiding this comment.
Suggestion: Use a context manager to ensure that resources are properly managed and exceptions are handled. [best practice]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| with pytest.raises(SourceError) as ex: | |
| command.add(parameters) | |
| with pytest.raises(SourceError) as ex: | |
| with command: | |
| command.add(parameters) |
| params = ApplicationAddSourceParameters( | ||
| source_list_file_name="google-chrome.sources", | ||
| sources="deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main", | ||
| sources=["deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main"], | ||
| gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub", | ||
| gpg_key_name="google-chrome.gpg" | ||
| ) |
There was a problem hiding this comment.
Suggestion: Replace the hardcoded source list file name with a variable or a constant. [maintainability]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| params = ApplicationAddSourceParameters( | |
| source_list_file_name="google-chrome.sources", | |
| sources="deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main", | |
| sources=["deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main"], | |
| gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub", | |
| gpg_key_name="google-chrome.gpg" | |
| ) | |
| CHROME_SOURCE_LIST_FILE = "google-chrome.sources" | |
| params = ApplicationAddSourceParameters( | |
| source_list_file_name=CHROME_SOURCE_LIST_FILE, | |
| sources=["deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main"], | |
| gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub", | |
| gpg_key_name="google-chrome.gpg" | |
| ) |
| gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub", | ||
| gpg_key_name="google-chrome.gpg" |
There was a problem hiding this comment.
Suggestion: Use a variable for the GPG key URI to avoid duplication and facilitate changes. [maintainability]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub", | |
| gpg_key_name="google-chrome.gpg" | |
| GOOGLE_GPG_KEY_URI = "https://dl-ssl.google.com/linux/linux_signing_key.pub" | |
| gpg_key_uri=GOOGLE_GPG_KEY_URI, | |
| gpg_key_name="google-chrome.gpg" |
There was a problem hiding this comment.
constant is being used. Just at the method level.
| parameters = ApplicationRemoveSourceParameters( | ||
| gpg_key_name="example_source.gpg", source_list_file_name="../example_source.list" | ||
| gpg_key_name="example_source.gpg", source_list_file_name="invalid/chars*in:name.list" |
There was a problem hiding this comment.
Suggestion: Avoid using hardcoded strings for file names and use a constant instead. [maintainability]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| gpg_key_name="example_source.gpg", source_list_file_name="invalid/chars*in:name.list" | |
| EXAMPLE_SOURCE_LIST = "example_source.list" | |
| parameters = ApplicationRemoveSourceParameters( | |
| gpg_key_name="example_source.gpg", source_list_file_name=EXAMPLE_SOURCE_LIST | |
| ) |
There was a problem hiding this comment.
A constant is used.
| parameters = ApplicationAddSourceParameters( | ||
| source_list_file_name="example_source.list", | ||
| sources="deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main", | ||
| sources=["deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main"], | ||
| gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub", | ||
| gpg_key_name="name" |
There was a problem hiding this comment.
Suggestion: Refactor the test to use a helper function for creating ApplicationAddSourceParameters to reduce duplication. [maintainability]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| parameters = ApplicationAddSourceParameters( | |
| source_list_file_name="example_source.list", | |
| sources="deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main", | |
| sources=["deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main"], | |
| gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub", | |
| gpg_key_name="name" | |
| def create_app_add_source_params(file_name, sources, gpg_uri, gpg_name): | |
| return ApplicationAddSourceParameters( | |
| source_list_file_name=file_name, | |
| sources=sources, | |
| gpg_key_uri=gpg_uri, | |
| gpg_key_name=gpg_name | |
| ) | |
| parameters = create_app_add_source_params( | |
| "example_source.list", | |
| ["deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main"], | |
| "https://dl-ssl.google.com/linux/linux_signing_key.pub", | |
| "name" | |
| ) |
| ) | ||
| broker = MockDispatcherBroker.build_mock_dispatcher_broker() | ||
| command = UbuntuApplicationSourceManager(broker) | ||
| with pytest.raises(SourceError) as ex: | ||
| command.remove(parameters) | ||
| assert str(ex.value) == "Invalid file name: ../example_source.list" | ||
| assert str(ex.value) == "Invalid file name: invalid/chars*in:name.list" |
There was a problem hiding this comment.
Suggestion: Abstract the error message construction to a method for consistency and maintainability. [maintainability]
NOTE: Ensure that the suggested code is comprehensive. This serves as an example of the code you could use. The user should verify it after accepting the suggestion.
| assert str(ex.value) == "Invalid file name: invalid/chars*in:name.list" | |
| assert str(ex.value) == build_error_message("Invalid file name", "invalid/chars*in:name.list") |
|
/describe |
|
PR Description updated to latest commit (61da41e) |
nmgaston
changed the title
Code must act as a teacher for future developers
Types of major changes in the pull request
enhancement, bug_fix
Pull request change summary
Details of Pull Request changes by File
1 files
utility.py
inbm-lib/inbm_common_lib/utility.py
Refactor type hints from List to list for Python 3.9+
compatibility and remove unused import.
2 files
source_command.py
inbm/dispatcher-agent/dispatcher/source/source_command.py
Add missing parameter documentation for dispatcher_broker in
do_source_command function.
xmlhandler.py
inbm/dispatcher-agent/dispatcher/xmlhandler.py
Fix grammatical error in get_root_elements method
documentation.
1 files
ubuntu_source_manager.py
inbm/dispatcher-agent/dispatcher/source/ubuntu_source_manager.py
Correct comments and exception messages related to GPG key
handling and improve source verification logic.
2 files
test_source_command.py
inbm/dispatcher-agent/tests/unit/source/test_source_command.py
Update copyright year and refactor test cases to match
updated parameter order in ApplicationAddSourceParameters
and ApplicationUpdateSourceParameters.
test_ubuntu_source_cmd.py
inbm/dispatcher-agent/tests/unit/source/test_ubuntu_source_cmd.py
Update test cases to use lists for sources parameter,
correct exception messages, and fix test method names.