jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-27 07:07:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rest test style (#42032)

Browse Source
This commit is contained in:
Julian Engelhardt 2020-10-22 10:00:48 +02:00 committed by GitHub
parent f92ebde75a
commit 5230ff6640
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 191 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

261
tests/components/rest/test_switch.py
View File

@ -16,70 +16,60 @@ from homeassistant.const import (
HTTP_OK, HTTP_OK,
) )
from homeassistant.helpers.template import Template from homeassistant.helpers.template import Template
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, get_test_home_assistant from tests.common import assert_setup_component
class TestRestSwitchSetup:
"""Tests for setting up the REST switch platform.""" """Tests for setting up the REST switch platform."""
def setup_method(self): NAME = "foo"
"""Set up things to be run when tests are started.""" METHOD = "post"
self.hass = get_test_home_assistant() RESOURCE = "http://localhost/"
STATE_RESOURCE = RESOURCE
HEADERS = {"Content-type": CONTENT_TYPE_JSON}
AUTH = None
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_missing_config(self): async def test_setup_missing_config(hass):
"""Test setup with configuration missing required entries.""" """Test setup with configuration missing required entries."""
assert not asyncio.run_coroutine_threadsafe( assert not await rest.async_setup_platform(hass, {CONF_PLATFORM: rest.DOMAIN}, None)
rest.async_setup_platform(self.hass, {CONF_PLATFORM: rest.DOMAIN}, None),
self.hass.loop,
).result()
def test_setup_missing_schema(self):
async def test_setup_missing_schema(hass):
"""Test setup with resource missing schema.""" """Test setup with resource missing schema."""
assert not asyncio.run_coroutine_threadsafe( assert not await rest.async_setup_platform(
rest.async_setup_platform( hass,
self.hass,
{CONF_PLATFORM: rest.DOMAIN, CONF_RESOURCE: "localhost"}, {CONF_PLATFORM: rest.DOMAIN, CONF_RESOURCE: "localhost"},
None, None,
), )
self.hass.loop,
).result()
def test_setup_failed_connect(self, aioclient_mock):
async def test_setup_failed_connect(hass, aioclient_mock):
"""Test setup when connection error occurs.""" """Test setup when connection error occurs."""
aioclient_mock.get("http://localhost", exc=aiohttp.ClientError) aioclient_mock.get("http://localhost", exc=aiohttp.ClientError)
assert not asyncio.run_coroutine_threadsafe( assert not await rest.async_setup_platform(
rest.async_setup_platform( hass,
self.hass,
{CONF_PLATFORM: rest.DOMAIN, CONF_RESOURCE: "http://localhost"}, {CONF_PLATFORM: rest.DOMAIN, CONF_RESOURCE: "http://localhost"},
None, None,
), )
self.hass.loop,
).result()
def test_setup_timeout(self, aioclient_mock):
async def test_setup_timeout(hass, aioclient_mock):
"""Test setup when connection timeout occurs.""" """Test setup when connection timeout occurs."""
aioclient_mock.get("http://localhost", exc=asyncio.TimeoutError()) aioclient_mock.get("http://localhost", exc=asyncio.TimeoutError())
assert not asyncio.run_coroutine_threadsafe( assert not await rest.async_setup_platform(
rest.async_setup_platform( hass,
self.hass,
{CONF_PLATFORM: rest.DOMAIN, CONF_RESOURCE: "http://localhost"}, {CONF_PLATFORM: rest.DOMAIN, CONF_RESOURCE: "http://localhost"},
None, None,
), )
self.hass.loop,
).result()
def test_setup_minimum(self, aioclient_mock):
async def test_setup_minimum(hass, aioclient_mock):
"""Test setup with minimum configuration.""" """Test setup with minimum configuration."""
aioclient_mock.get("http://localhost", status=HTTP_OK) aioclient_mock.get("http://localhost", status=HTTP_OK)
with assert_setup_component(1, SWITCH_DOMAIN): with assert_setup_component(1, SWITCH_DOMAIN):
assert setup_component( assert await async_setup_component(
self.hass, hass,
SWITCH_DOMAIN, SWITCH_DOMAIN,
{ {
SWITCH_DOMAIN: { SWITCH_DOMAIN: {
@ -90,11 +80,12 @@ class TestRestSwitchSetup:
) )
assert aioclient_mock.call_count == 1 assert aioclient_mock.call_count == 1
def test_setup(self, aioclient_mock):
async def test_setup(hass, aioclient_mock):
"""Test setup with valid configuration.""" """Test setup with valid configuration."""
aioclient_mock.get("http://localhost", status=HTTP_OK) aioclient_mock.get("http://localhost", status=HTTP_OK)
assert setup_component( assert await async_setup_component(
self.hass, hass,
SWITCH_DOMAIN, SWITCH_DOMAIN,
{ {
SWITCH_DOMAIN: { SWITCH_DOMAIN: {
@ -110,12 +101,13 @@ class TestRestSwitchSetup:
assert aioclient_mock.call_count == 1 assert aioclient_mock.call_count == 1
assert_setup_component(1, SWITCH_DOMAIN) assert_setup_component(1, SWITCH_DOMAIN)
def test_setup_with_state_resource(self, aioclient_mock):
async def test_setup_with_state_resource(hass, aioclient_mock):
"""Test setup with valid configuration.""" """Test setup with valid configuration."""
aioclient_mock.get("http://localhost", status=HTTP_NOT_FOUND) aioclient_mock.get("http://localhost", status=HTTP_NOT_FOUND)
aioclient_mock.get("http://localhost/state", status=HTTP_OK) aioclient_mock.get("http://localhost/state", status=HTTP_OK)
assert setup_component( assert await async_setup_component(
self.hass, hass,
SWITCH_DOMAIN, SWITCH_DOMAIN,
{ {
SWITCH_DOMAIN: { SWITCH_DOMAIN: {
@ -133,137 +125,130 @@ class TestRestSwitchSetup:
assert_setup_component(1, SWITCH_DOMAIN) assert_setup_component(1, SWITCH_DOMAIN)
class TestRestSwitch:
"""Tests for REST switch platform.""" """Tests for REST switch platform."""
def setup_method(self):
"""Set up things to be run when tests are started.""" def _setup_test_switch(hass):
self.hass = get_test_home_assistant() body_on = Template("on", hass)
self.name = "foo" body_off = Template("off", hass)
self.method = "post" switch = rest.RestSwitch(
self.resource = "http://localhost/" NAME,
self.state_resource = self.resource RESOURCE,
self.headers = {"Content-type": CONTENT_TYPE_JSON} STATE_RESOURCE,
self.auth = None METHOD,
self.body_on = Template("on", self.hass) HEADERS,
self.body_off = Template("off", self.hass) AUTH,
self.switch = rest.RestSwitch( body_on,
self.name, body_off,
self.resource,
self.state_resource,
self.method,
self.headers,
self.auth,
self.body_on,
self.body_off,
None, None,
10, 10,
True, True,
) )
self.switch.hass = self.hass switch.hass = hass
return switch, body_on, body_off
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_name(self): def test_name(hass):
"""Test the name.""" """Test the name."""
assert self.name == self.switch.name switch, body_on, body_off = _setup_test_switch(hass)
assert NAME == switch.name
def test_is_on_before_update(self):
def test_is_on_before_update(hass):
"""Test is_on in initial state.""" """Test is_on in initial state."""
assert self.switch.is_on is None switch, body_on, body_off = _setup_test_switch(hass)
assert switch.is_on is None
def test_turn_on_success(self, aioclient_mock):
async def test_turn_on_success(hass, aioclient_mock):
"""Test turn_on.""" """Test turn_on."""
aioclient_mock.post(self.resource, status=HTTP_OK) aioclient_mock.post(RESOURCE, status=HTTP_OK)
asyncio.run_coroutine_threadsafe( switch, body_on, body_off = _setup_test_switch(hass)
self.switch.async_turn_on(), self.hass.loop await switch.async_turn_on()
).result()
assert self.body_on.template == aioclient_mock.mock_calls[-1][2].decode() assert body_on.template == aioclient_mock.mock_calls[-1][2].decode()
assert self.switch.is_on assert switch.is_on
def test_turn_on_status_not_ok(self, aioclient_mock):
async def test_turn_on_status_not_ok(hass, aioclient_mock):
"""Test turn_on when error status returned.""" """Test turn_on when error status returned."""
aioclient_mock.post(self.resource, status=HTTP_INTERNAL_SERVER_ERROR) aioclient_mock.post(RESOURCE, status=HTTP_INTERNAL_SERVER_ERROR)
asyncio.run_coroutine_threadsafe( switch, body_on, body_off = _setup_test_switch(hass)
self.switch.async_turn_on(), self.hass.loop await switch.async_turn_on()
).result()
assert self.body_on.template == aioclient_mock.mock_calls[-1][2].decode() assert body_on.template == aioclient_mock.mock_calls[-1][2].decode()
assert self.switch.is_on is None assert switch.is_on is None
def test_turn_on_timeout(self, aioclient_mock):
async def test_turn_on_timeout(hass, aioclient_mock):
"""Test turn_on when timeout occurs.""" """Test turn_on when timeout occurs."""
aioclient_mock.post(self.resource, status=HTTP_INTERNAL_SERVER_ERROR) aioclient_mock.post(RESOURCE, status=HTTP_INTERNAL_SERVER_ERROR)
asyncio.run_coroutine_threadsafe( switch, body_on, body_off = _setup_test_switch(hass)
self.switch.async_turn_on(), self.hass.loop await switch.async_turn_on()
).result()
assert self.switch.is_on is None assert switch.is_on is None
def test_turn_off_success(self, aioclient_mock):
async def test_turn_off_success(hass, aioclient_mock):
"""Test turn_off.""" """Test turn_off."""
aioclient_mock.post(self.resource, status=HTTP_OK) aioclient_mock.post(RESOURCE, status=HTTP_OK)
asyncio.run_coroutine_threadsafe( switch, body_on, body_off = _setup_test_switch(hass)
self.switch.async_turn_off(), self.hass.loop await switch.async_turn_off()
).result()
assert self.body_off.template == aioclient_mock.mock_calls[-1][2].decode() assert body_off.template == aioclient_mock.mock_calls[-1][2].decode()
assert not self.switch.is_on assert not switch.is_on
def test_turn_off_status_not_ok(self, aioclient_mock):
async def test_turn_off_status_not_ok(hass, aioclient_mock):
"""Test turn_off when error status returned.""" """Test turn_off when error status returned."""
aioclient_mock.post(self.resource, status=HTTP_INTERNAL_SERVER_ERROR) aioclient_mock.post(RESOURCE, status=HTTP_INTERNAL_SERVER_ERROR)
asyncio.run_coroutine_threadsafe( switch, body_on, body_off = _setup_test_switch(hass)
self.switch.async_turn_off(), self.hass.loop await switch.async_turn_off()
).result()
assert self.body_off.template == aioclient_mock.mock_calls[-1][2].decode() assert body_off.template == aioclient_mock.mock_calls[-1][2].decode()
assert self.switch.is_on is None assert switch.is_on is None
def test_turn_off_timeout(self, aioclient_mock):
async def test_turn_off_timeout(hass, aioclient_mock):
"""Test turn_off when timeout occurs.""" """Test turn_off when timeout occurs."""
aioclient_mock.post(self.resource, exc=asyncio.TimeoutError()) aioclient_mock.post(RESOURCE, exc=asyncio.TimeoutError())
asyncio.run_coroutine_threadsafe( switch, body_on, body_off = _setup_test_switch(hass)
self.switch.async_turn_on(), self.hass.loop await switch.async_turn_on()
).result()
assert self.switch.is_on is None assert switch.is_on is None
def test_update_when_on(self, aioclient_mock):
async def test_update_when_on(hass, aioclient_mock):
"""Test update when switch is on.""" """Test update when switch is on."""
aioclient_mock.get(self.resource, text=self.body_on.template) switch, body_on, body_off = _setup_test_switch(hass)
asyncio.run_coroutine_threadsafe( aioclient_mock.get(RESOURCE, text=body_on.template)
self.switch.async_update(), self.hass.loop await switch.async_update()
).result()
assert self.switch.is_on assert switch.is_on
def test_update_when_off(self, aioclient_mock):
async def test_update_when_off(hass, aioclient_mock):
"""Test update when switch is off.""" """Test update when switch is off."""
aioclient_mock.get(self.resource, text=self.body_off.template) switch, body_on, body_off = _setup_test_switch(hass)
asyncio.run_coroutine_threadsafe( aioclient_mock.get(RESOURCE, text=body_off.template)
self.switch.async_update(), self.hass.loop await switch.async_update()
).result()
assert not self.switch.is_on assert not switch.is_on
def test_update_when_unknown(self, aioclient_mock):
async def test_update_when_unknown(hass, aioclient_mock):
"""Test update when unknown status returned.""" """Test update when unknown status returned."""
aioclient_mock.get(self.resource, text="unknown status") aioclient_mock.get(RESOURCE, text="unknown status")
asyncio.run_coroutine_threadsafe( switch, body_on, body_off = _setup_test_switch(hass)
self.switch.async_update(), self.hass.loop await switch.async_update()
).result()
assert self.switch.is_on is None assert switch.is_on is None
def test_update_timeout(self, aioclient_mock):
async def test_update_timeout(hass, aioclient_mock):
"""Test update when timeout occurs.""" """Test update when timeout occurs."""
aioclient_mock.get(self.resource, exc=asyncio.TimeoutError()) aioclient_mock.get(RESOURCE, exc=asyncio.TimeoutError())
asyncio.run_coroutine_threadsafe( switch, body_on, body_off = _setup_test_switch(hass)
self.switch.async_update(), self.hass.loop await switch.async_update()
).result()
assert self.switch.is_on is None assert switch.is_on is None