home-assistant-core/tests/test_requirements.py

665 lines
22 KiB
Python

"""Test requirements module."""
import asyncio
import logging
import os
from unittest.mock import call, patch
import pytest
from homeassistant import loader, setup
from homeassistant.core import HomeAssistant
from homeassistant.loader import async_get_integration
from homeassistant.requirements import (
CONSTRAINT_FILE,
RequirementsNotFound,
_async_get_manager,
async_clear_install_history,
async_get_integration_with_requirements,
async_process_requirements,
)
from .common import MockModule, mock_integration
def env_without_wheel_links():
"""Return env without wheel links."""
env = dict(os.environ)
env.pop("WHEEL_LINKS", None)
return env
async def test_requirement_installed_in_venv(hass: HomeAssistant) -> None:
"""Test requirement installed in virtual environment."""
with (
patch("os.path.dirname", return_value="ha_package_path"),
patch("homeassistant.util.package.is_virtual_env", return_value=True),
patch("homeassistant.util.package.is_docker_env", return_value=False),
patch(
"homeassistant.util.package.install_package", return_value=True
) as mock_install,
patch.dict(os.environ, env_without_wheel_links(), clear=True),
):
hass.config.skip_pip = False
mock_integration(hass, MockModule("comp", requirements=["package==0.0.1"]))
assert await setup.async_setup_component(hass, "comp", {})
assert "comp" in hass.config.components
assert mock_install.call_args == call(
"package==0.0.1",
constraints=os.path.join("ha_package_path", CONSTRAINT_FILE),
timeout=60,
)
async def test_requirement_installed_in_deps(hass: HomeAssistant) -> None:
"""Test requirement installed in deps directory."""
with (
patch("os.path.dirname", return_value="ha_package_path"),
patch("homeassistant.util.package.is_virtual_env", return_value=False),
patch("homeassistant.util.package.is_docker_env", return_value=False),
patch(
"homeassistant.util.package.install_package", return_value=True
) as mock_install,
patch.dict(os.environ, env_without_wheel_links(), clear=True),
):
hass.config.skip_pip = False
mock_integration(hass, MockModule("comp", requirements=["package==0.0.1"]))
assert await setup.async_setup_component(hass, "comp", {})
assert "comp" in hass.config.components
assert mock_install.call_args == call(
"package==0.0.1",
target=hass.config.path("deps"),
constraints=os.path.join("ha_package_path", CONSTRAINT_FILE),
timeout=60,
)
async def test_install_existing_package(hass: HomeAssistant) -> None:
"""Test an install attempt on an existing package."""
with patch(
"homeassistant.util.package.install_package", return_value=True
) as mock_inst:
await async_process_requirements(hass, "test_component", ["hello==1.0.0"])
assert len(mock_inst.mock_calls) == 1
with (
patch("homeassistant.util.package.is_installed", return_value=True),
patch("homeassistant.util.package.install_package") as mock_inst,
):
await async_process_requirements(hass, "test_component", ["hello==1.0.0"])
assert len(mock_inst.mock_calls) == 0
async def test_install_missing_package(hass: HomeAssistant) -> None:
"""Test an install attempt on an existing package."""
with (
patch(
"homeassistant.util.package.install_package", return_value=False
) as mock_inst,
pytest.raises(RequirementsNotFound),
):
await async_process_requirements(hass, "test_component", ["hello==1.0.0"])
assert len(mock_inst.mock_calls) == 3
async def test_install_skipped_package(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test an install attempt on a dependency that should be skipped."""
with patch(
"homeassistant.util.package.install_package", return_value=True
) as mock_inst:
hass.config.skip_pip_packages = ["hello"]
with caplog.at_level(logging.WARNING):
await async_process_requirements(
hass, "test_component", ["hello==1.0.0", "not_skipped==1.2.3"]
)
assert "Skipping requirement hello==1.0.0" in caplog.text
assert len(mock_inst.mock_calls) == 1
assert mock_inst.mock_calls[0].args[0] == "not_skipped==1.2.3"
async def test_get_integration_with_requirements(hass: HomeAssistant) -> None:
"""Check getting an integration with loaded requirements."""
hass.config.skip_pip = False
mock_integration(
hass, MockModule("test_component_dep", requirements=["test-comp-dep==1.0.0"])
)
mock_integration(
hass,
MockModule(
"test_component_after_dep", requirements=["test-comp-after-dep==1.0.0"]
),
)
mock_integration(
hass,
MockModule(
"test_component",
requirements=["test-comp==1.0.0"],
dependencies=["test_component_dep"],
partial_manifest={"after_dependencies": ["test_component_after_dep"]},
),
)
with (
patch(
"homeassistant.util.package.is_installed", return_value=False
) as mock_is_installed,
patch(
"homeassistant.util.package.install_package", return_value=True
) as mock_inst,
):
integration = await async_get_integration_with_requirements(
hass, "test_component"
)
assert integration
assert integration.domain == "test_component"
assert len(mock_is_installed.mock_calls) == 3
assert sorted(mock_call[1][0] for mock_call in mock_is_installed.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp==1.0.0",
]
assert len(mock_inst.mock_calls) == 3
assert sorted(mock_call[1][0] for mock_call in mock_inst.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp==1.0.0",
]
async def test_get_integration_with_requirements_cache(hass: HomeAssistant) -> None:
"""Check getting an integration with loaded requirements considers cache.
We want to make sure that we do not check requirements for dependencies
that we have already checked.
"""
hass.config.skip_pip = False
mock_integration(
hass, MockModule("test_component_dep", requirements=["test-comp-dep==1.0.0"])
)
mock_integration(
hass,
MockModule(
"test_component_after_dep", requirements=["test-comp-after-dep==1.0.0"]
),
)
mock_integration(
hass,
MockModule(
"test_component",
requirements=["test-comp==1.0.0"],
dependencies=["test_component_dep"],
partial_manifest={"after_dependencies": ["test_component_after_dep"]},
),
)
mock_integration(
hass,
MockModule(
"test_component2",
requirements=["test-comp2==1.0.0"],
dependencies=["test_component_dep"],
partial_manifest={"after_dependencies": ["test_component_after_dep"]},
),
)
with (
patch(
"homeassistant.util.package.is_installed", return_value=False
) as mock_is_installed,
patch(
"homeassistant.util.package.install_package", return_value=True
) as mock_inst,
patch(
"homeassistant.requirements.async_get_integration",
wraps=async_get_integration,
) as mock_async_get_integration,
):
integration = await async_get_integration_with_requirements(
hass, "test_component"
)
assert integration
assert integration.domain == "test_component"
assert len(mock_is_installed.mock_calls) == 3
assert sorted(
mock_call[1][0] for mock_call in mock_is_installed.mock_calls
) == [
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp==1.0.0",
]
assert len(mock_inst.mock_calls) == 3
assert sorted(mock_call[1][0] for mock_call in mock_inst.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp==1.0.0",
]
# The dependent integrations should be fetched since
assert len(mock_async_get_integration.mock_calls) == 3
assert sorted(
mock_call[1][1] for mock_call in mock_async_get_integration.mock_calls
) == ["test_component", "test_component_after_dep", "test_component_dep"]
# test_component2 has the same deps as test_component and we should
# not check the requirements for the deps again
mock_is_installed.reset_mock()
mock_inst.reset_mock()
mock_async_get_integration.reset_mock()
integration = await async_get_integration_with_requirements(
hass, "test_component2"
)
assert integration
assert integration.domain == "test_component2"
assert len(mock_is_installed.mock_calls) == 1
assert sorted(mock_call[1][0] for mock_call in mock_is_installed.mock_calls) == [
"test-comp2==1.0.0",
]
assert len(mock_inst.mock_calls) == 1
assert sorted(mock_call[1][0] for mock_call in mock_inst.mock_calls) == [
"test-comp2==1.0.0",
]
# The dependent integrations should not be fetched again
assert len(mock_async_get_integration.mock_calls) == 1
assert sorted(
mock_call[1][1] for mock_call in mock_async_get_integration.mock_calls
) == [
"test_component2",
]
async def test_get_integration_with_requirements_concurrency(
hass: HomeAssistant,
) -> None:
"""Test that we don't install the same requirement concurrently."""
hass.config.skip_pip = False
mock_integration(
hass, MockModule("test_component_dep", requirements=["test-comp-dep==1.0.0"])
)
process_integration_calls = 0
async def _async_process_integration_slowed(*args, **kwargs):
nonlocal process_integration_calls
process_integration_calls += 1
await asyncio.sleep(0)
manager = _async_get_manager(hass)
with patch.object(
manager, "_async_process_integration", _async_process_integration_slowed
):
tasks = [
async_get_integration_with_requirements(hass, "test_component_dep")
for _ in range(10)
]
results = await asyncio.gather(*tasks)
assert all(result.domain == "test_component_dep" for result in results)
assert process_integration_calls == 1
async def test_get_integration_with_requirements_pip_install_fails_two_passes(
hass: HomeAssistant,
) -> None:
"""Check getting an integration with loaded requirements and the pip install fails two passes."""
hass.config.skip_pip = False
mock_integration(
hass, MockModule("test_component_dep", requirements=["test-comp-dep==1.0.0"])
)
mock_integration(
hass,
MockModule(
"test_component_after_dep", requirements=["test-comp-after-dep==1.0.0"]
),
)
mock_integration(
hass,
MockModule(
"test_component",
requirements=["test-comp==1.0.0"],
dependencies=["test_component_dep"],
partial_manifest={"after_dependencies": ["test_component_after_dep"]},
),
)
def _mock_install_package(package, **kwargs):
if package == "test-comp==1.0.0":
return True
return False
# 1st pass
with (
pytest.raises(RequirementsNotFound),
patch(
"homeassistant.util.package.is_installed", return_value=False
) as mock_is_installed,
patch(
"homeassistant.util.package.install_package",
side_effect=_mock_install_package,
) as mock_inst,
):
integration = await async_get_integration_with_requirements(
hass, "test_component"
)
assert integration
assert integration.domain == "test_component"
assert len(mock_is_installed.mock_calls) == 3
assert sorted(mock_call[1][0] for mock_call in mock_is_installed.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp==1.0.0",
]
assert len(mock_inst.mock_calls) == 7
assert sorted(mock_call[1][0] for mock_call in mock_inst.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-after-dep==1.0.0",
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp==1.0.0",
]
# 2nd pass
with (
pytest.raises(RequirementsNotFound),
patch(
"homeassistant.util.package.is_installed", return_value=False
) as mock_is_installed,
patch(
"homeassistant.util.package.install_package",
side_effect=_mock_install_package,
) as mock_inst,
):
integration = await async_get_integration_with_requirements(
hass, "test_component"
)
assert integration
assert integration.domain == "test_component"
assert len(mock_is_installed.mock_calls) == 0
# On another attempt we remember failures and don't try again
assert len(mock_inst.mock_calls) == 0
# Now clear the history and so we try again
async_clear_install_history(hass)
with (
pytest.raises(RequirementsNotFound),
patch(
"homeassistant.util.package.is_installed", return_value=False
) as mock_is_installed,
patch(
"homeassistant.util.package.install_package",
side_effect=_mock_install_package,
) as mock_inst,
):
integration = await async_get_integration_with_requirements(
hass, "test_component"
)
assert integration
assert integration.domain == "test_component"
assert len(mock_is_installed.mock_calls) == 2
assert sorted(mock_call[1][0] for mock_call in mock_is_installed.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
]
assert len(mock_inst.mock_calls) == 6
assert sorted(mock_call[1][0] for mock_call in mock_inst.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-after-dep==1.0.0",
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp-dep==1.0.0",
"test-comp-dep==1.0.0",
]
# Now clear the history and mock success
async_clear_install_history(hass)
with (
patch(
"homeassistant.util.package.is_installed", return_value=False
) as mock_is_installed,
patch(
"homeassistant.util.package.install_package", return_value=True
) as mock_inst,
):
integration = await async_get_integration_with_requirements(
hass, "test_component"
)
assert integration
assert integration.domain == "test_component"
assert len(mock_is_installed.mock_calls) == 2
assert sorted(mock_call[1][0] for mock_call in mock_is_installed.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
]
assert len(mock_inst.mock_calls) == 2
assert sorted(mock_call[1][0] for mock_call in mock_inst.mock_calls) == [
"test-comp-after-dep==1.0.0",
"test-comp-dep==1.0.0",
]
async def test_get_integration_with_missing_dependencies(hass: HomeAssistant) -> None:
"""Check getting an integration with missing dependencies."""
hass.config.skip_pip = False
mock_integration(
hass,
MockModule("test_component_after_dep"),
)
mock_integration(
hass,
MockModule(
"test_component",
dependencies=["test_component_dep"],
partial_manifest={"after_dependencies": ["test_component_after_dep"]},
),
)
mock_integration(
hass,
MockModule(
"test_custom_component",
dependencies=["test_component_dep"],
partial_manifest={"after_dependencies": ["test_component_after_dep"]},
),
built_in=False,
)
with pytest.raises(loader.IntegrationNotFound):
await async_get_integration_with_requirements(hass, "test_component")
with pytest.raises(loader.IntegrationNotFound):
await async_get_integration_with_requirements(hass, "test_custom_component")
async def test_get_built_in_integration_with_missing_after_dependencies(
hass: HomeAssistant,
) -> None:
"""Check getting a built_in integration with missing after_dependencies results in exception."""
hass.config.skip_pip = False
mock_integration(
hass,
MockModule(
"test_component",
partial_manifest={"after_dependencies": ["test_component_after_dep"]},
),
built_in=True,
)
with pytest.raises(loader.IntegrationNotFound):
await async_get_integration_with_requirements(hass, "test_component")
async def test_get_custom_integration_with_missing_after_dependencies(
hass: HomeAssistant,
) -> None:
"""Check getting a custom integration with missing after_dependencies."""
hass.config.skip_pip = False
mock_integration(
hass,
MockModule(
"test_custom_component",
partial_manifest={"after_dependencies": ["test_component_after_dep"]},
),
built_in=False,
)
integration = await async_get_integration_with_requirements(
hass, "test_custom_component"
)
assert integration
assert integration.domain == "test_custom_component"
async def test_install_with_wheels_index(hass: HomeAssistant) -> None:
"""Test an install attempt with wheels index URL."""
hass.config.skip_pip = False
mock_integration(hass, MockModule("comp", requirements=["hello==1.0.0"]))
with (
patch("homeassistant.util.package.is_installed", return_value=False),
patch("homeassistant.util.package.is_docker_env", return_value=True),
patch("homeassistant.util.package.install_package") as mock_inst,
patch.dict(os.environ, {"WHEELS_LINKS": "https://wheels.hass.io/test"}),
patch(
"os.path.dirname",
) as mock_dir,
):
mock_dir.return_value = "ha_package_path"
assert await setup.async_setup_component(hass, "comp", {})
assert "comp" in hass.config.components
assert mock_inst.call_args == call(
"hello==1.0.0",
constraints=os.path.join("ha_package_path", CONSTRAINT_FILE),
timeout=60,
)
async def test_install_on_docker(hass: HomeAssistant) -> None:
"""Test an install attempt on an docker system env."""
hass.config.skip_pip = False
mock_integration(hass, MockModule("comp", requirements=["hello==1.0.0"]))
with (
patch("homeassistant.util.package.is_installed", return_value=False),
patch("homeassistant.util.package.is_docker_env", return_value=True),
patch("homeassistant.util.package.install_package") as mock_inst,
patch("os.path.dirname") as mock_dir,
patch.dict(os.environ, env_without_wheel_links(), clear=True),
):
mock_dir.return_value = "ha_package_path"
assert await setup.async_setup_component(hass, "comp", {})
assert "comp" in hass.config.components
assert mock_inst.call_args == call(
"hello==1.0.0",
constraints=os.path.join("ha_package_path", CONSTRAINT_FILE),
timeout=60,
)
async def test_discovery_requirements_mqtt(hass: HomeAssistant) -> None:
"""Test that we load discovery requirements."""
hass.config.skip_pip = False
mqtt = await loader.async_get_integration(hass, "mqtt")
mock_integration(
hass, MockModule("mqtt_comp", partial_manifest={"mqtt": ["foo/discovery"]})
)
with patch(
"homeassistant.requirements.RequirementsManager.async_process_requirements",
) as mock_process:
await async_get_integration_with_requirements(hass, "mqtt_comp")
assert len(mock_process.mock_calls) == 1
assert mock_process.mock_calls[0][1][1] == mqtt.requirements
async def test_discovery_requirements_ssdp(hass: HomeAssistant) -> None:
"""Test that we load discovery requirements."""
hass.config.skip_pip = False
ssdp = await loader.async_get_integration(hass, "ssdp")
mock_integration(
hass, MockModule("ssdp_comp", partial_manifest={"ssdp": [{"st": "roku:ecp"}]})
)
with patch(
"homeassistant.requirements.RequirementsManager.async_process_requirements",
) as mock_process:
await async_get_integration_with_requirements(hass, "ssdp_comp")
assert len(mock_process.mock_calls) == 3
assert mock_process.mock_calls[0][1][1] == ssdp.requirements
assert {
mock_process.mock_calls[1][1][0],
mock_process.mock_calls[2][1][0],
} == {"network", "recorder"}
@pytest.mark.parametrize(
"partial_manifest",
[{"zeroconf": ["_googlecast._tcp.local."]}, {"homekit": {"models": ["LIFX"]}}],
)
async def test_discovery_requirements_zeroconf(
hass: HomeAssistant, partial_manifest
) -> None:
"""Test that we load discovery requirements."""
hass.config.skip_pip = False
zeroconf = await loader.async_get_integration(hass, "zeroconf")
mock_integration(
hass,
MockModule("comp", partial_manifest=partial_manifest),
)
with patch(
"homeassistant.requirements.RequirementsManager.async_process_requirements",
) as mock_process:
await async_get_integration_with_requirements(hass, "comp")
assert len(mock_process.mock_calls) == 3
assert mock_process.mock_calls[0][1][1] == zeroconf.requirements
async def test_discovery_requirements_dhcp(hass: HomeAssistant) -> None:
"""Test that we load dhcp discovery requirements."""
hass.config.skip_pip = False
dhcp = await loader.async_get_integration(hass, "dhcp")
mock_integration(
hass,
MockModule(
"comp",
partial_manifest={
"dhcp": [{"hostname": "somfy_*", "macaddress": "B8B7F1*"}]
},
),
)
with patch(
"homeassistant.requirements.RequirementsManager.async_process_requirements",
) as mock_process:
await async_get_integration_with_requirements(hass, "comp")
assert len(mock_process.mock_calls) == 1 # dhcp does not depend on http
assert mock_process.mock_calls[0][1][1] == dhcp.requirements