From 399106462eed7c75f9100f8ef97655d72d9794dd Mon Sep 17 00:00:00 2001 From: arturo-seijas <102022572+arturo-seijas@users.noreply.github.com> Date: Thu, 7 Mar 2024 19:34:33 +0100 Subject: [PATCH] Refactor jenkins.py (#119) --- .github/workflows/integration_test.yaml | 2 +- src-docs/jenkins.py.md | 112 +++++++------------ src/actions.py | 4 +- src/agent.py | 8 +- src/charm.py | 8 +- src/jenkins.py | 139 +++++++++++------------- tests/integration/conftest.py | 2 +- tests/unit/conftest.py | 24 ++-- tests/unit/test_actions.py | 32 +++--- tests/unit/test_charm.py | 48 ++++---- tests/unit/test_jenkins.py | 9 +- tests/unit/test_timerange.py | 4 +- 12 files changed, 168 insertions(+), 224 deletions(-) diff --git a/.github/workflows/integration_test.yaml b/.github/workflows/integration_test.yaml index 4dd4afa5..1fe01486 100644 --- a/.github/workflows/integration_test.yaml +++ b/.github/workflows/integration_test.yaml @@ -11,7 +11,7 @@ jobs: channel: 1.28-strict/stable extra-arguments: | --kube-config ${GITHUB_WORKSPACE}/kube-config - modules: '["test_ingress.py", "test_jenkins.py", "test_k8s_agent.py", "test_machine_agent.py", "test_plugins.py", "test_proxy.py", "test_cos.py"]' + modules: '["test_cos.py", "test_ingress.py", "test_jenkins.py", "test_k8s_agent.py", "test_machine_agent.py", "test_plugins.py", "test_proxy.py"]' pre-run-script: | -c "sudo microk8s config > ${GITHUB_WORKSPACE}/kube-config chmod +x tests/integration/pre_run_script.sh diff --git a/src-docs/jenkins.py.md b/src-docs/jenkins.py.md index 7863636d..c6845cc6 100644 --- a/src-docs/jenkins.py.md +++ b/src-docs/jenkins.py.md @@ -19,6 +19,8 @@ Functions to operate Jenkins. - **WAR_DOWNLOAD_URL** - **SYSTEM_PROPERTY_HEADLESS** - **SYSTEM_PROPERTY_LOGGING** +- **DEFAULT_JENKINS_CONFIG** +- **JENKINS_LOGGING_CONFIG** - **PLUGIN_NAME_GROUP** - **WHITESPACE** - **VERSION_GROUP** @@ -28,87 +30,87 @@ Functions to operate Jenkins. --- - + -## function `wait_ready` +## function `get_admin_credentials` ```python -wait_ready(timeout: int = 300, check_interval: int = 10) → None +get_admin_credentials(container: Container) → Credentials ``` -Wait until Jenkins service is up. +Retrieve admin credentials. **Args:** - - `timeout`: Time in seconds to wait for jenkins to become ready in 10 second intervals. - - `check_interval`: Time in seconds to wait between ready checks. + - `container`: The Jenkins workload container to interact with filesystem. -**Raises:** - - - `TimeoutError`: if Jenkins status check did not pass within the timeout duration. +**Returns:** + The Jenkins admin account credentials. --- - + -## function `is_storage_ready` +## function `wait_ready` ```python -is_storage_ready(container: Container) → bool +wait_ready(timeout: int = 300, check_interval: int = 10) → None ``` -Return whether the Jenkins home directory is mounted and owned by jenkins. +Wait until Jenkins service is up. **Args:** - - `container`: The Jenkins workload container. + - `timeout`: Time in seconds to wait for jenkins to become ready in 10 second intervals. + - `check_interval`: Time in seconds to wait between ready checks. **Raises:** - - `StorageMountError`: if there was an error getting storage information. - - - -**Returns:** - True if home directory is mounted and owned by jenkins, False otherwise. + - `TimeoutError`: if Jenkins status check did not pass within the timeout duration. --- - + -## function `get_admin_credentials` +## function `is_storage_ready` ```python -get_admin_credentials(container: Container) → Credentials +is_storage_ready(container: Optional[Container]) → bool ``` -Retrieve admin credentials. +Return whether the Jenkins home directory is mounted and owned by jenkins. **Args:** - - `container`: The Jenkins workload container to interact with filesystem. + - `container`: The Jenkins workload container. + + + +**Raises:** + + - `StorageMountError`: if there was an error getting storage information. **Returns:** - The Jenkins admin account credentials. + True if home directory is mounted and owned by jenkins, False otherwise. --- - + ## function `calculate_env` @@ -126,7 +128,7 @@ Return a dictionary for Jenkins Pebble layer. --- - + ## function `get_version` @@ -150,7 +152,7 @@ Get the Jenkins server version. --- - + ## function `bootstrap` @@ -176,7 +178,7 @@ Initialize and install Jenkins. --- - + ## function `get_node_secret` @@ -207,7 +209,7 @@ Get node secret from jenkins. --- - + ## function `add_agent_node` @@ -233,7 +235,7 @@ Add a Jenkins agent node. --- - + ## function `remove_agent_node` @@ -259,7 +261,7 @@ Remove a Jenkins agent node. --- - + ## function `safe_restart` @@ -284,7 +286,7 @@ Safely restart Jenkins server after all jobs are done executing. --- - + ## function `get_agent_name` @@ -308,7 +310,7 @@ Infer agent name from unit name. --- - + ## function `remove_unlisted_plugins` @@ -339,7 +341,7 @@ Remove plugins that are not in the list of desired plugins. --- - + ## function `rotate_credentials` @@ -416,15 +418,6 @@ Base exception for Jenkins errors. ---- - -## class `JenkinsNetworkError` -An error occurred communicating with the upstream Jenkins server. - - - - - --- ## class `JenkinsPluginError` @@ -434,33 +427,6 @@ An error occurred installing Jenkins plugin. ---- - -## class `JenkinsProxyError` -An error occurred configuring Jenkins proxy. - - - - - ---- - -## class `JenkinsRestartError` -An error occurred trying to restart Jenkins. - - - - - ---- - -## class `JenkinsUpdateError` -An error occurred trying to update Jenkins. - - - - - --- ## class `StorageMountError` @@ -472,7 +438,7 @@ Represents an error probing for Jenkins storage mount. - `msg`: Explanation of the error. - + ### function `__init__` diff --git a/src/actions.py b/src/actions.py index 505cf653..0b959c6f 100644 --- a/src/actions.py +++ b/src/actions.py @@ -36,7 +36,7 @@ def on_get_admin_password(self, event: ops.ActionEvent) -> None: event: The event fired from get-admin-password action. """ container = self.charm.unit.get_container(JENKINS_SERVICE_NAME) - if not container.can_connect() or not jenkins.is_storage_ready(container): + if not jenkins.is_storage_ready(container): event.fail("Service not yet ready.") return credentials = jenkins.get_admin_credentials(container) @@ -49,7 +49,7 @@ def on_rotate_credentials(self, event: ops.ActionEvent) -> None: event: The rotate credentials event. """ container = self.charm.unit.get_container(JENKINS_SERVICE_NAME) - if not container.can_connect() or not jenkins.is_storage_ready(container): + if not jenkins.is_storage_ready(container): event.fail("Service not yet ready.") return try: diff --git a/src/agent.py b/src/agent.py index c5a8e736..f4ba7df5 100644 --- a/src/agent.py +++ b/src/agent.py @@ -112,7 +112,7 @@ def _on_deprecated_agent_relation_joined(self, event: ops.RelationJoinedEvent) - event: The event fired from an agent joining the relationship. """ container = self.charm.unit.get_container(JENKINS_SERVICE_NAME) - if not container.can_connect() or not jenkins.is_storage_ready(container): + if not jenkins.is_storage_ready(container): logger.warning("Service not yet ready. Deferring.") event.defer() # The event needs to be handled after Jenkins has started(pebble ready). return @@ -149,7 +149,7 @@ def _on_agent_relation_joined(self, event: ops.RelationJoinedEvent) -> None: event: The event fired from an agent joining the relationship. """ container = self.charm.unit.get_container(JENKINS_SERVICE_NAME) - if not container.can_connect() or not jenkins.is_storage_ready(container): + if not jenkins.is_storage_ready(container): logger.warning("Service not yet ready. Deferring.") event.defer() # The event needs to be handled after Jenkins has started(pebble ready). return @@ -187,7 +187,7 @@ def _on_deprecated_agent_relation_departed(self, event: ops.RelationDepartedEven """ # the event unit cannot be None. container = self.charm.unit.get_container(JENKINS_SERVICE_NAME) - if not container.can_connect() or not jenkins.is_storage_ready(container): + if not jenkins.is_storage_ready(container): logger.warning("Relation departed before service ready.") return @@ -215,7 +215,7 @@ def _on_agent_relation_departed(self, event: ops.RelationDepartedEvent) -> None: """ # the event unit cannot be None. container = self.charm.unit.get_container(JENKINS_SERVICE_NAME) - if not container.can_connect() or not jenkins.is_storage_ready(container): + if not jenkins.is_storage_ready(container): logger.warning("Relation departed before service ready.") return diff --git a/src/charm.py b/src/charm.py index 241c925f..0a96e88a 100755 --- a/src/charm.py +++ b/src/charm.py @@ -61,8 +61,8 @@ def __init__(self, *args: typing.Any): self.agent_observer = agent.Observer( self, self.state, self.agent_discovery_ingress_observer ) - self.cos_observer = cos.Observer(self) self.ingress_observer = ingress.Observer(self, "ingress-observer", INGRESS_RELATION_NAME) + self.cos_observer = cos.Observer(self) self.framework.observe( self.on.jenkins_home_storage_attached, self._on_jenkins_home_storage_attached ) @@ -78,6 +78,7 @@ def _get_pebble_layer(self, jenkins_env: jenkins.Environment) -> ops.pebble.Laye Returns: The pebble layer defining Jenkins service layer. """ + env_dict = typing.cast(typing.Dict[str, str], jenkins_env) layer: LayerDict = { "summary": "jenkins layer", "description": "pebble config layer for jenkins", @@ -89,8 +90,7 @@ def _get_pebble_layer(self, jenkins_env: jenkins.Environment) -> ops.pebble.Laye f"-D{jenkins.SYSTEM_PROPERTY_LOGGING} " f"-jar {jenkins.EXECUTABLES_PATH}/jenkins.war", "startup": "enabled", - # TypedDict and Dict[str,str] are not compatible. - "environment": typing.cast(typing.Dict[str, str], jenkins_env), + "environment": env_dict, "user": jenkins.USER, "group": jenkins.GROUP, }, @@ -183,7 +183,7 @@ def _on_update_status(self, _: ops.UpdateStatusEvent) -> None: 2. Update Jenkins patch version if available and is within restart-time-range config value. """ container = self.unit.get_container(JENKINS_SERVICE_NAME) - if not container.can_connect() or not jenkins.is_storage_ready(container): + if not jenkins.is_storage_ready(container): self.unit.status = ops.WaitingStatus("Waiting for container/storage.") return diff --git a/src/jenkins.py b/src/jenkins.py index 2264c7bb..eff95bd0 100644 --- a/src/jenkins.py +++ b/src/jenkins.py @@ -15,6 +15,7 @@ from datetime import datetime, timedelta from pathlib import Path from time import sleep +from typing import Optional import jenkinsapi.custom_exceptions import jenkinsapi.jenkins @@ -49,37 +50,31 @@ LOGGING_CONFIG_PATH = JENKINS_HOME_PATH / "logging.properties" # The Jenkins logging path as defined in templates/logging.properties file LOGGING_PATH = JENKINS_HOME_PATH / "jenkins.log" - # The plugins that are required for Jenkins to work REQUIRED_PLUGINS = [ "instance-identity", # required to connect agent nodes to server "prometheus", # required for COS integration "monitoring", # required for session invalidation ] - USER = "jenkins" GROUP = "jenkins" - BUILT_IN_NODE_NAME = "Built-In Node" # The Jenkins stable version RSS feed URL RSS_FEED_URL = "https://www.jenkins.io/changelog-stable/rss.xml" # The Jenkins WAR downloads page WAR_DOWNLOAD_URL = "https://updates.jenkins.io/download/war" - # Java system property to run Jenkins in headless mode SYSTEM_PROPERTY_HEADLESS = "java.awt.headless=true" # Java system property to load logging configuration from file SYSTEM_PROPERTY_LOGGING = f"java.util.logging.config.file={LOGGING_CONFIG_PATH}" +DEFAULT_JENKINS_CONFIG = "templates/jenkins-config.xml" +JENKINS_LOGGING_CONFIG = "templates/logging.properties" class JenkinsError(Exception): """Base exception for Jenkins errors.""" -class JenkinsProxyError(JenkinsError): - """An error occurred configuring Jenkins proxy.""" - - class JenkinsPluginError(JenkinsError): """An error occurred installing Jenkins plugin.""" @@ -92,16 +87,61 @@ class ValidationError(Exception): """An unexpected data is encountered.""" -class JenkinsNetworkError(JenkinsError): - """An error occurred communicating with the upstream Jenkins server.""" +class Environment(typing.TypedDict): + """Dictionary mapping of Jenkins environment variables. + + Attributes: + JENKINS_HOME: The Jenkins home directory. + """ + + JENKINS_HOME: str + + +@dataclasses.dataclass(frozen=True) +class Credentials: + """Information needed to log into Jenkins. + + Attributes: + username: The Jenkins account username used to log into Jenkins. + password_or_token: The Jenkins API token or account password used to log into Jenkins. + """ + + username: str + password_or_token: str + + +def get_admin_credentials(container: ops.Container) -> Credentials: + """Retrieve admin credentials. + Args: + container: The Jenkins workload container to interact with filesystem. -class JenkinsUpdateError(JenkinsError): - """An error occurred trying to update Jenkins.""" + Returns: + The Jenkins admin account credentials. + """ + user = "admin" + password_file_contents = str(container.pull(PASSWORD_FILE_PATH, encoding="utf-8").read()) + return Credentials(username=user, password_or_token=password_file_contents.strip()) -class JenkinsRestartError(JenkinsError): - """An error occurred trying to restart Jenkins.""" +def _get_api_credentials(container: ops.Container) -> Credentials: + """Retrieve admin API credentials. + + Args: + container: The Jenkins workload container. + + Returns: + Credentials: The Jenkins API Credentials. + + Raises: + JenkinsBootstrapError: if no API credential has been setup yet. + """ + try: + token = str(container.pull(API_TOKEN_PATH, encoding="utf-8").read()) + return Credentials(username="admin", password_or_token=token.strip()) + except ops.pebble.PathError as exc: + logger.debug("Admin API token not yet setup.") + raise JenkinsBootstrapError("Admin API credentials not yet setup.") from exc def _wait_for( @@ -174,7 +214,7 @@ def __init__(self, msg: str): self.msg = msg -def is_storage_ready(container: ops.Container) -> bool: +def is_storage_ready(container: Optional[ops.Container]) -> bool: """Return whether the Jenkins home directory is mounted and owned by jenkins. Args: @@ -186,6 +226,8 @@ def is_storage_ready(container: ops.Container) -> bool: Returns: True if home directory is mounted and owned by jenkins, False otherwise. """ + if not container or not container.can_connect(): + return False mount_info: str = container.pull("/proc/mounts").read() if str(JENKINS_HOME_PATH) not in mount_info: return False @@ -197,63 +239,6 @@ def is_storage_ready(container: ops.Container) -> bool: return "jenkins" in stdout -@dataclasses.dataclass(frozen=True) -class Credentials: - """Information needed to log into Jenkins. - - Attributes: - username: The Jenkins account username used to log into Jenkins. - password_or_token: The Jenkins API token or account password used to log into Jenkins. - """ - - username: str - password_or_token: str - - -def get_admin_credentials(container: ops.Container) -> Credentials: - """Retrieve admin credentials. - - Args: - container: The Jenkins workload container to interact with filesystem. - - Returns: - The Jenkins admin account credentials. - """ - user = "admin" - password_file_contents = str(container.pull(PASSWORD_FILE_PATH, encoding="utf-8").read()) - return Credentials(username=user, password_or_token=password_file_contents.strip()) - - -def _get_api_credentials(container: ops.Container) -> Credentials: - """Retrieve admin API credentials. - - Args: - container: The Jenkins workload container. - - Returns: - Credentials: The Jenkins API Credentials. - - Raises: - JenkinsBootstrapError: if no API credential has been setup yet. - """ - try: - token = str(container.pull(API_TOKEN_PATH, encoding="utf-8").read()) - return Credentials(username="admin", password_or_token=token.strip()) - except ops.pebble.PathError as exc: - logger.debug("Admin API token not yet setup.") - raise JenkinsBootstrapError("Admin API credentials not yet setup.") from exc - - -class Environment(typing.TypedDict): - """Dictionary mapping of Jenkins environment variables. - - Attributes: - JENKINS_HOME: The Jenkins home directory. - """ - - JENKINS_HOME: str - - def calculate_env() -> Environment: """Return a dictionary for Jenkins Pebble layer. @@ -313,9 +298,9 @@ def _install_configs(container: ops.Container) -> None: Args: container: The Jenkins workload container. """ - with open("templates/jenkins-config.xml", encoding="utf-8") as jenkins_config_file: + with open(DEFAULT_JENKINS_CONFIG, encoding="utf-8") as jenkins_config_file: container.push(CONFIG_FILE_PATH, jenkins_config_file, user=USER, group=GROUP) - with open("templates/logging.properties", encoding="utf-8") as jenkins_logging_config_file: + with open(JENKINS_LOGGING_CONFIG, encoding="utf-8") as jenkins_logging_config_file: container.push(LOGGING_CONFIG_PATH, jenkins_logging_config_file, user=USER, group=GROUP) @@ -366,7 +351,7 @@ def _configure_proxy( proxy_config: The proxy settings to apply. Raises: - JenkinsProxyError: if an error occurred running proxy configuration script. + JenkinsBootstrapError: if an error occurred running proxy configuration script. """ if not proxy_config: return @@ -377,7 +362,7 @@ def _configure_proxy( client.run_groovy_script(f"proxy = new ProxyConfiguration({parsed_args})\nproxy.save()") except jenkinsapi.custom_exceptions.JenkinsAPIException as exc: logger.error("Failed to configure proxy, %s", exc) - raise JenkinsProxyError("Proxy configuration failed.") from exc + raise JenkinsBootstrapError("Proxy configuration failed.") from exc def _get_java_proxy_args(proxy_config: state.ProxyConfig) -> typing.Iterable[str]: @@ -464,7 +449,7 @@ def bootstrap(container: ops.Container, proxy_config: state.ProxyConfig | None = try: _configure_proxy(container, proxy_config) _install_plugins(container, proxy_config) - except (JenkinsProxyError, JenkinsPluginError) as exc: + except (JenkinsBootstrapError, JenkinsPluginError) as exc: raise JenkinsBootstrapError("Failed to bootstrap Jenkins.") from exc diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 68c4e83a..2427894f 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -276,7 +276,7 @@ async def jenkins_machine_agents_fixture( # stable. app: Application = await machine_model.deploy( "jenkins-agent", - channel="latest/edge", + channel="latest/stable", config={"jenkins_agent_labels": "machine"}, application_name=f"jenkins-agent-{app_suffix}", num_units=num_units, diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index d979304d..44295e8b 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -4,11 +4,11 @@ """Fixtures for Jenkins-k8s-operator charm unit tests.""" import textwrap -import typing -import unittest.mock from ipaddress import IPv4Address from pathlib import Path from secrets import token_hex +from typing import Any, Callable, Tuple, cast +from unittest.mock import MagicMock import jenkinsapi.jenkins import pytest @@ -50,7 +50,7 @@ def jenkins_version_fixture(): def mocked_get_request_fixture(jenkins_version: str): """Mock get request with given status code.""" - def mocked_get(_: str, status_code: int = 200, **_kwargs: typing.Any): + def mocked_get(_: str, status_code: int = 200, **_kwargs: Any): """Mock get request with predefined status code. Args: @@ -74,9 +74,9 @@ def admin_credentials_fixture() -> jenkins.Credentials: @pytest.fixture(scope="function", name="mock_client") -def mock_client_fixture(monkeypatch: pytest.MonkeyPatch) -> unittest.mock.MagicMock: +def mock_client_fixture(monkeypatch: pytest.MonkeyPatch) -> MagicMock: """Mock Jenkins API client.""" - mock_client = unittest.mock.MagicMock(spec=jenkinsapi.jenkins.Jenkins) + mock_client = MagicMock(spec=jenkinsapi.jenkins.Jenkins) monkeypatch.setattr(jenkins, "_get_client", lambda *_args, **_kwargs: mock_client) return mock_client @@ -88,7 +88,7 @@ def inject_register_command_handler(monkeypatch: pytest.MonkeyPatch, harness: Ha monkeypatch: The pytest monkeypatch object. harness: The testing harness. """ - handler_table: dict[str, typing.Callable[[list[str]], tuple[int, str, str]]] = {} + handler_table: dict[str, Callable[[list[str]], tuple[int, str, str]]] = {} # This is a stub implementation only. class ExecProcessStub: # pylint: disable=too-few-public-methods @@ -126,7 +126,7 @@ def wait_output(self): stderr=self._stderr, ) - def exec_stub(command: list[str], **_kwargs: typing.Any): + def exec_stub(command: list[str], **_kwargs: Any): """A mock implementation of the `exec` method of the container object. Args: @@ -143,7 +143,7 @@ def exec_stub(command: list[str], **_kwargs: typing.Any): def register_command_handler( container: Container | str, executable: str, - handler=typing.Callable[[list[str]], typing.Tuple[int, str, str]], + handler=Callable[[list[str]], Tuple[int, str, str]], ): """Registers a handler for a specific executable command. @@ -200,7 +200,7 @@ def cmd_handler(argv: list[str]) -> tuple[int, str, str]: """ required_plugins = " ".join(set(jenkins.REQUIRED_PLUGINS)) # type cast since the fixture contains no_proxy values - no_proxy_hosts = "|".join(typing.cast(str, proxy_config.no_proxy).split(",")) + no_proxy_hosts = "|".join(cast(str, proxy_config.no_proxy).split(",")) # assert for types that cannot be None. assert proxy_config.http_proxy, "Http proxy fixture should not be None." assert proxy_config.https_proxy, "Https proxy fixture should not be None." @@ -439,7 +439,7 @@ def plugin_groovy_script_result_fixture(): @pytest.fixture(scope="module", name="mock_charm") def mock_charm_fixture(): """A valid mock charm.""" - mock_charm = unittest.mock.MagicMock(spec=CharmBase) + mock_charm = MagicMock(spec=CharmBase) mock_charm.app.planned_units.return_value = 1 return mock_charm @@ -455,7 +455,7 @@ def patch_os_environ_fixture(monkeypatch: pytest.MonkeyPatch): @pytest.fixture(scope="function", name="patch_jenkins_node") def patch_jenkins_node_fixture(monkeypatch: pytest.MonkeyPatch): """Monkeypatch jenkinsapi Node to enable node creation.""" - mock_node = unittest.mock.MagicMock(spec=jenkinsapi.node.Node) + mock_node = MagicMock(spec=jenkinsapi.node.Node) mock_node.return_value.get_node_attributes.return_value = { "json": '{"launcher": {"tunnel": ""}}' } @@ -465,4 +465,4 @@ def patch_jenkins_node_fixture(monkeypatch: pytest.MonkeyPatch): @pytest.fixture(scope="function", name="mock_ip_addr") def mock_ip_addr_fixture(): """Mock IPV4 fixture.""" - return unittest.mock.MagicMock(spec=IPv4Address) + return MagicMock(spec=IPv4Address) diff --git a/tests/unit/test_actions.py b/tests/unit/test_actions.py index d5c16198..6c7ba7e3 100644 --- a/tests/unit/test_actions.py +++ b/tests/unit/test_actions.py @@ -7,7 +7,7 @@ # Need access to protected functions for testing import typing -import unittest.mock +from unittest.mock import MagicMock import ops import pytest @@ -36,8 +36,8 @@ def test_workload_not_ready(harness: Harness, event_handler: str): harness.begin() jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) handler_func = getattr(jenkins_charm.actions_observer, event_handler) - mock_event = unittest.mock.MagicMock(spec=ops.ActionEvent) - mock_event.workload = unittest.mock.MagicMock(spec=ops.model.Container) + mock_event = MagicMock(spec=ops.ActionEvent) + mock_event.workload = MagicMock(spec=ops.model.Container) mock_event.workload.can_connect.return_value = False handler_func(mock_event) @@ -53,7 +53,7 @@ def test_on_get_admin_password_action( act: when get-admin-password action is run. assert: the correct admin password is returned. """ - mock_event = unittest.mock.MagicMock(spec=ops.ActionEvent) + mock_event = MagicMock(spec=ops.ActionEvent) harness_container.harness.begin() jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness_container.harness.charm) @@ -76,15 +76,14 @@ def test_on_rotate_credentials_action_api_error( monkeypatch.setattr( charm.actions.jenkins, "rotate_credentials", - unittest.mock.MagicMock(side_effect=jenkins.JenkinsError), + MagicMock(side_effect=jenkins.JenkinsError), ) - harness_container.harness.set_can_connect( - harness_container.harness.model.unit.containers["jenkins"], True - ) - mock_event = unittest.mock.MagicMock(spec=ops.ActionEvent) - harness_container.harness.begin() + harness = harness_container.harness + harness.set_can_connect(harness_container.harness.model.unit.containers["jenkins"], True) + mock_event = MagicMock(spec=ops.ActionEvent) + harness.begin() - jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness_container.harness.charm) + jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) jenkins_charm.actions_observer.on_rotate_credentials(mock_event) mock_event.fail.assert_called_once() @@ -104,13 +103,12 @@ def test_on_rotate_credentials_action( "rotate_credentials", lambda *_args, **_kwargs: password, ) - harness_container.harness.set_can_connect( - harness_container.harness.model.unit.containers["jenkins"], True - ) - mock_event = unittest.mock.MagicMock(spec=ops.ActionEvent) - harness_container.harness.begin() + harness = harness_container.harness + harness.set_can_connect(harness_container.harness.model.unit.containers["jenkins"], True) + mock_event = MagicMock(spec=ops.ActionEvent) + harness.begin() - jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness_container.harness.charm) + jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) jenkins_charm.actions_observer.on_rotate_credentials(mock_event) mock_event.set_results.assert_called_once_with({"password": password}) diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py index dd110c04..c9601c2e 100644 --- a/tests/unit/test_charm.py +++ b/tests/unit/test_charm.py @@ -9,7 +9,7 @@ import datetime import functools import typing -import unittest.mock +from unittest.mock import MagicMock import ops import pytest @@ -58,8 +58,8 @@ def test_workload_not_ready(harness: Harness, event_handler: str): harness.begin() jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) handler_func = getattr(jenkins_charm, event_handler) - mock_event = unittest.mock.MagicMock(spec=ops.WorkloadEvent) - mock_event.workload = unittest.mock.MagicMock(spec=ops.model.Container) + mock_event = MagicMock(spec=ops.WorkloadEvent) + mock_event.workload = MagicMock(spec=ops.model.Container) mock_event.workload.can_connect.return_value = False handler_func(mock_event) @@ -84,8 +84,8 @@ def test_storage_not_ready(harness: Harness, event_handler: str): harness.begin() jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) handler_func = getattr(jenkins_charm, event_handler) - mock_event = unittest.mock.MagicMock(spec=ops.WorkloadEvent) - mock_event.workload = unittest.mock.MagicMock(spec=ops.model.Container) + mock_event = MagicMock(spec=ops.WorkloadEvent) + mock_event.workload = MagicMock(spec=ops.model.Container) mock_event.workload.can_connect.return_value = True handler_func(mock_event) @@ -109,14 +109,14 @@ def test__on_jenkins_pebble_ready_error( monkeypatch.setattr( jenkins, "bootstrap", - unittest.mock.MagicMock(side_effect=jenkins.JenkinsBootstrapError), + MagicMock(side_effect=jenkins.JenkinsBootstrapError), ) monkeypatch.setattr(requests, "get", functools.partial(mocked_get_request, status_code=200)) harness.begin() jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) with pytest.raises(jenkins.JenkinsBootstrapError): - jenkins_charm._on_jenkins_pebble_ready(unittest.mock.MagicMock(spec=ops.PebbleReadyEvent)) + jenkins_charm._on_jenkins_pebble_ready(MagicMock(spec=ops.PebbleReadyEvent)) def test__on_jenkins_pebble_ready_get_version_error( @@ -131,9 +131,7 @@ def test__on_jenkins_pebble_ready_get_version_error( """ harness = harness_container.harness # speed up waiting by changing default argument values - monkeypatch.setattr( - jenkins, "get_version", unittest.mock.MagicMock(side_effect=jenkins.JenkinsError) - ) + monkeypatch.setattr(jenkins, "get_version", MagicMock(side_effect=jenkins.JenkinsError)) monkeypatch.setattr(jenkins.wait_ready, "__defaults__", (1, 1)) monkeypatch.setattr(jenkins, "bootstrap", lambda *_args: None) monkeypatch.setattr(requests, "get", functools.partial(mocked_get_request, status_code=200)) @@ -142,7 +140,7 @@ def test__on_jenkins_pebble_ready_get_version_error( jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) with pytest.raises(jenkins.JenkinsError): - jenkins_charm._on_jenkins_pebble_ready(unittest.mock.MagicMock(spec=ops.PebbleReadyEvent)) + jenkins_charm._on_jenkins_pebble_ready(MagicMock(spec=ops.PebbleReadyEvent)) @pytest.mark.usefixtures("patch_os_environ") @@ -155,9 +153,7 @@ def test__on_jenkins_pebble_jenkins_not_ready( assert: the charm raises an error. """ harness = harness_container.harness - monkeypatch.setattr( - jenkins, "wait_ready", unittest.mock.MagicMock(side_effect=[None, TimeoutError]) - ) + monkeypatch.setattr(jenkins, "wait_ready", MagicMock(side_effect=[None, TimeoutError])) monkeypatch.setattr(jenkins, "bootstrap", lambda *_args, **_kwargs: None) monkeypatch.setattr(jenkins, "get_version", lambda *_args, **_kwargs: "1") harness.begin() @@ -165,7 +161,7 @@ def test__on_jenkins_pebble_jenkins_not_ready( jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) with pytest.raises(TimeoutError): - jenkins_charm._on_jenkins_pebble_ready(unittest.mock.MagicMock(spec=ops.PebbleReadyEvent)) + jenkins_charm._on_jenkins_pebble_ready(MagicMock(spec=ops.PebbleReadyEvent)) @pytest.mark.usefixtures("patch_os_environ") @@ -178,13 +174,13 @@ def test__on_jenkins_pebble_ready( assert: the unit status should show expected status and the jenkins port should be open. """ harness = harness_container.harness - monkeypatch.setattr(jenkins, "wait_ready", unittest.mock.MagicMock(return_value=None)) + monkeypatch.setattr(jenkins, "wait_ready", MagicMock(return_value=None)) monkeypatch.setattr(jenkins, "bootstrap", lambda *_args, **_kwargs: None) monkeypatch.setattr(jenkins, "get_version", lambda *_args, **_kwargs: "1") harness.begin() jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) - jenkins_charm._on_jenkins_pebble_ready(unittest.mock.MagicMock(spec=ops.PebbleReadyEvent)) + jenkins_charm._on_jenkins_pebble_ready(MagicMock(spec=ops.PebbleReadyEvent)) jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness.charm) assert ( @@ -226,7 +222,7 @@ def test__remove_unlisted_plugins_error( monkeypatch.setattr( jenkins, "remove_unlisted_plugins", - unittest.mock.MagicMock(side_effect=exception), + MagicMock(side_effect=exception), ) harness_container.harness.begin() @@ -263,9 +259,9 @@ def test__on_update_status_not_in_time_range( act: when update_status action is triggered. assert: no further functions are called. """ - mock_datetime = unittest.mock.MagicMock(spec=datetime.datetime) + mock_datetime = MagicMock(spec=datetime.datetime) mock_datetime.utcnow.return_value = datetime.datetime(2023, 1, 1, 23) - mock_remove_unlisted_plugins_func = unittest.mock.MagicMock( + mock_remove_unlisted_plugins_func = MagicMock( spec=JenkinsK8sOperatorCharm._remove_unlisted_plugins ) monkeypatch.setattr(timerange, "datetime", mock_datetime) @@ -278,7 +274,7 @@ def test__on_update_status_not_in_time_range( monkeypatch.setattr( jenkins_charm, "_remove_unlisted_plugins", mock_remove_unlisted_plugins_func ) - jenkins_charm._on_update_status(unittest.mock.MagicMock(spec=ops.UpdateStatusEvent)) + jenkins_charm._on_update_status(MagicMock(spec=ops.UpdateStatusEvent)) assert jenkins_charm.unit.status.name == original_status mock_remove_unlisted_plugins_func.assert_not_called() @@ -340,7 +336,7 @@ def test__on_update_status( harness_container.harness.begin() jenkins_charm = typing.cast(JenkinsK8sOperatorCharm, harness_container.harness.charm) - jenkins_charm._on_update_status(unittest.mock.MagicMock(spec=ops.UpdateStatusEvent)) + jenkins_charm._on_update_status(MagicMock(spec=ops.UpdateStatusEvent)) assert jenkins_charm.unit.status == expected_status @@ -357,10 +353,10 @@ def test__on_jenkins_home_storage_attached(harness: Harness, monkeypatch: pytest harness.set_can_connect(container, True) # We don't use harness.handle_exec here because we want to assert # the parameters passed to exec() - exec_handler = unittest.mock.MagicMock() + exec_handler = MagicMock() monkeypatch.setattr(container, "exec", exec_handler) - event = unittest.mock.MagicMock() + event = MagicMock() mock_jenkins_home_path = "/var/lib/jenkins" event.storage.location.resolve = lambda: mock_jenkins_home_path jenkins_charm._on_jenkins_home_storage_attached(event) @@ -384,9 +380,9 @@ def test__on_jenkins_home_storage_attached_container_not_ready( harness.set_can_connect(container, False) # We don't use harness.handle_exec here because we want to assert # whether exec() was called - exec_handler = unittest.mock.MagicMock() + exec_handler = MagicMock() monkeypatch.setattr(container, "exec", exec_handler) - jenkins_charm._on_jenkins_home_storage_attached(unittest.mock.MagicMock()) + jenkins_charm._on_jenkins_home_storage_attached(MagicMock()) exec_handler.assert_not_called() diff --git a/tests/unit/test_jenkins.py b/tests/unit/test_jenkins.py index d94d579e..e5656521 100644 --- a/tests/unit/test_jenkins.py +++ b/tests/unit/test_jenkins.py @@ -13,9 +13,8 @@ import secrets import textwrap import typing -import unittest.mock from functools import partial -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import jenkinsapi.jenkins import ops @@ -395,13 +394,13 @@ def test__configure_proxy_fail( """ arrange: given a test proxy config and a monkeypatched jenkins client that raises an exception. act: when _configure_proxy is called. - assert: JenkinsProxyError is raised. + assert: JenkinsBootstrapError is raised. """ mock_client.run_groovy_script = MagicMock( side_effect=jenkinsapi.custom_exceptions.JenkinsAPIException ) - with pytest.raises(jenkins.JenkinsProxyError) as exc: + with pytest.raises(jenkins.JenkinsBootstrapError) as exc: jenkins._configure_proxy(harness_container.container, proxy_config) assert exc.value.args[0] == "Proxy configuration failed." @@ -544,7 +543,7 @@ def test_get_client(admin_credentials: jenkins.Credentials): """ expected_client = MagicMock(spec=jenkinsapi.jenkins.Jenkins) - with unittest.mock.patch("jenkinsapi.jenkins.Jenkins", return_value=expected_client): + with patch("jenkinsapi.jenkins.Jenkins", return_value=expected_client): client = jenkins._get_client(admin_credentials) assert client == expected_client diff --git a/tests/unit/test_timerange.py b/tests/unit/test_timerange.py index 4132d764..33c72db3 100644 --- a/tests/unit/test_timerange.py +++ b/tests/unit/test_timerange.py @@ -3,7 +3,7 @@ """Jenkins-k8s time range module tests.""" -import unittest.mock +from unittest.mock import MagicMock import pytest @@ -79,7 +79,7 @@ def test_restart_time_range_check_now( act: when check_now is called. assert: expected value returning whether now is within time range is returned. """ - mock_datetime = unittest.mock.MagicMock(spec=timerange.datetime) + mock_datetime = MagicMock(spec=timerange.datetime) test_time = timerange.datetime(2023, 1, 1, patch_hour) mock_datetime.utcnow.return_value = test_time monkeypatch.setattr(timerange, "datetime", mock_datetime)