Fix KNX Keyfile upload (#89029)

* Fix KNX Keyfile upload

* use shutil.move instead
pull/82635/head
Matthias Alphart 2023-03-02 16:13:02 +01:00 committed by GitHub
parent f69aa7ad9d
commit e849878a48
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 14 additions and 8 deletions

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator
from pathlib import Path
import shutil
from typing import Any, Final
import voluptuous as vol
@ -549,9 +550,12 @@ class KNXCommonFlow(ABC, FlowHandler):
),
None,
)
_tunnel_identifier = selected_tunnel_ia or self.new_entry_data.get(
CONF_HOST
)
_tunnel_suffix = f" @ {_tunnel_identifier}" if _tunnel_identifier else ""
self.new_title = (
f"{'Secure ' if _if_user_id else ''}"
f"Tunneling @ {selected_tunnel_ia or self.new_entry_data[CONF_HOST]}"
f"{'Secure ' if _if_user_id else ''}Tunneling{_tunnel_suffix}"
)
return self.finish_flow()
@ -708,7 +712,8 @@ class KNXCommonFlow(ABC, FlowHandler):
else:
dest_path = Path(self.hass.config.path(STORAGE_DIR, DOMAIN))
dest_path.mkdir(exist_ok=True)
file_path.rename(dest_path / DEFAULT_KNX_KEYRING_FILENAME)
dest_file = dest_path / DEFAULT_KNX_KEYRING_FILENAME
shutil.move(file_path, dest_file)
return keyring, errors
keyring, errors = await self.hass.async_add_executor_job(_process_upload)

View File

@ -77,16 +77,17 @@ def patch_file_upload(return_value=FIXTURE_KEYRING, side_effect=None):
side_effect=side_effect,
), patch(
"pathlib.Path.mkdir"
) as mkdir_mock:
file_path_mock = Mock()
file_upload_mock.return_value.__enter__.return_value = file_path_mock
) as mkdir_mock, patch(
"shutil.move"
) as shutil_move_mock:
file_upload_mock.return_value.__enter__.return_value = Mock()
yield return_value
if side_effect:
mkdir_mock.assert_not_called()
file_path_mock.rename.assert_not_called()
shutil_move_mock.assert_not_called()
else:
mkdir_mock.assert_called_once()
file_path_mock.rename.assert_called_once()
shutil_move_mock.assert_called_once()
def _gateway_descriptor(