diff --git a/inbc-program/inbc/parser/source_app_parser.py b/inbc-program/inbc/parser/source_app_parser.py
index 7251ee2a5..65dd67da0 100644
--- a/inbc-program/inbc/parser/source_app_parser.py
+++ b/inbc-program/inbc/parser/source_app_parser.py
@@ -8,12 +8,13 @@
from ..inbc_exception import InbcException
from ..xml_tag import create_xml_tag
+from inbm_common_lib.utility import clean_input
logger = logging.getLogger(__name__)
def application_add(args: argparse.Namespace) -> str:
- if (args.gpgKeyUri and args.gpgKeyName is None) or (args.gpgKeyName and args.gpgKeyUri is None):
+ if bool(args.gpgKeyUri) != bool(args.gpgKeyName):
raise InbcException(
"Source requires either both gpgKeyUri and gpgKeyName to be provided, or neither of them.")
@@ -35,8 +36,8 @@ def application_add(args: argparse.Namespace) -> str:
manifest += ''
- for source in args.sources:
- manifest += '' + source + ''
+ source_tags = (f'{clean_input(source)}' for source in args.sources)
+ manifest += ''.join(source_tags)
manifest += (''
f'{create_xml_tag(arguments, "filename")}'
@@ -82,7 +83,7 @@ def application_update(args: argparse.Namespace) -> str:
'')
for source in args.sources:
- manifest += '' + source.strip() + ''
+ manifest += '' + clean_input(source.strip()) + ''
manifest += (f'{create_xml_tag(arguments, "filename")}' +
'' +
diff --git a/inbc-program/inbc/parser/source_os_parser.py b/inbc-program/inbc/parser/source_os_parser.py
index f184bfa2e..43ef14a48 100644
--- a/inbc-program/inbc/parser/source_os_parser.py
+++ b/inbc-program/inbc/parser/source_os_parser.py
@@ -4,13 +4,14 @@
SPDX-License-Identifier: Apache-2.0
"""
import argparse
+from inbm_common_lib.utility import clean_input
def os_add(args: argparse.Namespace) -> str:
manifest = 'source' \
''
for source in args.sources:
- manifest += '' + source.strip() + ''
+ manifest += '' + clean_input(source.strip()) + ''
manifest += ''
print("manifest {0}".format(manifest))
@@ -21,7 +22,7 @@ def os_remove(args: argparse.Namespace) -> str:
manifest = 'source' \
''
for source in args.sources:
- manifest += '' + source.strip() + ''
+ manifest += '' + clean_input(source.strip()) + ''
manifest += ''
print("manifest {0}".format(manifest))
@@ -32,7 +33,7 @@ def os_update(args: argparse.Namespace) -> str:
manifest = 'source' \
''
for source in args.sources:
- manifest += '' + source.strip() + ''
+ manifest += '' + clean_input(source.strip()) + ''
manifest += ''
print("manifest {0}".format(manifest))
diff --git a/inbc-program/tests/unit/test_source_app_parser.py b/inbc-program/tests/unit/test_source_app_parser.py
index c16d98595..d338a829c 100644
--- a/inbc-program/tests/unit/test_source_app_parser.py
+++ b/inbc-program/tests/unit/test_source_app_parser.py
@@ -30,11 +30,11 @@ def test_parse_add_arguments_deb822_format_separate_lines_successfully(self):
f = self.arg_parser.parse_args(
['source', 'application', 'add',
'--sources', 'X-Repolib-Name: Google Chrome',
- 'Enabled: yes',
- 'Types: deb',
- 'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub',
- 'Suites: stable',
- 'Components: main',
+ 'Enabled: yes',
+ 'Types: deb',
+ 'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub',
+ 'Suites: stable',
+ 'Components: main',
'--filename', 'intel-gpu-jammy.list'])
self.assertEqual(f.gpgKeyUri, None)
self.assertEqual(f.gpgKeyName, None)
@@ -46,56 +46,55 @@ def test_parse_add_arguments_deb822_format_separate_lines_successfully(self):
'Components: main'])
self.assertEqual(f.filename, 'intel-gpu-jammy.list')
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_raise_application_add_with_only_one_gpgKeyUri_param(self, m_connect):
+ def test_raise_application_add_with_only_one_gpgKeyUri_param(self):
p = self.arg_parser.parse_args(
['source', 'application', 'add',
'--gpgKeyUri', 'https://repositories.intel.com/gpu/intel-graphics.key',
'--sources', 'deb http://example.com/ focal main restricted universe',
'deb-src http://example.com/ focal-security main',
'--filename', 'intel-gpu-jammy.list'])
- with pytest.raises(InbcException,
- match="Source requires either both gpgKeyUri and gpgKeyName "
- "to be provided, or neither of them."):
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ with pytest.raises(InbcException,
+ match="Source requires either both gpgKeyUri and gpgKeyName "
+ "to be provided, or neither of them."):
+ Inbc(p, 'source', False)
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_raise_application_add_with_only_one_gpgKeyName_param(self, m_connect):
+ def test_raise_application_add_with_only_one_gpgKeyName_param(self):
p = self.arg_parser.parse_args(
['source', 'application', 'add',
'--gpgKeyName', 'intel-graphics.gpg',
'--sources', 'deb http://example.com/ focal main restricted universe',
'deb-src http://example.com/ focal-security main',
'--filename', 'intel-gpu-jammy.list'])
- with pytest.raises(InbcException,
- match="Source requires either both gpgKeyUri and gpgKeyName "
- "to be provided, or neither of them."):
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ with pytest.raises(InbcException,
+ match="Source requires either both gpgKeyUri and gpgKeyName "
+ "to be provided, or neither of them."):
+ Inbc(p, 'source', False)
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_create_add_deb_822_format_manifest_successfully(self, m_connect):
+ def test_create_add_deb_822_format_manifest_successfully(self):
p = self.arg_parser.parse_args(
['source', 'application', 'add',
'--sources', 'X-Repolib-Name: Google Chrome',
- 'Enabled: yes',
- 'Types: deb',
- 'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub',
- 'Suites: stable',
- 'Components: main',
+ 'Enabled: yes',
+ 'Types: deb',
+ 'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub',
+ 'Suites: stable',
+ 'Components: main',
'--filename', 'google-chrome.sources'])
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ Inbc(p, 'source', False)
expected = 'source' \
'X-Repolib-Name: Google Chrome' \
'Enabled: yes' \
- 'Types: deb'\
+ 'Types: deb' \
'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub' \
'Suites: stable' \
'Components: main' \
'google-chrome.sources'
self.assertEqual(p.func(p), expected)
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_create_add_all_param_manifest_successfully(self, m_connect):
+ def test_create_add_all_param_manifest_successfully(self):
p = self.arg_parser.parse_args(
['source', 'application', 'add',
'--gpgKeyUri', 'https://repositories.intel.com/gpu/intel-graphics.key',
@@ -103,7 +102,8 @@ def test_create_add_all_param_manifest_successfully(self, m_connect):
'--sources', 'deb http://example.com/ focal main restricted universe',
'deb-src http://example.com/ focal-security main',
'--filename', 'intel-gpu-jammy.list'])
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ Inbc(p, 'source', False)
expected = 'source' \
'https://repositories.intel.com/gpu/intel-graphics.key' \
'intel-graphics.gpg' \
@@ -112,14 +112,14 @@ def test_create_add_all_param_manifest_successfully(self, m_connect):
'intel-gpu-jammy.list'
self.assertEqual(p.func(p), expected)
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_create_add_minus_gpg_manifest_successfully(self, m_connect):
+ def test_create_add_minus_gpg_manifest_successfully(self):
p = self.arg_parser.parse_args(
['source', 'application', 'add',
'--sources', 'deb http://example.com/ focal main restricted universe',
'deb-src http://example.com/ focal-security main',
'--filename', 'intel-gpu-jammy.list'])
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ Inbc(p, 'source', False)
expected = 'source' \
'' \
'deb http://example.com/ focal main restricted universe' \
@@ -142,24 +142,24 @@ def test_parse_minus_gpg_remove_arguments_successfully(self):
self.assertEqual(f.gpgKeyName, None)
self.assertEqual(f.filename, 'intel-gpu-jammy.list')
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_create_remove_manifest_all_params_successfully(self, m_connect):
+ def test_create_remove_manifest_all_params_successfully(self):
p = self.arg_parser.parse_args(
['source', 'application', 'remove',
'--gpgKeyName', 'intel-gpu-jammy.gpg',
'--filename', 'intel-gpu-jammy.list'])
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ Inbc(p, 'source', False)
expected = 'source' \
'intel-gpu-jammy.gpg' \
'intel-gpu-jammy.list'
self.assertEqual(p.func(p), expected)
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_create_remove_manifest_minus_gpg_successfully(self, m_connect):
+ def test_create_remove_manifest_minus_gpg_successfully(self):
p = self.arg_parser.parse_args(
['source', 'application', 'remove',
'--filename', 'intel-gpu-jammy.list'])
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ Inbc(p, 'source', False)
expected = 'source' \
'intel-gpu-jammy.list'
self.assertEqual(p.func(p), expected)
@@ -174,25 +174,25 @@ def test_parse_update_arguments_successfully(self):
'deb-src http://example.com/ focal-security main'])
self.assertEqual(f.filename, 'intel-gpu-jammy.list')
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_create_update_manifest_successfully(self, m_connect):
+ def test_create_update_manifest_successfully(self):
p = self.arg_parser.parse_args(
['source', 'application', 'update',
'--sources', 'deb http://example.com/ focal main restricted universe',
'deb-src http://example.com/ focal-security main',
'--filename', 'intel-gpu-jammy.list'])
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ Inbc(p, 'source', False)
expected = 'source' \
'deb http://example.com/ focal main restricted universe' \
'deb-src http://example.com/ focal-security main' \
'intel-gpu-jammy.list'
self.assertEqual(p.func(p), expected)
- @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
- def test_create_list_manifest_successfully(self, m_connect):
+ def test_create_list_manifest_successfully(self):
p = self.arg_parser.parse_args(
['source', 'application', 'list'])
- Inbc(p, 'source', False)
+ with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+ Inbc(p, 'source', False)
expected = 'source' \
'
'
self.assertEqual(p.func(p), expected)
diff --git a/inbm-lib/inbm_common_lib/utility.py b/inbm-lib/inbm_common_lib/utility.py
index 6c10c4d59..31633b100 100644
--- a/inbm-lib/inbm_common_lib/utility.py
+++ b/inbm-lib/inbm_common_lib/utility.py
@@ -180,11 +180,11 @@ def canonicalize_uri(url: str) -> CanonicalUri:
WARNING: a string with no forward slashes will have a scheme added. e.g., 'a' -> 'https://a'
- @param url: the url
+ @param url: URL
@return canonicalized version"""
if url and URL_NULL_CHAR in url:
- raise UrlSecurityException("Unsafe characters detected in the url. Cannot process the request.")
+ raise UrlSecurityException("Unsafe characters detected in the URL. Cannot process the request.")
return CanonicalUri(value=url_normalize.url_normalize(url))
@@ -205,7 +205,9 @@ def is_within_directory(directory: str, target: str) -> bool:
return prefix == abs_directory
-def safe_extract(tarball: tarfile.TarFile, path: str = ".", members: Optional[Iterable[tarfile.TarInfo]] = None, *, numeric_owner: bool = False) -> None:
+def safe_extract(tarball: tarfile.TarFile,
+ path: str = ".",
+ members: Optional[Iterable[tarfile.TarInfo]] = None, *, numeric_owner: bool = False) -> None:
"""Avoid path traversal when extracting tarball
@param tarball: tarball to extract
diff --git a/inbm/dispatcher-agent/dispatcher/source/constants.py b/inbm/dispatcher-agent/dispatcher/source/constants.py
index 908aab9cb..f066afe5b 100644
--- a/inbm/dispatcher-agent/dispatcher/source/constants.py
+++ b/inbm/dispatcher-agent/dispatcher/source/constants.py
@@ -6,7 +6,7 @@
from enum import Enum, unique
from dataclasses import dataclass, field
-from typing import Optional
+from typing import Optional, List
UBUNTU_APT_SOURCES_LIST = "/etc/apt/sources.list"
UBUNTU_APT_SOURCES_LIST_D = "/etc/apt/sources.list.d"
@@ -21,27 +21,27 @@ class ApplicationSourceList:
@dataclass(kw_only=True, frozen=True)
class SourceParameters:
- sources: list[str] = field(default_factory=lambda: [])
+ sources: List[str] = field(default_factory=list)
@dataclass(kw_only=True, frozen=True)
class ApplicationAddSourceParameters:
- file_name: str
- sources: list[str] = field(default_factory=lambda: [])
+ source_list_file_name: str
+ sources: List[str] = field(default_factory=list)
gpg_key_uri: Optional[str] = field(default=None)
gpg_key_name: Optional[str] = field(default=None)
@dataclass(kw_only=True, frozen=True)
class ApplicationRemoveSourceParameters:
- file_name: str
+ source_list_file_name: str
gpg_key_name: Optional[str] = field(default=None)
@dataclass(kw_only=True, frozen=True)
class ApplicationUpdateSourceParameters:
- file_name: str
- sources: list[str] = field(default_factory=lambda: [])
+ source_list_file_name: str
+ sources: List[str] = field(default_factory=list)
@unique
diff --git a/inbm/dispatcher-agent/dispatcher/source/linux_gpg_key.py b/inbm/dispatcher-agent/dispatcher/source/linux_gpg_key.py
index ba1fe4be1..62317d893 100644
--- a/inbm/dispatcher-agent/dispatcher/source/linux_gpg_key.py
+++ b/inbm/dispatcher-agent/dispatcher/source/linux_gpg_key.py
@@ -13,21 +13,6 @@
logger = logging.getLogger(__name__)
-def remove_gpg_key_if_exists(gpg_key_name: str) -> None:
- """Linux - Removes a GPG key file if it exists
-
- @param gpg_key_name: name of GPG key file to remove (file under LINUX_GPG_KEY_PATH)
- """
- try:
- key_path = os.path.join(LINUX_GPG_KEY_PATH, gpg_key_name)
- if os.path.exists(key_path):
- os.remove(key_path)
- # it's OK if the key is not there
-
- except OSError as e:
- raise SourceError(f"Error checking or deleting GPG key: {gpg_key_name}") from e
-
-
def add_gpg_key(remote_key_path: str, key_store_name: str) -> None:
"""Linux - Adds a GPG key from a remote source
@@ -61,4 +46,4 @@ def add_gpg_key(remote_key_path: str, key_store_name: str) -> None:
raise SourceError(f"Error getting GPG key from remote source: {e}")
except subprocess.CalledProcessError as e:
- raise SourceError(f"Error running gpg command to dearmor key: {e}")
+ raise SourceError(f"Error running GPG command to dearmor key: {e}")
diff --git a/inbm/dispatcher-agent/dispatcher/source/source_command.py b/inbm/dispatcher-agent/dispatcher/source/source_command.py
index 4b30b5cc8..c15aaa6e7 100644
--- a/inbm/dispatcher-agent/dispatcher/source/source_command.py
+++ b/inbm/dispatcher-agent/dispatcher/source/source_command.py
@@ -99,7 +99,7 @@ def _handle_app_source_command(
Handle the application source commands.
@param parsed_head: XmlHandler with command information
- @param os_type: os type
+ @param os_type: OS type
@param app_action: The action to be performed
@return Result
"""
@@ -120,7 +120,7 @@ def _handle_app_source_command(
logger.info(f"Optional GPG key parameters not present in manifest")
filename = parsed_head.get_children("applicationSource/remove/repo")["filename"]
application_source_manager.remove(
- ApplicationRemoveSourceParameters(file_name=filename, gpg_key_name=keyname)
+ ApplicationRemoveSourceParameters(source_list_file_name=filename, gpg_key_name=keyname)
)
return Result(status=200, message="SUCCESS")
@@ -144,7 +144,7 @@ def _handle_app_source_command(
application_source_manager.add(
ApplicationAddSourceParameters(
- file_name=repo_filename,
+ source_list_file_name=repo_filename,
gpg_key_name=gpg_key_name,
gpg_key_uri=gpg_key_uri,
sources=add_source_pkgs,
@@ -161,7 +161,7 @@ def _handle_app_source_command(
application_source_manager.update(
ApplicationUpdateSourceParameters(
- file_name=repo_filename,
+ source_list_file_name=repo_filename,
sources=update_source_pkgs,
)
)
diff --git a/inbm/dispatcher-agent/dispatcher/source/ubuntu_source_manager.py b/inbm/dispatcher-agent/dispatcher/source/ubuntu_source_manager.py
index 8838ff58a..fa321d08a 100644
--- a/inbm/dispatcher-agent/dispatcher/source/ubuntu_source_manager.py
+++ b/inbm/dispatcher-agent/dispatcher/source/ubuntu_source_manager.py
@@ -17,8 +17,9 @@
ApplicationUpdateSourceParameters,
SourceParameters,
)
-from dispatcher.source.source_manager import ApplicationSourceManager, OsSourceManager
-from dispatcher.source.linux_gpg_key import remove_gpg_key_if_exists, add_gpg_key
+from .constants import LINUX_GPG_KEY_PATH
+from .source_manager import ApplicationSourceManager, OsSourceManager
+from .linux_gpg_key import add_gpg_key
from inbm_common_lib.utility import (
get_canonical_representation_of_path,
@@ -105,7 +106,7 @@ def add(self, parameters: ApplicationAddSourceParameters) -> None:
# Step 2: Add the source
try:
create_file_with_contents(
- os.path.join(UBUNTU_APT_SOURCES_LIST_D, parameters.file_name), parameters.sources
+ os.path.join(UBUNTU_APT_SOURCES_LIST_D, parameters.source_list_file_name), parameters.sources
)
except (IOError, OSError) as e:
raise SourceError(f"Error adding application source list: {e}")
@@ -141,23 +142,25 @@ def remove(self, parameters: ApplicationRemoveSourceParameters) -> None:
"""
if parameters.gpg_key_name:
# Remove the GPG key (Optional)
- remove_gpg_key_if_exists(parameters.gpg_key_name)
+ path = os.path.join(LINUX_GPG_KEY_PATH, parameters.gpg_key_name)
+ if not remove_file(path):
+ logger.warning(f"Unable to remove GPG key: {path}")
# Remove the file under /etc/apt/sources.list.d
try:
if (
- os.path.sep in parameters.file_name
- or parameters.file_name == ".."
- or parameters.file_name == "."
+ os.path.sep in parameters.source_list_file_name
+ or parameters.source_list_file_name == ".."
+ or parameters.source_list_file_name == "."
):
- raise SourceError(f"Invalid file name: {parameters.file_name}")
+ raise SourceError(f"Invalid file name: {parameters.source_list_file_name}")
if not remove_file(
get_canonical_representation_of_path(
- os.path.join(UBUNTU_APT_SOURCES_LIST_D, parameters.file_name)
+ os.path.join(UBUNTU_APT_SOURCES_LIST_D, parameters.source_list_file_name)
)
):
- raise SourceError(f"Error removing file: {parameters.file_name}")
+ raise SourceError(f"Error removing file: {parameters.source_list_file_name}")
except OSError as e:
raise SourceError(f"Error removing file: {e}") from e
@@ -165,7 +168,7 @@ def update(self, parameters: ApplicationUpdateSourceParameters) -> None:
"""Updates a source file in Ubuntu OS source file list under /etc/apt/sources.list.d"""
try:
create_file_with_contents(
- os.path.join(UBUNTU_APT_SOURCES_LIST_D, parameters.file_name), parameters.sources
+ os.path.join(UBUNTU_APT_SOURCES_LIST_D, parameters.source_list_file_name), parameters.sources
)
except IOError as e:
raise SourceError(f"Error occurred while trying to update sources: {e}") from e
diff --git a/inbm/dispatcher-agent/dispatcher/xmlhandler.py b/inbm/dispatcher-agent/dispatcher/xmlhandler.py
index 08c1d535e..533cf3d3c 100644
--- a/inbm/dispatcher-agent/dispatcher/xmlhandler.py
+++ b/inbm/dispatcher-agent/dispatcher/xmlhandler.py
@@ -148,7 +148,7 @@ def get_attribute(self, xpath: str, attribute_name: str) -> str:
@param xpath: path to key
@param attribute_name: name of attribute
- @return: attribute str if found else None
+ @return: value if found else None
"""
logger.debug("XML get attr")
element = self._root.find(xpath)
@@ -190,7 +190,7 @@ def set_attribute(self, xpath, attribute_value) -> bytes:
@param xpath: path to key
@param attribute_value: value of attribute to set
- @return: Xml in bytes
+ @return: XML in bytes
@raises: XmlException when failed to update
"""
try:
@@ -208,7 +208,7 @@ def remove_attribute(self, xpath) -> bytes:
"""Remove the attribute from xml if found.
@param xpath: path to key
- @return: Xml in bytes
+ @return: XML in bytes
@raises: XmlException when failed to update
"""
try:
@@ -226,7 +226,7 @@ def remove_attribute(self, xpath) -> bytes:
def get_root_elements(self, key: str, attr: str) -> list:
"""This function retrieves all the elements matching
- the specified element and it's attribute
+ the specified element, and it's attribute
@param key: element name
@param attr: element's attribute name
@return: list
@@ -242,8 +242,8 @@ def get_root_elements(self, key: str, attr: str) -> list:
raise XmlException(f"ERROR while fetching elements from root : {e}")
def _getroot(self, xml: str) -> Any:
- """This function validates and returns the root of the xml
- @param xml: xml contents
+ """This function validates and returns the root of the XML
+ @param xml: XML contents
@return: root path
"""
logger.debug(f"XML : {mask_security_info(xml)}")
diff --git a/inbm/dispatcher-agent/tests/unit/source/test_linux_gpg_key.py b/inbm/dispatcher-agent/tests/unit/source/test_linux_gpg_key.py
index 011ae5c58..df77197f6 100644
--- a/inbm/dispatcher-agent/tests/unit/source/test_linux_gpg_key.py
+++ b/inbm/dispatcher-agent/tests/unit/source/test_linux_gpg_key.py
@@ -1,32 +1,11 @@
import pytest
-from unittest.mock import mock_open, patch
from requests import RequestException
-from dispatcher.dispatcher_exception import DispatcherException
-from dispatcher.source.linux_gpg_key import add_gpg_key, remove_gpg_key_if_exists
+from dispatcher.source.linux_gpg_key import add_gpg_key
from dispatcher.source.source_exception import SourceError
class TestLinuxGpgKey:
- def test_remove_gpg_key_success(self, mocker):
- mocker.patch("os.path.join", return_value="dummy/path/to/key")
- mocker.patch("os.path.exists", return_value=True)
- mock_remove = mocker.patch("os.remove")
-
- remove_gpg_key_if_exists("mock_key_name")
-
- mock_remove.assert_called_once_with("dummy/path/to/key")
-
- def test_remove_gpg_key_os_error(self, mocker):
- mocker.patch("os.path.join", return_value="dummy/path/to/key")
- mocker.patch("os.path.exists", return_value=True)
-
- mocker.patch("os.remove", side_effect=OSError("Mock OS Error"))
-
- with pytest.raises(SourceError) as e:
- remove_gpg_key_if_exists("mock_key_name")
-
- assert "Error checking or deleting GPG key: mock_key_name" in str(e.value)
def test_add_gpg_key_success(self, mocker):
mock_get = mocker.patch("dispatcher.source.linux_gpg_key.requests.get", autospec=True)
diff --git a/inbm/dispatcher-agent/tests/unit/source/test_source_command.py b/inbm/dispatcher-agent/tests/unit/source/test_source_command.py
index a0885ee4f..4d6d8fd19 100644
--- a/inbm/dispatcher-agent/tests/unit/source/test_source_command.py
+++ b/inbm/dispatcher-agent/tests/unit/source/test_source_command.py
@@ -99,7 +99,7 @@ def test_do_source_command_list(
OsType.Ubuntu,
ApplicationRemoveSourceParameters(
gpg_key_name="intel-gpu-jammy.gpg",
- file_name="intel-gpu-jammy.list",
+ source_list_file_name="intel-gpu-jammy.list",
),
),
],
@@ -163,7 +163,7 @@ def test_do_source_command_remove(
"dispatcher.source.source_command.create_application_source_manager",
OsType.Ubuntu,
ApplicationAddSourceParameters(
- file_name="repofilename",
+ source_list_file_name="repofilename",
gpg_key_name="keyname",
gpg_key_uri="gpguri",
sources=["sourceA", "sourceB"],
@@ -226,7 +226,7 @@ def test_do_source_command_add(
"dispatcher.source.source_command.create_application_source_manager",
OsType.Ubuntu,
ApplicationUpdateSourceParameters(
- file_name="filename", sources=["source1", "source2"]
+ source_list_file_name="filename", sources=["source1", "source2"]
),
),
],
diff --git a/inbm/dispatcher-agent/tests/unit/source/test_ubuntu_source_cmd.py b/inbm/dispatcher-agent/tests/unit/source/test_ubuntu_source_cmd.py
index c9172e583..749ab3a4e 100644
--- a/inbm/dispatcher-agent/tests/unit/source/test_ubuntu_source_cmd.py
+++ b/inbm/dispatcher-agent/tests/unit/source/test_ubuntu_source_cmd.py
@@ -196,7 +196,7 @@ class TestUbuntuApplicationSourceManager:
def test_add_app_with_gpg_key_successfully(self):
try:
params = ApplicationAddSourceParameters(
- file_name="intel-gpu-jammy.list",
+ source_list_file_name="intel-gpu-jammy.list",
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"
@@ -206,12 +206,12 @@ def test_add_app_with_gpg_key_successfully(self):
patch("dispatcher.source.ubuntu_source_manager.add_gpg_key")):
command.add(params)
except SourceError as err:
- assert False, f"'UbuntuApplicationSourceManager.add' raised an exception {err}"
+ pytest.fail(f"'UbuntuApplicationSourceManager.add' raised an exception {err}")
def test_add_app_deb_822_format_successfully(self):
try:
params = ApplicationAddSourceParameters(
- file_name="google-chrome.sources",
+ source_list_file_name="google-chrome.sources",
sources="X-Repolib-Name: Google Chrome"
"Enabled: yes"
"Types: deb"
@@ -223,18 +223,18 @@ def test_add_app_deb_822_format_successfully(self):
with patch("builtins.open", new_callable=mock_open()):
command.add(params)
except SourceError as err:
- assert False, f"'UbuntuApplicationSourceManager.add' raised an exception {err}"
+ pytest.fail(f"'UbuntuApplicationSourceManager.add' raised an exception {err}")
def test_update_app_source_successfully(self):
try:
params = ApplicationUpdateSourceParameters(
- file_name="intel-gpu-jammy.list", sources=APP_SOURCE
+ source_list_file_name="intel-gpu-jammy.list", sources=APP_SOURCE
)
command = UbuntuApplicationSourceManager()
with patch("builtins.open", new_callable=mock_open()):
command.update(params)
except SourceError as err:
- assert False, f"'UbuntuApplicationSourceManager.update' raised an exception {err}"
+ pytest.fail(f"'UbuntuApplicationSourceManager.update' raised an exception {err}")
# def test_raises_exception_on_io_error_during_update_app_source_ubuntu(self):
# params = ApplicationUpdateSourceParameters(file_name="intel-gpu-jammy.list",
@@ -267,12 +267,11 @@ def test_list_raises_exception(self):
assert "Error listing application sources" in str(exc_info.value)
@patch("dispatcher.source.ubuntu_source_manager.remove_file", return_value=True)
- @patch("dispatcher.source.ubuntu_source_manager.remove_gpg_key_if_exists")
def test_successfully_remove_gpg_key_and_source_list(
- self, mock_remove_gpg_key, mock_remove_file
+ self, mock_remove_file
):
parameters = ApplicationRemoveSourceParameters(
- gpg_key_name="example_source.gpg", file_name="example_source.list"
+ gpg_key_name="example_source.gpg", source_list_file_name="example_source.list"
)
command = UbuntuApplicationSourceManager()
try:
@@ -280,10 +279,9 @@ def test_successfully_remove_gpg_key_and_source_list(
except SourceError:
self.fail("Remove GPG key raised DispatcherException unexpectedly!")
- @patch("dispatcher.source.ubuntu_source_manager.remove_gpg_key_if_exists")
- def test_raises_when_space_check_fails(self, mock_remove_gpg_key):
+ def test_raises_when_space_check_fails(self):
parameters = ApplicationRemoveSourceParameters(
- gpg_key_name="example_source.gpg", file_name="../example_source.list"
+ gpg_key_name="example_source.gpg", source_list_file_name="../example_source.list"
)
command = UbuntuApplicationSourceManager()
with pytest.raises(SourceError) as ex:
@@ -291,27 +289,11 @@ def test_raises_when_space_check_fails(self, mock_remove_gpg_key):
assert str(ex.value) == "Invalid file name: ../example_source.list"
@patch("dispatcher.source.ubuntu_source_manager.remove_file", return_value=False)
- @patch("dispatcher.source.ubuntu_source_manager.remove_gpg_key_if_exists")
- def test_raises_when_unable_to_remove_file(self, mock_remove_gpg_key, mock_remove_file):
+ def test_raises_when_unable_to_remove_file(self, mock_remove_file):
parameters = ApplicationRemoveSourceParameters(
- gpg_key_name="example_source.gpg", file_name="example_source.list"
+ gpg_key_name="example_source.gpg", source_list_file_name="example_source.list"
)
command = UbuntuApplicationSourceManager()
with pytest.raises(SourceError) as ex:
command.remove(parameters)
assert str(ex.value) == "Error removing file: example_source.list"
-
- @patch(
- "dispatcher.source.ubuntu_source_manager.os.path.join",
- side_effect=OSError("unable to join path"),
- )
- @patch("dispatcher.source.ubuntu_source_manager.remove_file", return_value=False)
- @patch("dispatcher.source.ubuntu_source_manager.remove_gpg_key_if_exists")
- def test_raises_on_os_error(self, mock_remove_gpg_key, mock_remove_file, mock_os_error):
- parameters = ApplicationRemoveSourceParameters(
- gpg_key_name="example_source.gpg", file_name="example_source.list"
- )
- command = UbuntuApplicationSourceManager()
- with pytest.raises(SourceError) as ex:
- command.remove(parameters)
- assert str(ex.value) == "Error removing file: unable to join path"