Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

No longer rely on requests (#23685) 

* No longer rely on requests

* Lint

* Missed a few parts

* Fix types

* Fix more types

* Update __main__.py

* Fix tests

* Lint

* Fix script
pull/23760/head^2
Paulus Schoutsen 2019-05-08 11:15:04 -07:00 committed by GitHub
parent f019e2a204
commit cc13713abd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 271 additions and 197 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
homeassistant/__main__.py
View File

@ -7,8 +7,9 @@ import platform
import subprocess
import sys
import threading
from typing import List, Dict, Any # noqa pylint: disable=unused-import
from typing import ( # noqa pylint: disable=unused-import
List, Dict, Any, TYPE_CHECKING
)
from homeassistant import monkey_patch
from homeassistant.const import (
@ -18,6 +19,9 @@ from homeassistant.const import (
RESTART_EXIT_CODE,
)
if TYPE_CHECKING:
from homeassistant import core
def set_loop() -> None:
"""Attempt to use uvloop."""
@ -86,10 +90,12 @@ def ensure_config_path(config_dir: str) -> None:
sys.exit(1)
def ensure_config_file(config_dir: str) -> str:
async def ensure_config_file(hass: 'core.HomeAssistant', config_dir: str) \
-> str:
"""Ensure configuration file exists."""
import homeassistant.config as config_util
config_path = config_util.ensure_config_exists(config_dir)
config_path = await config_util.async_ensure_config_exists(
hass, config_dir)
if config_path is None:
print('Error getting configuration path')
@ -261,6 +267,7 @@ def cmdline() -> List[str]:
async def setup_and_run_hass(config_dir: str,
args: argparse.Namespace) -> int:
"""Set up HASS and run."""
# pylint: disable=redefined-outer-name
from homeassistant import bootstrap, core
hass = core.HomeAssistant()
@ -275,7 +282,7 @@ async def setup_and_run_hass(config_dir: str,
skip_pip=args.skip_pip, log_rotate_days=args.log_rotate_days,
log_file=args.log_file, log_no_color=args.log_no_color)
else:
config_file = ensure_config_file(config_dir)
config_file = await ensure_config_file(hass, config_dir)
print('Config directory:', config_dir)
await bootstrap.async_from_config_file(
config_file, hass, verbose=args.verbose, skip_pip=args.skip_pip,
@ -390,7 +397,7 @@ def main() -> int:
if exit_code == RESTART_EXIT_CODE and not args.runner:
try_to_restart()
return exit_code # type: ignore # mypy cannot yet infer it
return exit_code # type: ignore
if __name__ == "__main__":

4
homeassistant/components/ps4/__init__.py
View File

@ -48,7 +48,9 @@ async def async_migrate_entry(hass, entry):
# Migrate Version 1 -> Version 2: New region codes.
if version == 1:
loc = await hass.async_add_executor_job(location.detect_location_info)
loc = await location.async_detect_location_info(
hass.helpers.aiohttp_client.async_get_clientsession()
)
if loc:
country = loc.country_name
if country in COUNTRIES:

5
homeassistant/components/ps4/config_flow.py
View File

@ -167,8 +167,9 @@ class PlayStation4FlowHandler(config_entries.ConfigFlow):
# Try to find region automatically.
if not self.location:
self.location = await self.hass.async_add_executor_job(
location.detect_location_info)
self.location = await location.async_detect_location_info(
self.hass.helpers.aiohttp_client.async_get_clientsession()
)
if self.location:
country = self.location.country_name
if country in COUNTRIES:

71
homeassistant/config.py
View File

@ -205,7 +205,8 @@ def get_default_config_dir() -> str:
return os.path.join(data_dir, CONFIG_DIR_NAME) # type: ignore
def ensure_config_exists(config_dir: str, detect_location: bool = True)\
async def async_ensure_config_exists(hass: HomeAssistant, config_dir: str,
detect_location: bool = True)\
-> Optional[str]:
"""Ensure a configuration file exists in given configuration directory.
@ -217,18 +218,51 @@ def ensure_config_exists(config_dir: str, detect_location: bool = True)\
if config_path is None:
print("Unable to find configuration. Creating default one in",
config_dir)
config_path = create_default_config(config_dir, detect_location)
config_path = await async_create_default_config(
hass, config_dir, detect_location)
return config_path
def create_default_config(config_dir: str, detect_location: bool = True)\
-> Optional[str]:
async def async_create_default_config(
hass: HomeAssistant, config_dir: str, detect_location: bool = True
) -> Optional[str]:
"""Create a default configuration file in given configuration directory.
Return path to new config file if success, None if failed.
This method needs to run in an executor.
"""
info = {attr: default for attr, default, _, _ in DEFAULT_CORE_CONFIG}
if detect_location:
session = hass.helpers.aiohttp_client.async_get_clientsession()
location_info = await loc_util.async_detect_location_info(session)
else:
location_info = None
if location_info:
if location_info.use_metric:
info[CONF_UNIT_SYSTEM] = CONF_UNIT_SYSTEM_METRIC
else:
info[CONF_UNIT_SYSTEM] = CONF_UNIT_SYSTEM_IMPERIAL
for attr, default, prop, _ in DEFAULT_CORE_CONFIG:
if prop is None:
continue
info[attr] = getattr(location_info, prop) or default
if location_info.latitude and location_info.longitude:
info[CONF_ELEVATION] = await loc_util.async_get_elevation(
session, location_info.latitude, location_info.longitude)
return await hass.async_add_executor_job(
_write_default_config, config_dir, info
)
def _write_default_config(config_dir: str, info: Dict)\
-> Optional[str]:
"""Write the default config."""
from homeassistant.components.config.group import (
CONFIG_PATH as GROUP_CONFIG_PATH)
from homeassistant.components.config.automation import (
@ -246,25 +280,6 @@ def create_default_config(config_dir: str, detect_location: bool = True)\
script_yaml_path = os.path.join(config_dir, SCRIPT_CONFIG_PATH)
customize_yaml_path = os.path.join(config_dir, CUSTOMIZE_CONFIG_PATH)
info = {attr: default for attr, default, _, _ in DEFAULT_CORE_CONFIG}
location_info = detect_location and loc_util.detect_location_info()
if location_info:
if location_info.use_metric:
info[CONF_UNIT_SYSTEM] = CONF_UNIT_SYSTEM_METRIC
else:
info[CONF_UNIT_SYSTEM] = CONF_UNIT_SYSTEM_IMPERIAL
for attr, default, prop, _ in DEFAULT_CORE_CONFIG:
if prop is None:
continue
info[attr] = getattr(location_info, prop) or default
if location_info.latitude and location_info.longitude:
info[CONF_ELEVATION] = loc_util.elevation(
location_info.latitude, location_info.longitude)
# Writing files with YAML does not create the most human readable results
# So we're hard coding a YAML template.
try:
@ -576,8 +591,9 @@ async def async_process_ha_core_config(
# If we miss some of the needed values, auto detect them
if None in (hac.latitude, hac.longitude, hac.units,
hac.time_zone):
info = await hass.async_add_executor_job(
loc_util.detect_location_info)
info = await loc_util.async_detect_location_info(
hass.helpers.aiohttp_client.async_get_clientsession()
)
if info is None:
_LOGGER.error("Could not detect location information")
@ -602,8 +618,9 @@ async def async_process_ha_core_config(
if hac.elevation is None and hac.latitude is not None and \
hac.longitude is not None:
elevation = await hass.async_add_executor_job(
loc_util.elevation, hac.latitude, hac.longitude)
elevation = await loc_util.async_get_elevation(
hass.helpers.aiohttp_client.async_get_clientsession(),
hac.latitude, hac.longitude)
hac.elevation = elevation
discovered.append(('elevation', elevation))

11
homeassistant/scripts/ensure_config.py
View File

@ -2,6 +2,7 @@
import argparse
import os
from homeassistant.core import HomeAssistant
import homeassistant.config as config_util
@ -28,6 +29,14 @@ def run(args):
print('Creating directory', config_dir)
os.makedirs(config_dir)
config_path = config_util.ensure_config_exists(config_dir)
hass = HomeAssistant()
config_path = hass.loop.run_until_complete(async_run(hass, config_dir))
print('Configuration file:', config_path)
return 0
async def async_run(hass, config_dir):
"""Make sure config exists."""
path = await config_util.async_ensure_config_exists(hass, config_dir)
await hass.async_stop(force=True)
return path

56
homeassistant/util/location.py
View File

@ -3,11 +3,12 @@ Module with location helpers.
detect_location_info and elevation are mocked by default during tests.
"""
import asyncio
import collections
import math
from typing import Any, Optional, Tuple, Dict
import requests
import aiohttp
ELEVATION_URL = 'https://api.open-elevation.com/api/v1/lookup'
IP_API = 'http://ip-api.com/json'
@ -33,12 +34,13 @@ LocationInfo = collections.namedtuple(
'use_metric'])
def detect_location_info() -> Optional[LocationInfo]:
async def async_detect_location_info(session: aiohttp.ClientSession) \
-> Optional[LocationInfo]:
"""Detect location information."""
data = _get_ipapi()
data = await _get_ipapi(session)
if data is None:
data = _get_ip_api()
data = await _get_ip_api(session)
if data is None:
return None
@ -63,23 +65,26 @@ def distance(lat1: Optional[float], lon1: Optional[float],
return result * 1000
def elevation(latitude: float, longitude: float) -> int:
async def async_get_elevation(session: aiohttp.ClientSession, latitude: float,
longitude: float) -> int:
"""Return elevation for given latitude and longitude."""
try:
req = requests.get(
ELEVATION_URL,
params={
'locations': '{},{}'.format(latitude, longitude),
},
timeout=10)
except requests.RequestException:
resp = await session.get(ELEVATION_URL, params={
'locations': '{},{}'.format(latitude, longitude),
}, timeout=5)
except (aiohttp.ClientError, asyncio.TimeoutError):
return 0
if req.status_code != 200:
if resp.status != 200:
return 0
try:
return int(float(req.json()['results'][0]['elevation']))
raw_info = await resp.json()
except (aiohttp.ClientError, ValueError):
return 0
try:
return int(float(raw_info['results'][0]['elevation']))
except (ValueError, KeyError, IndexError):
return 0
@ -158,11 +163,17 @@ def vincenty(point1: Tuple[float, float], point2: Tuple[float, float],
return round(s, 6)
def _get_ipapi() -> Optional[Dict[str, Any]]:
async def _get_ipapi(session: aiohttp.ClientSession) \
-> Optional[Dict[str, Any]]:
"""Query ipapi.co for location data."""
try:
raw_info = requests.get(IPAPI, timeout=5).json()
except (requests.RequestException, ValueError):
resp = await session.get(IPAPI, timeout=5)
except (aiohttp.ClientError, asyncio.TimeoutError):
return None
try:
raw_info = await resp.json()
except (aiohttp.ClientError, ValueError):
return None
return {
@ -179,13 +190,18 @@ def _get_ipapi() -> Optional[Dict[str, Any]]:
}
def _get_ip_api() -> Optional[Dict[str, Any]]:
async def _get_ip_api(session: aiohttp.ClientSession) \
-> Optional[Dict[str, Any]]:
"""Query ip-api.com for location data."""
try:
raw_info = requests.get(IP_API, timeout=5).json()
except (requests.RequestException, ValueError):
resp = await session.get(IP_API, timeout=5)
except (aiohttp.ClientError, asyncio.TimeoutError):
return None
try:
raw_info = await resp.json()
except (aiohttp.ClientError, ValueError):
return None
return {
'ip': raw_info.get('query'),
'country_code': raw_info.get('countryCode'),

6
tests/conftest.py
View File

@ -42,8 +42,10 @@ def check_real(func):
# Guard a few functions that would make network connections
location.detect_location_info = check_real(location.detect_location_info)
location.elevation = check_real(location.elevation)
location.async_detect_location_info = \
check_real(location.async_detect_location_info)
location.async_get_elevation = \
check_real(location.async_get_elevation)
util.get_local_ip = lambda: '127.0.0.1'

4
tests/test_bootstrap.py
View File

@ -22,8 +22,8 @@ _LOGGER = logging.getLogger(__name__)
# prevent .HA_VERSION file from being written
@patch(
'homeassistant.bootstrap.conf_util.process_ha_config_upgrade', Mock())
@patch('homeassistant.util.location.detect_location_info',
Mock(return_value=None))
@patch('homeassistant.util.location.async_detect_location_info',
Mock(return_value=mock_coro(None)))
@patch('os.path.isfile', Mock(return_value=True))
@patch('os.access', Mock(return_value=True))
@patch('homeassistant.bootstrap.async_enable_logging',

66
tests/test_config.py
View File

@ -34,7 +34,7 @@ from homeassistant.components.config.customize import (
import homeassistant.scripts.check_config as check_config
from tests.common import (
get_test_config_dir, patch_yaml_files)
get_test_config_dir, patch_yaml_files, mock_coro)
CONFIG_DIR = get_test_config_dir()
YAML_PATH = os.path.join(CONFIG_DIR, config_util.YAML_CONFIG_FILE)
@ -79,9 +79,9 @@ def teardown():
os.remove(CUSTOMIZE_PATH)
def test_create_default_config():
async def test_create_default_config(hass):
"""Test creation of default config."""
config_util.create_default_config(CONFIG_DIR, False)
await config_util.async_create_default_config(hass, CONFIG_DIR, False)
assert os.path.isfile(YAML_PATH)
assert os.path.isfile(SECRET_PATH)
@ -98,22 +98,22 @@ def test_find_config_file_yaml():
assert YAML_PATH == config_util.find_config_file(CONFIG_DIR)
@mock.patch('builtins.print')
def test_ensure_config_exists_creates_config(mock_print):
async def test_ensure_config_exists_creates_config(hass):
"""Test that calling ensure_config_exists.
If not creates a new config file.
"""
config_util.ensure_config_exists(CONFIG_DIR, False)
with mock.patch('builtins.print') as mock_print:
await config_util.async_ensure_config_exists(hass, CONFIG_DIR, False)
assert os.path.isfile(YAML_PATH)
assert mock_print.called
def test_ensure_config_exists_uses_existing_config():
async def test_ensure_config_exists_uses_existing_config(hass):
"""Test that calling ensure_config_exists uses existing config."""
create_file(YAML_PATH)
config_util.ensure_config_exists(CONFIG_DIR, False)
await config_util.async_ensure_config_exists(hass, CONFIG_DIR, False)
with open(YAML_PATH) as f:
content = f.read()
@ -166,17 +166,17 @@ def test_load_yaml_config_preserves_key_order():
list(config_util.load_yaml_config_file(YAML_PATH).items())
@mock.patch('homeassistant.util.location.detect_location_info',
return_value=location_util.LocationInfo(
'0.0.0.0', 'US', 'United States', 'CA', 'California',
'San Diego', '92122', 'America/Los_Angeles', 32.8594,
-117.2073, True))
@mock.patch('homeassistant.util.location.elevation', return_value=101)
@mock.patch('builtins.print')
def test_create_default_config_detect_location(mock_detect,
mock_elev, mock_print):
async def test_create_default_config_detect_location(hass):
"""Test that detect location sets the correct config keys."""
config_util.ensure_config_exists(CONFIG_DIR)
with mock.patch('homeassistant.util.location.async_detect_location_info',
return_value=mock_coro(location_util.LocationInfo(
'0.0.0.0', 'US', 'United States', 'CA', 'California',
'San Diego', '92122', 'America/Los_Angeles', 32.8594,
-117.2073, True))), \
mock.patch('homeassistant.util.location.async_get_elevation',
return_value=mock_coro(101)), \
mock.patch('builtins.print') as mock_print:
await config_util.async_ensure_config_exists(hass, CONFIG_DIR)
config = config_util.load_yaml_config_file(YAML_PATH)
@ -198,14 +198,14 @@ def test_create_default_config_detect_location(mock_detect,
assert mock_print.called
@mock.patch('builtins.print')
def test_create_default_config_returns_none_if_write_error(mock_print):
async def test_create_default_config_returns_none_if_write_error(hass):
"""Test the writing of a default configuration.
Non existing folder returns None.
"""
assert config_util.create_default_config(
os.path.join(CONFIG_DIR, 'non_existing_dir/'), False) is None
with mock.patch('builtins.print') as mock_print:
assert await config_util.async_create_default_config(
hass, os.path.join(CONFIG_DIR, 'non_existing_dir/'), False) is None
assert mock_print.called
@ -490,13 +490,14 @@ async def test_loading_configuration_from_packages(hass):
})
@asynctest.mock.patch('homeassistant.util.location.detect_location_info',
autospec=True, return_value=location_util.LocationInfo(
'0.0.0.0', 'US', 'United States', 'CA', 'California',
'San Diego', '92122', 'America/Los_Angeles', 32.8594,
-117.2073, True))
@asynctest.mock.patch('homeassistant.util.location.elevation',
autospec=True, return_value=101)
@asynctest.mock.patch(
'homeassistant.util.location.async_detect_location_info',
autospec=True, return_value=mock_coro(location_util.LocationInfo(
'0.0.0.0', 'US', 'United States', 'CA',
'California', 'San Diego', '92122',
'America/Los_Angeles', 32.8594, -117.2073, True)))
@asynctest.mock.patch('homeassistant.util.location.async_get_elevation',
autospec=True, return_value=mock_coro(101))
async def test_discovering_configuration(mock_detect, mock_elevation, hass):
"""Test auto discovery for missing core configs."""
hass.config.latitude = None
@ -516,9 +517,10 @@ async def test_discovering_configuration(mock_detect, mock_elevation, hass):
assert hass.config.time_zone.zone == 'America/Los_Angeles'
@asynctest.mock.patch('homeassistant.util.location.detect_location_info',
autospec=True, return_value=None)
@asynctest.mock.patch('homeassistant.util.location.elevation', return_value=0)
@asynctest.mock.patch('homeassistant.util.location.async_detect_location_info',
autospec=True, return_value=mock_coro(None))
@asynctest.mock.patch('homeassistant.util.location.async_get_elevation',
return_value=mock_coro(0))
async def test_discovering_configuration_auto_detect_fails(mock_detect,
mock_elevation,
hass):

226
tests/util/test_location.py
View File

@ -1,13 +1,12 @@
"""Test Home Assistant location util methods."""
from unittest import TestCase
from unittest.mock import patch
from unittest.mock import patch, Mock
import requests
import requests_mock
import aiohttp
import pytest
import homeassistant.util.location as location_util
from tests.common import load_fixture
from tests.common import load_fixture, mock_coro
# Paris
COORDINATES_PARIS = (48.864716, 2.349014)
@ -25,120 +24,139 @@ DISTANCE_KM = 5846.39
DISTANCE_MILES = 3632.78
class TestLocationUtil(TestCase):
"""Test util location methods."""
@pytest.fixture
async def session(hass):
"""Return aioclient session."""
return hass.helpers.aiohttp_client.async_get_clientsession()
def test_get_distance_to_same_place(self):
"""Test getting the distance."""
meters = location_util.distance(
COORDINATES_PARIS[0], COORDINATES_PARIS[1],
COORDINATES_PARIS[0], COORDINATES_PARIS[1])
assert meters == 0
@pytest.fixture
async def raising_session(loop):
"""Return an aioclient session that only fails."""
return Mock(get=Mock(side_effect=aiohttp.ClientError))
def test_get_distance(self):
"""Test getting the distance."""
meters = location_util.distance(
COORDINATES_PARIS[0], COORDINATES_PARIS[1],
COORDINATES_NEW_YORK[0], COORDINATES_NEW_YORK[1])
assert meters/1000 - DISTANCE_KM < 0.01
def test_get_distance_to_same_place():
"""Test getting the distance."""
meters = location_util.distance(
COORDINATES_PARIS[0], COORDINATES_PARIS[1],
COORDINATES_PARIS[0], COORDINATES_PARIS[1])
def test_get_kilometers(self):
"""Test getting the distance between given coordinates in km."""
kilometers = location_util.vincenty(
COORDINATES_PARIS, COORDINATES_NEW_YORK)
assert round(kilometers, 2) == DISTANCE_KM
assert meters == 0
def test_get_miles(self):
"""Test getting the distance between given coordinates in miles."""
miles = location_util.vincenty(
COORDINATES_PARIS, COORDINATES_NEW_YORK, miles=True)
assert round(miles, 2) == DISTANCE_MILES
@requests_mock.Mocker()
def test_detect_location_info_ipapi(self, m):
"""Test detect location info using ipapi.co."""
m.get(
location_util.IPAPI, text=load_fixture('ipapi.co.json'))
def test_get_distance():
"""Test getting the distance."""
meters = location_util.distance(
COORDINATES_PARIS[0], COORDINATES_PARIS[1],
COORDINATES_NEW_YORK[0], COORDINATES_NEW_YORK[1])
info = location_util.detect_location_info(_test_real=True)
assert meters/1000 - DISTANCE_KM < 0.01
assert info is not None
assert info.ip == '1.2.3.4'
assert info.country_code == 'CH'
assert info.country_name == 'Switzerland'
assert info.region_code == 'BE'
assert info.region_name == 'Bern'
assert info.city == 'Bern'
assert info.zip_code == '3000'
assert info.time_zone == 'Europe/Zurich'
assert info.latitude == 46.9480278
assert info.longitude == 7.4490812
assert info.use_metric
@requests_mock.Mocker()
@patch('homeassistant.util.location._get_ipapi', return_value=None)
def test_detect_location_info_ip_api(self, mock_req, mock_ipapi):
"""Test detect location info using ip-api.com."""
mock_req.get(
location_util.IP_API, text=load_fixture('ip-api.com.json'))
def test_get_kilometers():
"""Test getting the distance between given coordinates in km."""
kilometers = location_util.vincenty(
COORDINATES_PARIS, COORDINATES_NEW_YORK)
assert round(kilometers, 2) == DISTANCE_KM
info = location_util.detect_location_info(_test_real=True)
assert info is not None
assert info.ip == '1.2.3.4'
assert info.country_code == 'US'
assert info.country_name == 'United States'
assert info.region_code == 'CA'
assert info.region_name == 'California'
assert info.city == 'San Diego'
assert info.zip_code == '92122'
assert info.time_zone == 'America/Los_Angeles'
assert info.latitude == 32.8594
assert info.longitude == -117.2073
assert not info.use_metric
def test_get_miles():
"""Test getting the distance between given coordinates in miles."""
miles = location_util.vincenty(
COORDINATES_PARIS, COORDINATES_NEW_YORK, miles=True)
assert round(miles, 2) == DISTANCE_MILES
@patch('homeassistant.util.location.elevation', return_value=0)
@patch('homeassistant.util.location._get_ipapi', return_value=None)
@patch('homeassistant.util.location._get_ip_api', return_value=None)
def test_detect_location_info_both_queries_fail(
self, mock_ipapi, mock_ip_api, mock_elevation):
"""Ensure we return None if both queries fail."""
info = location_util.detect_location_info(_test_real=True)
assert info is None
@patch('homeassistant.util.location.requests.get',
side_effect=requests.RequestException)
def test_freegeoip_query_raises(self, mock_get):
"""Test ipapi.co query when the request to API fails."""
info = location_util._get_ipapi()
assert info is None
async def test_detect_location_info_ipapi(aioclient_mock, session):
"""Test detect location info using ipapi.co."""
aioclient_mock.get(
location_util.IPAPI, text=load_fixture('ipapi.co.json'))
@patch('homeassistant.util.location.requests.get',
side_effect=requests.RequestException)
def test_ip_api_query_raises(self, mock_get):
"""Test ip api query when the request to API fails."""
info = location_util._get_ip_api()
assert info is None
info = await location_util.async_detect_location_info(
session, _test_real=True)
@patch('homeassistant.util.location.requests.get',
side_effect=requests.RequestException)
def test_elevation_query_raises(self, mock_get):
"""Test elevation when the request to API fails."""
elevation = location_util.elevation(10, 10, _test_real=True)
assert elevation == 0
assert info is not None
assert info.ip == '1.2.3.4'
assert info.country_code == 'CH'
assert info.country_name == 'Switzerland'
assert info.region_code == 'BE'
assert info.region_name == 'Bern'
assert info.city == 'Bern'
assert info.zip_code == '3000'
assert info.time_zone == 'Europe/Zurich'
assert info.latitude == 46.9480278
assert info.longitude == 7.4490812
assert info.use_metric
@requests_mock.Mocker()
def test_elevation_query_fails(self, mock_req):
"""Test elevation when the request to API fails."""
mock_req.get(location_util.ELEVATION_URL, text='{}', status_code=401)
elevation = location_util.elevation(10, 10, _test_real=True)
assert elevation == 0
@requests_mock.Mocker()
def test_elevation_query_nonjson(self, mock_req):
"""Test if elevation API returns a non JSON value."""
mock_req.get(location_util.ELEVATION_URL, text='{ I am not JSON }')
elevation = location_util.elevation(10, 10, _test_real=True)
assert elevation == 0
async def test_detect_location_info_ip_api(aioclient_mock, session):
"""Test detect location info using ip-api.com."""
aioclient_mock.get(
location_util.IP_API, text=load_fixture('ip-api.com.json'))
with patch('homeassistant.util.location._get_ipapi',
return_value=mock_coro(None)):
info = await location_util.async_detect_location_info(
session, _test_real=True)
assert info is not None
assert info.ip == '1.2.3.4'
assert info.country_code == 'US'
assert info.country_name == 'United States'
assert info.region_code == 'CA'
assert info.region_name == 'California'
assert info.city == 'San Diego'
assert info.zip_code == '92122'
assert info.time_zone == 'America/Los_Angeles'
assert info.latitude == 32.8594
assert info.longitude == -117.2073
assert not info.use_metric
async def test_detect_location_info_both_queries_fail(session):
"""Ensure we return None if both queries fail."""
with patch('homeassistant.util.location.async_get_elevation',
return_value=mock_coro(0)), \
patch('homeassistant.util.location._get_ipapi',
return_value=mock_coro(None)), \
patch('homeassistant.util.location._get_ip_api',
return_value=mock_coro(None)):
info = await location_util.async_detect_location_info(
session, _test_real=True)
assert info is None
async def test_freegeoip_query_raises(raising_session):
"""Test ipapi.co query when the request to API fails."""
info = await location_util._get_ipapi(raising_session)
assert info is None
async def test_ip_api_query_raises(raising_session):
"""Test ip api query when the request to API fails."""
info = await location_util._get_ip_api(raising_session)
assert info is None
async def test_elevation_query_raises(raising_session):
"""Test elevation when the request to API fails."""
elevation = await location_util.async_get_elevation(
raising_session, 10, 10, _test_real=True)
assert elevation == 0
async def test_elevation_query_fails(aioclient_mock, session):
"""Test elevation when the request to API fails."""
aioclient_mock.get(location_util.ELEVATION_URL, text='{}', status=401)
elevation = await location_util.async_get_elevation(
session, 10, 10, _test_real=True)
assert elevation == 0
async def test_elevation_query_nonjson(aioclient_mock, session):
"""Test if elevation API returns a non JSON value."""
aioclient_mock.get(location_util.ELEVATION_URL, text='{ I am not JSON }')
elevation = await location_util.async_get_elevation(
session, 10, 10, _test_real=True)
assert elevation == 0