SmartThings-Local/tests/test_public_api_contract.py

"""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.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_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"],
    )