249 lines
7.9 KiB
Python
249 lines
7.9 KiB
Python
"""Compatibility baseline for the published API and LocalThings consumer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
|
|
from smartthings_local.ocf.observe_refresh import ObserveRefreshTask
|
|
from smartthings_local.ocf.state_cache import StateCache
|
|
from smartthings_local.protocol.auth import (
|
|
AuthenticationProvider,
|
|
CertificateAuth,
|
|
PskAuth,
|
|
SamsungServerProfile,
|
|
SamsungServerRole,
|
|
ServerCertificateAuth,
|
|
)
|
|
from smartthings_local.protocol.dtls_session import (
|
|
ConnectCancellation,
|
|
DtlsCoapSession,
|
|
)
|
|
from smartthings_local.protocol.ocf_multicast import (
|
|
OcfResponderPortDiscoveryResult,
|
|
discover_ocf_responder_ports,
|
|
)
|
|
from smartthings_local.protocol.owner_psk import derive_mfg_certificate_owner_psk
|
|
|
|
|
|
def _assert_compatible_signature(callable_object, expected: list[str]) -> None:
|
|
"""Require the existing call surface while allowing safe extensions."""
|
|
parameters = list(inspect.signature(callable_object).parameters.values())
|
|
assert [parameter.name for parameter in parameters[: len(expected)]] == expected
|
|
for parameter in parameters[len(expected) :]:
|
|
assert (
|
|
parameter.kind
|
|
in (
|
|
inspect.Parameter.VAR_POSITIONAL,
|
|
inspect.Parameter.VAR_KEYWORD,
|
|
)
|
|
or parameter.default is not inspect.Parameter.empty
|
|
)
|
|
|
|
|
|
def test_dtls_session_constructor_keeps_file_memory_and_local_port_inputs():
|
|
_assert_compatible_signature(
|
|
DtlsCoapSession,
|
|
[
|
|
"host",
|
|
"port",
|
|
"cert_path",
|
|
"key_path",
|
|
"cert_pem",
|
|
"key_pem",
|
|
"on_notification",
|
|
"mtu",
|
|
"rate_limit_rps",
|
|
"local_port",
|
|
],
|
|
)
|
|
auth_parameter = inspect.signature(DtlsCoapSession).parameters["auth"]
|
|
assert auth_parameter.kind is inspect.Parameter.KEYWORD_ONLY
|
|
assert auth_parameter.default is None
|
|
|
|
|
|
def test_known_host_multicast_discovery_has_a_bounded_explicit_interface_api():
|
|
parameters = inspect.signature(discover_ocf_responder_ports).parameters
|
|
assert list(parameters) == [
|
|
"target_address",
|
|
"interface_address",
|
|
"discovery_port",
|
|
"timeout",
|
|
"rounds",
|
|
]
|
|
assert parameters["target_address"].default is inspect.Parameter.empty
|
|
for name in ("interface_address", "discovery_port", "timeout", "rounds"):
|
|
assert parameters[name].kind is inspect.Parameter.KEYWORD_ONLY
|
|
assert parameters["interface_address"].default is inspect.Parameter.empty
|
|
assert parameters["discovery_port"].default == 5683
|
|
assert parameters["timeout"].default == 3.0
|
|
assert parameters["rounds"].default == 2
|
|
|
|
result = OcfResponderPortDiscoveryResult(
|
|
ports=(43123,),
|
|
attempts=2,
|
|
responses=1,
|
|
)
|
|
assert result.found is True
|
|
assert result.ports == (43123,)
|
|
|
|
|
|
def test_certificate_auth_is_a_public_authentication_provider():
|
|
provider = CertificateAuth.from_files("/synthetic/cert.pem", "/synthetic/key")
|
|
assert isinstance(provider, AuthenticationProvider)
|
|
|
|
for factory in (CertificateAuth.from_files, CertificateAuth.from_memory):
|
|
profile_parameter = inspect.signature(factory).parameters["server_profile"]
|
|
assert profile_parameter.kind is inspect.Parameter.KEYWORD_ONLY
|
|
assert profile_parameter.default is None
|
|
|
|
|
|
def test_samsung_server_profile_is_public_and_explicitly_bound():
|
|
parameters = inspect.signature(SamsungServerProfile.bound_device).parameters
|
|
assert list(parameters) == [
|
|
"expected_certificate_identity",
|
|
"role",
|
|
"additional_ca_pem",
|
|
]
|
|
assert (
|
|
parameters["expected_certificate_identity"].default
|
|
is inspect.Parameter.empty
|
|
)
|
|
assert parameters["role"].kind is inspect.Parameter.KEYWORD_ONLY
|
|
assert parameters["role"].default is SamsungServerRole.HOME_APPLIANCE
|
|
assert parameters["additional_ca_pem"].kind is inspect.Parameter.KEYWORD_ONLY
|
|
assert parameters["additional_ca_pem"].default is None
|
|
|
|
|
|
def test_server_certificate_auth_is_a_public_authentication_provider():
|
|
profile = SamsungServerProfile.bound_device(
|
|
"abababab-abab-abab-abab-abababababab",
|
|
role=SamsungServerRole.VD_DEVICE,
|
|
)
|
|
provider = ServerCertificateAuth(server_profile=profile)
|
|
assert isinstance(provider, AuthenticationProvider)
|
|
session = DtlsCoapSession("device.example", 5684, auth=provider)
|
|
assert session.auth is provider
|
|
assert session.cert_path is None
|
|
assert session.key_path is None
|
|
assert session.cert_pem is None
|
|
assert session.key_pem is None
|
|
parameters = inspect.signature(ServerCertificateAuth).parameters
|
|
assert list(parameters) == ["server_profile"]
|
|
assert parameters["server_profile"].kind is inspect.Parameter.KEYWORD_ONLY
|
|
assert parameters["server_profile"].default is inspect.Parameter.empty
|
|
|
|
|
|
def test_psk_auth_is_a_public_authentication_provider():
|
|
provider = PskAuth(identity=b"i" * 16, key=b"k" * 16)
|
|
assert isinstance(provider, AuthenticationProvider)
|
|
parameters = inspect.signature(PskAuth).parameters
|
|
assert list(parameters) == ["identity", "key"]
|
|
assert all(
|
|
parameter.kind is inspect.Parameter.KEYWORD_ONLY
|
|
and parameter.default is inspect.Parameter.empty
|
|
for parameter in parameters.values()
|
|
)
|
|
|
|
|
|
def test_owner_psk_derivation_keeps_every_security_input_explicit():
|
|
parameters = inspect.signature(
|
|
derive_mfg_certificate_owner_psk
|
|
).parameters
|
|
assert list(parameters) == [
|
|
"master_secret",
|
|
"client_random",
|
|
"server_random",
|
|
"owner_uuid",
|
|
"device_uuid",
|
|
"cipher_name",
|
|
"oxm_label",
|
|
]
|
|
assert all(
|
|
parameter.kind is inspect.Parameter.KEYWORD_ONLY
|
|
and parameter.default is inspect.Parameter.empty
|
|
for parameter in parameters.values()
|
|
)
|
|
|
|
|
|
def test_dtls_session_keeps_current_consumer_methods():
|
|
expected = {
|
|
"close",
|
|
"connect",
|
|
"get",
|
|
"join",
|
|
"pace",
|
|
"ping",
|
|
"post",
|
|
"refresh_observes",
|
|
"start_reader",
|
|
"subscribe",
|
|
}
|
|
assert expected <= set(dir(DtlsCoapSession))
|
|
assert "abort" not in DtlsCoapSession.__dict__
|
|
assert "quiesce_for_close" not in DtlsCoapSession.__dict__
|
|
_assert_compatible_signature(DtlsCoapSession.connect, ["self"])
|
|
connect_timeout = inspect.signature(DtlsCoapSession.connect).parameters[
|
|
"timeout"
|
|
]
|
|
assert connect_timeout.kind is inspect.Parameter.KEYWORD_ONLY
|
|
assert connect_timeout.default is None
|
|
connect_cancel = inspect.signature(DtlsCoapSession.connect).parameters[
|
|
"cancel"
|
|
]
|
|
assert connect_cancel.kind is inspect.Parameter.KEYWORD_ONLY
|
|
assert connect_cancel.default is None
|
|
assert callable(ConnectCancellation().set)
|
|
_assert_compatible_signature(
|
|
DtlsCoapSession.get,
|
|
[
|
|
"self",
|
|
"path_segs",
|
|
"query",
|
|
"timeout",
|
|
],
|
|
)
|
|
_assert_compatible_signature(
|
|
DtlsCoapSession.post,
|
|
[
|
|
"self",
|
|
"path_segs",
|
|
"body_cbor",
|
|
"timeout",
|
|
],
|
|
)
|
|
_assert_compatible_signature(
|
|
DtlsCoapSession.subscribe,
|
|
["self", "path_segs"],
|
|
)
|
|
|
|
|
|
def test_state_cache_keeps_current_consumer_surface():
|
|
_assert_compatible_signature(StateCache, ["descriptor"])
|
|
expected = {
|
|
"apply_optimistic",
|
|
"apply_rep",
|
|
"freshness_s",
|
|
"get",
|
|
"index_device_tree",
|
|
"set_on_change",
|
|
"snapshot",
|
|
"stalest",
|
|
}
|
|
assert expected <= set(dir(StateCache))
|
|
|
|
|
|
def test_observe_refresh_task_keeps_current_consumer_surface():
|
|
_assert_compatible_signature(
|
|
ObserveRefreshTask,
|
|
[
|
|
"session",
|
|
"paths",
|
|
"interval_s",
|
|
"logger",
|
|
],
|
|
)
|
|
_assert_compatible_signature(
|
|
ObserveRefreshTask.run_forever,
|
|
["self", "stop"],
|
|
)
|