diff --git a/cmk/notification_plugins/utils.py b/cmk/notification_plugins/utils.py index d1bd9faf7fb..3692270c4de 100644 --- a/cmk/notification_plugins/utils.py +++ b/cmk/notification_plugins/utils.py @@ -18,7 +18,6 @@ import cmk.utils.paths from cmk.utils.escaping import escape, escape_permissive from cmk.utils.http_proxy_config import deserialize_http_proxy_config -from cmk.utils.misc import typeshed_issue_7724 from cmk.utils.notify import find_wato_folder, NotificationContext from cmk.utils.notify_types import PluginNotificationContext @@ -255,9 +254,7 @@ def post_request( response = requests.post( # nosec B113 url=url, json=message_constructor(context), - proxies=typeshed_issue_7724( - deserialize_http_proxy_config(serialized_proxy_config).to_requests_proxies() - ), + proxies=deserialize_http_proxy_config(serialized_proxy_config).to_requests_proxies(), headers=headers, verify=verify, ) diff --git a/cmk/special_agents/agent_datadog.py b/cmk/special_agents/agent_datadog.py index 83133c5d17a..79bac4ae461 100644 --- a/cmk/special_agents/agent_datadog.py +++ b/cmk/special_agents/agent_datadog.py @@ -27,7 +27,6 @@ from cmk.utils import paths, store from cmk.utils.http_proxy_config import deserialize_http_proxy_config -from cmk.utils.misc import typeshed_issue_7724 from cmk.ec.export import ( # pylint: disable=cmk-module-layer-violation SyslogForwarderUnixSocket, @@ -253,7 +252,7 @@ def get_request( f"{self._api_url}/{version}/{api_endpoint}", headers=self._query_heads, params=params, - proxies=typeshed_issue_7724(self._proxy.to_requests_proxies()), + proxies=self._proxy.to_requests_proxies(), ) def post_request( @@ -266,7 +265,7 @@ def post_request( f"{self._api_url}/{version}/{api_endpoint}", headers=self._query_heads, json=body, - proxies=typeshed_issue_7724(self._proxy.to_requests_proxies()), + proxies=self._proxy.to_requests_proxies(), ) diff --git a/cmk/special_agents/agent_fritzbox.py b/cmk/special_agents/agent_fritzbox.py index f018487776d..464fe21b174 100644 --- a/cmk/special_agents/agent_fritzbox.py +++ b/cmk/special_agents/agent_fritzbox.py @@ -37,8 +37,6 @@ import requests -from cmk.utils.misc import typeshed_issue_7724 - from cmk.special_agents.utils import vcrtrace UPNPInfo = tuple[Mapping[str, str], str, str] @@ -131,7 +129,7 @@ def post(self, url: str, data: str, headers: Mapping[str, str]) -> requests.Resp return self._session.post( f"{self._urls[self._urlidx]}{url}", data=data, - headers=typeshed_issue_7724(headers), + headers=headers, timeout=self._timeout, ) diff --git a/cmk/special_agents/agent_mobileiron.py b/cmk/special_agents/agent_mobileiron.py index ee3c34c9dda..688def2dd69 100644 --- a/cmk/special_agents/agent_mobileiron.py +++ b/cmk/special_agents/agent_mobileiron.py @@ -24,7 +24,6 @@ import requests from cmk.utils.http_proxy_config import deserialize_http_proxy_config -from cmk.utils.misc import typeshed_issue_7724 from cmk.utils.regex import regex, REGEX_HOST_NAME_CHARS from cmk.special_agents.utils.agent_common import ( @@ -146,7 +145,7 @@ def __init__( self._devices_per_request = 200 self.regex_patterns = regex_patterns self._proxy = deserialize_http_proxy_config(proxy) - if _requests_proxy := typeshed_issue_7724(self._proxy.to_requests_proxies()): + if _requests_proxy := self._proxy.to_requests_proxies(): self._session.proxies = _requests_proxy def __enter__(self) -> MobileironAPI: diff --git a/cmk/utils/http_proxy_config.py b/cmk/utils/http_proxy_config.py index 5866f487684..0283acb62fd 100644 --- a/cmk/utils/http_proxy_config.py +++ b/cmk/utils/http_proxy_config.py @@ -3,13 +3,12 @@ # This file is part of Checkmk (https://checkmk.com). It is subject to the terms and # conditions defined in the file COPYING, which is part of this source code package. - -from collections.abc import Mapping +from collections.abc import Mapping, MutableMapping from typing import Protocol class HTTPProxyConfig(Protocol): - def to_requests_proxies(self) -> Mapping[str, str] | None: + def to_requests_proxies(self) -> MutableMapping[str, str] | None: ... def serialize(self) -> str: @@ -36,7 +35,7 @@ def __eq__(self, o: object) -> bool: class NoProxyConfig: SERIALIZED = "NO_PROXY" - def to_requests_proxies(self) -> Mapping[str, str]: + def to_requests_proxies(self) -> dict[str, str]: return { "http": "", "https": "", @@ -53,7 +52,7 @@ class ExplicitProxyConfig: def __init__(self, url: str) -> None: self._url = url - def to_requests_proxies(self) -> Mapping[str, str]: + def to_requests_proxies(self) -> dict[str, str]: return { "http": self._url, "https": self._url, diff --git a/cmk/utils/misc.py b/cmk/utils/misc.py index e52a3b0efa9..7ed6e7720eb 100644 --- a/cmk/utils/misc.py +++ b/cmk/utils/misc.py @@ -10,7 +10,7 @@ import itertools import sys import time -from collections.abc import Callable, Iterator, Mapping, MutableMapping +from collections.abc import Callable, Iterator from pathlib import Path from typing import Any @@ -94,9 +94,3 @@ def cachefile_age(path: Path | str) -> float: path = Path(path) return time.time() - path.stat().st_mtime - - -def typeshed_issue_7724(x: Mapping[str, str] | None) -> MutableMapping[str, str] | None: - """Temporary workaround for https://github.com/python/typeshed/issues/7724 - TODO: Remove this when the issue a above is fixed!""" - return None if x is None else dict(x) diff --git a/tests/unit/agent_receiver/test_endpoints.py b/tests/unit/agent_receiver/test_endpoints.py index a557088ef5e..77f71d262d4 100644 --- a/tests/unit/agent_receiver/test_endpoints.py +++ b/tests/unit/agent_receiver/test_endpoints.py @@ -5,7 +5,7 @@ import io import stat -from collections.abc import Mapping +from collections.abc import MutableMapping from pathlib import Path from uuid import uuid4 from zlib import compress @@ -24,8 +24,6 @@ from tests.testlib.certs import generate_csr_pair -from cmk.utils.misc import typeshed_issue_7724 - @pytest.fixture(name="symlink_push_host") def fixture_symlink_push_host( @@ -556,7 +554,7 @@ def test_register_new_ongoing_success( @pytest.fixture(name="agent_data_headers") -def fixture_agent_data_headers(uuid: UUID4) -> Mapping[str, str]: +def fixture_agent_data_headers(uuid: UUID4) -> dict[str, str]: return { "compression": "zlib", "verified-uuid": str(uuid), @@ -571,12 +569,12 @@ def fixture_compressed_agent_data() -> io.BytesIO: def test_agent_data_uuid_mismatch( client: TestClient, uuid: UUID4, - agent_data_headers: Mapping[str, str], + agent_data_headers: MutableMapping[str, str], compressed_agent_data: io.BytesIO, ) -> None: response = client.post( "/agent_data/123", - headers=typeshed_issue_7724(agent_data_headers), + headers=dict(agent_data_headers), files={"monitoring_data": ("filename", compressed_agent_data)}, ) assert response.status_code == 400 @@ -588,12 +586,12 @@ def test_agent_data_uuid_mismatch( def test_agent_data_no_host( client: TestClient, uuid: UUID4, - agent_data_headers: Mapping[str, str], + agent_data_headers: MutableMapping[str, str], compressed_agent_data: io.BytesIO, ) -> None: response = client.post( f"/agent_data/{uuid}", - headers=typeshed_issue_7724(agent_data_headers), + headers=agent_data_headers, files={"monitoring_data": ("filename", compressed_agent_data)}, ) assert response.status_code == 403 @@ -604,7 +602,7 @@ def test_agent_data_pull_host( tmp_path: Path, client: TestClient, uuid: UUID4, - agent_data_headers: Mapping[str, str], + agent_data_headers: MutableMapping[str, str], compressed_agent_data: io.BytesIO, ) -> None: source = site_context.agent_output_dir() / str(uuid) @@ -612,7 +610,7 @@ def test_agent_data_pull_host( response = client.post( f"/agent_data/{uuid}", - headers=typeshed_issue_7724(agent_data_headers), + headers=agent_data_headers, files={ "monitoring_data": ( "filename", @@ -628,7 +626,7 @@ def test_agent_data_pull_host( def test_agent_data_invalid_compression( client: TestClient, uuid: UUID4, - agent_data_headers: Mapping[str, str], + agent_data_headers: MutableMapping[str, str], ) -> None: response = client.post( f"/agent_data/{uuid}", @@ -646,11 +644,11 @@ def test_agent_data_invalid_compression( def test_agent_data_invalid_data( client: TestClient, uuid: UUID4, - agent_data_headers: Mapping[str, str], + agent_data_headers: MutableMapping[str, str], ) -> None: response = client.post( f"/agent_data/{uuid}", - headers=typeshed_issue_7724(agent_data_headers), + headers=agent_data_headers, files={"monitoring_data": ("filename", io.BytesIO(b"certainly invalid"))}, ) assert response.status_code == 400 @@ -662,12 +660,12 @@ def test_agent_data_success( tmp_path: Path, client: TestClient, uuid: UUID4, - agent_data_headers: Mapping[str, str], + agent_data_headers: MutableMapping[str, str], compressed_agent_data: io.BytesIO, ) -> None: response = client.post( f"/agent_data/{uuid}", - headers=typeshed_issue_7724(agent_data_headers), + headers=agent_data_headers, files={"monitoring_data": ("filename", compressed_agent_data)}, ) @@ -678,7 +676,7 @@ def test_agent_data_success( @pytest.fixture(name="registration_status_headers") -def fixture_registration_status_headers(uuid: UUID4) -> Mapping[str, str]: +def fixture_registration_status_headers(uuid: UUID4) -> dict[str, str]: return { "verified-uuid": str(uuid), } @@ -687,11 +685,11 @@ def fixture_registration_status_headers(uuid: UUID4) -> Mapping[str, str]: def test_registration_status_uuid_mismtach( client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: response = client.get( "/registration_status/123", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 400 @@ -703,7 +701,7 @@ def test_registration_status_uuid_mismtach( def test_registration_status_declined( client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: R4R( status=R4RStatus.DECLINED, @@ -722,7 +720,7 @@ def test_registration_status_declined( response = client.get( f"/registration_status/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 200 @@ -738,11 +736,11 @@ def test_registration_status_declined( def test_registration_status_host_not_registered( client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: response = client.get( f"/registration_status/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 404 @@ -753,7 +751,7 @@ def test_registration_status_host_not_registered( def test_registration_status_push_host( client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: R4R( status=R4RStatus.DISCOVERABLE, @@ -767,7 +765,7 @@ def test_registration_status_push_host( response = client.get( f"/registration_status/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 200 @@ -784,7 +782,7 @@ def test_registration_status_pull_host( tmp_path: Path, client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: source = site_context.agent_output_dir() / str(uuid) target_dir = tmp_path / "hostname" @@ -792,7 +790,7 @@ def test_registration_status_pull_host( response = client.get( f"/registration_status/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 200 @@ -808,11 +806,11 @@ def test_registration_status_pull_host( def test_registration_status_v2_uuid_mismtach( client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: response = client.get( "/registration_status_v2/123", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 400 @@ -824,11 +822,11 @@ def test_registration_status_v2_uuid_mismtach( def test_registration_status_v2_host_not_registered( client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: response = client.get( f"/registration_status_v2/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 200 @@ -839,11 +837,11 @@ def test_registration_status_v2_host_not_registered( def test_registration_status_v2_push_host( client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: response = client.get( f"/registration_status_v2/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 200 @@ -858,7 +856,7 @@ def test_registration_status_v2_pull_host( tmp_path: Path, client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: source = site_context.agent_output_dir() / str(uuid) target_dir = tmp_path / "hostname" @@ -866,7 +864,7 @@ def test_registration_status_v2_pull_host( response = client.get( f"/registration_status_v2/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, ) assert response.status_code == 200 @@ -880,12 +878,12 @@ def test_registration_status_v2_pull_host( def test_renew_certificate_uuid_csr_mismatch( client: TestClient, uuid: UUID4, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: _key, wrong_csr = generate_csr_pair(str(uuid4()), 512) response = client.post( f"/renew_certificate/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, json={"csr": serialize_to_pem(wrong_csr)}, ) @@ -897,11 +895,11 @@ def test_renew_certificate_not_registered( client: TestClient, uuid: UUID4, serialized_csr: str, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: response = client.post( f"/renew_certificate/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, json={"csr": serialized_csr}, ) @@ -914,7 +912,7 @@ def test_renew_certificate_ok( client: TestClient, uuid: UUID4, serialized_csr: str, - registration_status_headers: Mapping[str, str], + registration_status_headers: MutableMapping[str, str], ) -> None: source = site_context.agent_output_dir() / str(uuid) target_dir = tmp_path / "hostname" @@ -922,7 +920,7 @@ def test_renew_certificate_ok( response = client.post( f"/renew_certificate/{uuid}", - headers=typeshed_issue_7724(registration_status_headers), + headers=registration_status_headers, json={"csr": serialized_csr}, )