jeans/supervisor
jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-07-24 09:36:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Network connection check only about ipv4 (#3830)

Browse Source 
* Network connection test only about ipv4

* Better test of change being made
This commit is contained in:
Mike Degatano 2022-08-30 02:39:08 -04:00 committed by GitHub
parent 3dc36c3402
commit 4ea27f6311
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 37 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
supervisor/host/network.py
View File

@ -32,7 +32,7 @@ from ..exceptions import (
)
from ..jobs.const import JobCondition
from ..jobs.decorator import Job
from ..resolution.checks.network_interface import CheckNetworkInterface
from ..resolution.checks.network_interface_ipv4 import CheckNetworkInterfaceIPV4
from .const import AuthMethod, InterfaceMethod, InterfaceType, WifiMode
_LOGGER: logging.Logger = logging.getLogger(__name__)
@ -113,7 +113,7 @@ class NetworkManager(CoreSysAttributes):
interfaces = [
Interface.from_dbus_interface(interface)
for interface in self.sys_dbus.network.interfaces.values()
if not CheckNetworkInterface.check_interface(interface)
if not CheckNetworkInterfaceIPV4.check_interface(interface)
]
with suppress(HostNetworkNotFound):
await asyncio.gather(

15
supervisor/resolution/checks/network_interface.py → supervisor/resolution/checks/network_interface_ipv4.py
View File

@ -9,18 +9,18 @@ from .base import CheckBase
def setup(coresys: CoreSys) -> CheckBase:
"""Check setup function."""
return CheckNetworkInterface(coresys)
return CheckNetworkInterfaceIPV4(coresys)
class CheckNetworkInterface(CheckBase):
"""CheckNetworkInterface class for check."""
class CheckNetworkInterfaceIPV4(CheckBase):
"""CheckNetworkInterfaceIPV4 class for check."""
async def run_check(self) -> None:
"""Run check if not affected by issue."""
for interface in self.sys_dbus.network.interfaces.values():
if CheckNetworkInterface.check_interface(interface):
if CheckNetworkInterfaceIPV4.check_interface(interface):
self.sys_resolution.create_issue(
IssueType.NETWORK_CONNECTION_PROBLEM,
IssueType.IPV4_CONNECTION_PROBLEM,
ContextType.SYSTEM,
interface.name,
)
@ -32,7 +32,7 @@ class CheckNetworkInterface(CheckBase):
interface = self.sys_dbus.network.interfaces.get(reference)
return interface and CheckNetworkInterface.check_interface(interface)
return interface and CheckNetworkInterfaceIPV4.check_interface(interface)
@staticmethod
def check_interface(interface: NetworkInterface) -> bool:
@ -44,13 +44,12 @@ class CheckNetworkInterface(CheckBase):
interface.connection.state
in [ConnectionStateType.ACTIVATED, ConnectionStateType.ACTIVATING]
and ConnectionStateFlags.IP4_READY in interface.connection.state_flags
and ConnectionStateFlags.IP6_READY in interface.connection.state_flags
)
@property
def issue(self) -> IssueType:
"""Return a IssueType enum."""
return IssueType.NETWORK_CONNECTION_PROBLEM
return IssueType.IPV4_CONNECTION_PROBLEM
@property
def context(self) -> ContextType:

2
supervisor/resolution/const.py
View File

@ -72,8 +72,8 @@ class IssueType(str, Enum):
DOCKER_RATELIMIT = "docker_ratelimit"
FATAL_ERROR = "fatal_error"
FREE_SPACE = "free_space"
IPV4_CONNECTION_PROBLEM = "ipv4_connection_problem"
MISSING_IMAGE = "missing_image"
NETWORK_CONNECTION_PROBLEM = "network_connection_problem"
PWNED = "pwned"
SECURITY = "security"
TRUST = "trust"

48
tests/resolution/check/test_check_network_interface.py → tests/resolution/check/test_check_network_interface_ipv4.py
View File

@ -6,29 +6,35 @@ import pytest
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.dbus.const import ConnectionStateFlags
from supervisor.resolution.checks.network_interface import CheckNetworkInterface
from supervisor.resolution.checks.network_interface_ipv4 import (
CheckNetworkInterfaceIPV4,
)
from supervisor.resolution.const import ContextType, IssueType
from supervisor.resolution.data import Issue
TEST_ISSUE = Issue(IssueType.IPV4_CONNECTION_PROBLEM, ContextType.SYSTEM, "eth0")
async def test_base(coresys: CoreSys):
"""Test check basics."""
network_interface = CheckNetworkInterface(coresys)
assert network_interface.slug == "network_interface"
network_interface = CheckNetworkInterfaceIPV4(coresys)
assert network_interface.slug == "network_interface_ipv4"
assert network_interface.enabled
@pytest.mark.parametrize(
"state_flags",
"state_flags,issues",
[
{ConnectionStateFlags.IP4_READY},
{ConnectionStateFlags.IP6_READY},
{ConnectionStateFlags.NONE},
({ConnectionStateFlags.IP4_READY}, []),
({ConnectionStateFlags.IP6_READY}, [TEST_ISSUE]),
({ConnectionStateFlags.NONE}, [TEST_ISSUE]),
],
)
async def test_check(coresys: CoreSys, state_flags: set[ConnectionStateFlags]):
async def test_check(
coresys: CoreSys, state_flags: set[ConnectionStateFlags], issues: list[Issue]
):
"""Test check."""
network_interface = CheckNetworkInterface(coresys)
network_interface = CheckNetworkInterfaceIPV4(coresys)
coresys.core.state = CoreState.RUNNING
assert len(coresys.resolution.issues) == 0
@ -42,22 +48,22 @@ async def test_check(coresys: CoreSys, state_flags: set[ConnectionStateFlags]):
):
await network_interface.run_check()
assert coresys.resolution.issues == [
Issue(IssueType.NETWORK_CONNECTION_PROBLEM, ContextType.SYSTEM, "eth0")
]
assert coresys.resolution.issues == issues
@pytest.mark.parametrize(
"state_flags",
"state_flags,approved",
[
{ConnectionStateFlags.IP4_READY},
{ConnectionStateFlags.IP6_READY},
{ConnectionStateFlags.NONE},
({ConnectionStateFlags.IP4_READY}, False),
({ConnectionStateFlags.IP6_READY}, True),
({ConnectionStateFlags.NONE}, True),
],
)
async def test_approve(coresys: CoreSys, state_flags: set[ConnectionStateFlags]):
async def test_approve(
coresys: CoreSys, state_flags: set[ConnectionStateFlags], approved: bool
):
"""Test check."""
network_interface = CheckNetworkInterface(coresys)
network_interface = CheckNetworkInterfaceIPV4(coresys)
coresys.core.state = CoreState.RUNNING
assert not await network_interface.approve_check("eth0")
@ -66,19 +72,19 @@ async def test_approve(coresys: CoreSys, state_flags: set[ConnectionStateFlags])
"supervisor.dbus.network.connection.NetworkConnection.state_flags",
new=PropertyMock(return_value=state_flags),
):
assert await network_interface.approve_check("eth0")
assert await network_interface.approve_check("eth0") is approved
async def test_did_run(coresys: CoreSys):
"""Test that the check ran as expected."""
network_interface = CheckNetworkInterface(coresys)
network_interface = CheckNetworkInterfaceIPV4(coresys)
should_run = network_interface.states
should_not_run = [state for state in CoreState if state not in should_run]
assert len(should_run) != 0
assert len(should_not_run) != 0
with patch(
"supervisor.resolution.checks.network_interface.CheckNetworkInterface.run_check",
"supervisor.resolution.checks.network_interface_ipv4.CheckNetworkInterfaceIPV4.run_check",
return_value=None,
) as check:
for state in should_run: