diff --git a/snap/src/charmlibs/snap/__init__.py b/snap/src/charmlibs/snap/__init__.py index 4865ba3f0..d827f6f37 100644 --- a/snap/src/charmlibs/snap/__init__.py +++ b/snap/src/charmlibs/snap/__init__.py @@ -63,6 +63,10 @@ ensure, ensure_revision, ) +from ._snapd_interfaces import ( + connect, + disconnect, +) from ._snapd_logs import ( LogEntry, logs, @@ -94,6 +98,8 @@ 'OptionNotFoundError', 'RevisionNotAvailableError', 'TimeoutError', + 'connect', + 'disconnect', 'ensure', 'ensure_revision', 'hold', diff --git a/snap/src/charmlibs/snap/_snapd_interfaces.py b/snap/src/charmlibs/snap/_snapd_interfaces.py new file mode 100644 index 000000000..50cbb9b75 --- /dev/null +++ b/snap/src/charmlibs/snap/_snapd_interfaces.py @@ -0,0 +1,108 @@ +# Copyright 2026 Canonical Ltd. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Snap interface operations, implemented as calls to the snapd API's /v2/interfaces endpoint.""" + +from __future__ import annotations + +from typing import Any + +from . import _client, _errors + +# /v2/interfaces + + +def connect( + plug_snap: str, plug: str, slot_snap: str | None = None, slot: str | None = None, / +) -> None: + """Connect a snap and plug, to a target snap and slot. + + Connecting an already-connected plug and slot succeeds silently. + + Args: + plug_snap: The name of the snap providing the plug. + plug: The name of the plug on ``plug_snap``. + slot_snap: The name of the snap providing the slot. If omitted, snapd auto-resolves + the slot, typically to the system snap (``snapd`` or ``core``). + slot: The name of the slot on ``slot_snap``. If omitted, snapd auto-resolves it. + + Raises: + APIError: if the plug snap or slot snap is not installed, or the named plug or slot + does not exist. The error has an empty ``kind``; inspect ``message`` for details. + ChangeError: if the connection fails after starting (for example, an interface hook + errors). + """ + data = { + 'action': 'connect', + 'plugs': [{'snap': plug_snap, 'plug': plug}], + 'slots': [{'snap': slot_snap or '', 'slot': slot or ''}], + } + _client.post('/v2/interfaces', body=data) + + +def disconnect( + plug_or_slot_snap: str, + plug_or_slot: str, + slot_snap: str | None = None, + slot: str | None = None, + /, + *, + forget: bool = False, +) -> None: + """Disconnect a plug from a slot. + + May be called in two forms: + + - ``(snap, plug_or_slot)`` disconnects everything connected to the named plug or slot + on ``snap``. + - ``(plug_snap, plug, slot_snap[, slot])`` disconnects the plug from the slot. ``slot`` + may be omitted to disconnect the plug from any slot on ``slot_snap``. + + Disconnecting a plug and slot that are not connected is a no-op and does not raise + (the underlying ``interfaces-unchanged`` error is suppressed, mirroring the snap CLI). + + Args: + plug_or_slot_snap: The snap providing the plug (explicit form) or the snap providing + the plug or slot to disconnect (two-argument form). + plug_or_slot: The plug on ``plug_or_slot_snap`` (explicit form) or the plug or slot + name to disconnect (two-argument form). + slot_snap: The snap providing the slot. Omit for the two-argument form. + slot: The slot on ``slot_snap``. May be omitted to match any slot on ``slot_snap``. + forget: If ``True``, also forget any manual connection preference, so the interface + is not automatically reconnected on the next refresh. + + Raises: + APIError: if a named snap is not installed, or the named plug or slot does not exist. + The error has an empty ``kind``; inspect ``message`` for details. + ChangeError: if the disconnection fails after starting (for example, an interface hook + errors). + """ + data: dict[str, Any] = {'action': 'disconnect'} + if slot_snap is None: + assert slot is None + # Called with 2 arguments, treat as `snap disconnect :`. + data['plugs'] = [{'snap': '', 'plug': ''}] + data['slots'] = [{'snap': plug_or_slot_snap, 'slot': plug_or_slot}] + else: + # Called with 3 or 4 arguments, treat as `snap disconnect : :`. + data['plugs'] = [{'snap': plug_or_slot_snap, 'plug': plug_or_slot}] + data['slots'] = [{'snap': slot_snap, 'slot': slot or ''}] + if forget: + data['forget'] = True + # NOTE: Unlike connect, the API raises interfaces-unchanged if already disconnected. + # We suppress this to make disconnect symmetric with connect (following the snap CLI). + try: + _client.post('/v2/interfaces', body=data) + except _errors._InterfacesUnchangedError: + pass # Follow the snap CLI's lead and suppress this error. diff --git a/snap/tests/functional/test_snapd_interfaces.py b/snap/tests/functional/test_snapd_interfaces.py new file mode 100644 index 000000000..7bd1f821e --- /dev/null +++ b/snap/tests/functional/test_snapd_interfaces.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +"""Functional tests for _snapd_interfaces: connect, disconnect.""" + +from __future__ import annotations + +import dataclasses +import typing +from typing import Any + +import pytest + +from charmlibs.snap import _client, _errors, _snapd_interfaces +from conftest import ensure_installed + +_SNAP = 'htop' +_PLUG = 'mount-observe' +# snapd auto-resolves the mount-observe slot to snapd. +_SLOT_SNAP = 'snapd' +_SLOT = 'mount-observe' + +# A snap name that is never installed — used for error paths where any absent +# snap produces the same error response, avoiding unnecessary remove operations. +_ABSENT_SNAP = 'this-snap-does-not-exist-xyz-abc-123' + + +# --------------------------------------------------------------------------- +# Test helpers and possible future candidates for library public API. +# --------------------------------------------------------------------------- + + +@dataclasses.dataclass +class _Plug: + interface: str + plug: str + + +@dataclasses.dataclass +class _Slot: + interface: str + slot: str + + +def _list_plugs(snap: str, connected_only: bool = False) -> list[_Plug]: + interfaces = _list_interfaces(snap, connected_only=connected_only) + return [ + _Plug(interface=i['name'], plug=p['plug']) + for i in interfaces + for p in i.get('plugs', []) + if p['snap'] == snap + ] + + +def _list_slots(snap: str, connected_only: bool = False) -> list[_Slot]: + interfaces = _list_interfaces(snap, connected_only=connected_only) + return [ + _Slot(interface=i['name'], slot=s['slot']) + for i in interfaces + for s in i.get('slots', []) + if s['snap'] == snap + ] + + +def _list_interfaces( + snap: str | None = None, connected_only: bool = False +) -> list[dict[str, Any]]: + """List snap interfaces.""" + query = {'select': 'connected' if connected_only else 'all', 'slots': 'true', 'plugs': 'true'} + interfaces = _client.get('/v2/interfaces', query=query) + assert isinstance(interfaces, list) + interfaces = typing.cast('list[dict[str, Any]]', interfaces) + if snap is None: + return interfaces + return [ + i + for i in interfaces + if any(p['snap'] == snap for p in i.get('plugs', [])) + or any(s['snap'] == snap for s in i.get('slots', [])) + ] + + +# --------------------------------------------------------------------------- + + +def _is_connected() -> bool: + return any(p.plug == _PLUG for p in _list_plugs(_SNAP, connected_only=True)) + + +def _ensure_disconnected() -> None: + try: + _snapd_interfaces.disconnect(_SNAP, _PLUG) + except Exception: # noqa: S110 + pass + # Post-condition: the plug really is no longer connected. + assert not any(p.plug == _PLUG for p in _list_plugs(_SNAP, connected_only=True)) + + +def _ensure_connected() -> None: + # Pre-condition: the slot side (snapd) actually provides the mount-observe slot, + # otherwise the connection could never succeed. + assert any(s.slot == _SLOT for s in _list_slots(_SLOT_SNAP)) + if not _is_connected(): + _snapd_interfaces.connect(_SNAP, _PLUG) + # Post-condition: the plug is now connected. + assert _is_connected() + + +# --------------------------------------------------------------------------- +# connect +# --------------------------------------------------------------------------- + + +def test_connect(): + ensure_installed(_SNAP) + _ensure_disconnected() + assert not _is_connected() + _snapd_interfaces.connect(_SNAP, _PLUG) + assert _is_connected() + + +def test_connect_already_connected_no_error(): + # Connecting an already-connected plug should not raise. + ensure_installed(_SNAP) + _ensure_connected() + assert _is_connected() + _snapd_interfaces.connect(_SNAP, _PLUG) # Should not raise. + assert _is_connected() + + +def test_connect_nonexistent_plug_raises(): + # Connecting a nonexistent plug raises a base Error (no kind from snapd). + ensure_installed(_SNAP) + with pytest.raises(_errors.Error) as ctx: + _snapd_interfaces.connect(_SNAP, 'nonexistent-plug') + assert not ctx.value.kind + assert 'nonexistent-plug' in ctx.value.message + + +def test_connect_with_explicit_slot(): + # connect() accepts an explicit slot snap and slot name. + ensure_installed(_SNAP) + _ensure_disconnected() + _snapd_interfaces.connect(_SNAP, _PLUG, _SLOT_SNAP, _SLOT) + assert _is_connected() + + +# --------------------------------------------------------------------------- +# disconnect +# --------------------------------------------------------------------------- + + +def test_disconnect(): + ensure_installed(_SNAP) + _ensure_connected() + assert _is_connected() + _snapd_interfaces.disconnect(_SNAP, _PLUG) + assert not _is_connected() + + +def test_disconnect_not_connected_no_error(): + # Disconnecting a plug that is not connected is a no-op (interfaces-unchanged suppressed). + # This makes disconnect symmetric with connect: both succeed silently when no change needed. + ensure_installed(_SNAP) + _ensure_disconnected() + _snapd_interfaces.disconnect(_SNAP, _PLUG) # Should not raise. + assert not _is_connected() + + +def test_disconnect_forget_connected_no_error(): + # disconnect forget=True on a connected interface works without error. + ensure_installed(_SNAP) + _ensure_connected() + _snapd_interfaces.disconnect(_SNAP, _PLUG, forget=True) # Should not raise. + + +def test_disconnect_forget_not_connected_no_error(): + # disconnect forget=True on a not-connected interface is a no-op + # (interfaces-unchanged suppressed, same as without forget=True). + ensure_installed(_SNAP) + _ensure_disconnected() + _snapd_interfaces.disconnect(_SNAP, _PLUG, forget=True) # Should not raise. + + +def test_disconnect_nonexistent_plug_or_slot_raises(): + # disconnect: plug/slot name doesn't exist on the installed snap. + ensure_installed(_SNAP) + with pytest.raises(_errors.APIError) as ctx: + _snapd_interfaces.disconnect(_SNAP, 'nonexistent-slot') + assert not ctx.value.kind + assert 'no plug or slot named' in ctx.value.message + + +# --------------------------------------------------------------------------- +# not-installed snap (uses a never-installed name to avoid churn) +# --------------------------------------------------------------------------- + + +def test_connect_not_installed_snap_raises(): + with pytest.raises(_errors.APIError) as ctx: + _snapd_interfaces.connect(_ABSENT_SNAP, 'home') + assert not ctx.value.kind + assert 'not installed' in ctx.value.message + + +def test_disconnect_not_installed_snap_raises(): + with pytest.raises(_errors.APIError) as ctx: + _snapd_interfaces.disconnect(_ABSENT_SNAP, 'home') + assert not ctx.value.kind + assert 'not installed' in ctx.value.message + + +def test_connect_slot_snap_not_installed_raises(): + # connect: slot snap not installed raises APIError with empty kind. + ensure_installed(_SNAP) + with pytest.raises(_errors.APIError) as ctx: + _snapd_interfaces.connect(_SNAP, _PLUG, _ABSENT_SNAP, _SLOT) + assert not ctx.value.kind + assert 'not installed' in ctx.value.message diff --git a/snap/tests/unit/test_snapd_interfaces.py b/snap/tests/unit/test_snapd_interfaces.py new file mode 100644 index 000000000..15c57592a --- /dev/null +++ b/snap/tests/unit/test_snapd_interfaces.py @@ -0,0 +1,94 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +# pyright: reportPrivateUsage=false + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from charmlibs.snap import _snapd_interfaces +from charmlibs.snap._errors import _InterfacesUnchangedError + +if TYPE_CHECKING: + from conftest import MockClient + + +class TestConnect: + def test_connect_plug_only(self, mock_client: MockClient): + _snapd_interfaces.connect('vlc', 'mount-observe') + body = mock_client.post.call_args.kwargs['body'] + assert body['slots'] == [{'snap': '', 'slot': ''}] + + def test_connect_with_slot_snap_and_slot(self, mock_client: MockClient): + _snapd_interfaces.connect('vlc', 'plug', 'core', 'myslot') + body = mock_client.post.call_args.kwargs['body'] + assert body['slots'] == [{'snap': 'core', 'slot': 'myslot'}] + assert body['plugs'] == [{'snap': 'vlc', 'plug': 'plug'}] + + def test_connect_slot_snap_no_slot(self, mock_client: MockClient): + _snapd_interfaces.connect('vlc', 'plug', 'core') + body = mock_client.post.call_args.kwargs['body'] + assert body['slots'] == [{'snap': 'core', 'slot': ''}] + + def test_connect_action(self, mock_client: MockClient): + _snapd_interfaces.connect('vlc', 'mount-observe') + body = mock_client.post.call_args.kwargs['body'] + assert body['action'] == 'connect' + + def test_connect_endpoint(self, mock_client: MockClient): + _snapd_interfaces.connect('vlc', 'mount-observe') + mock_client.post.assert_called_once() + assert mock_client.post.call_args.args[0] == '/v2/interfaces' + + +class TestDisconnect: + def test_disconnect_2_arg(self, mock_client: MockClient): + _snapd_interfaces.disconnect('vlc', 'mount-observe') + body = mock_client.post.call_args.kwargs['body'] + assert body['plugs'] == [{'snap': '', 'plug': ''}] + assert body['slots'] == [{'snap': 'vlc', 'slot': 'mount-observe'}] + + def test_disconnect_4_arg(self, mock_client: MockClient): + _snapd_interfaces.disconnect('vlc', 'plug', 'core', 'slot') + body = mock_client.post.call_args.kwargs['body'] + assert body['plugs'] == [{'snap': 'vlc', 'plug': 'plug'}] + assert body['slots'] == [{'snap': 'core', 'slot': 'slot'}] + + def test_disconnect_3_arg(self, mock_client: MockClient): + _snapd_interfaces.disconnect('vlc', 'plug', 'core') + body = mock_client.post.call_args.kwargs['body'] + assert body['slots'] == [{'snap': 'core', 'slot': ''}] + + def test_disconnect_forget(self, mock_client: MockClient): + _snapd_interfaces.disconnect('vlc', 'mount-observe', forget=True) + body = mock_client.post.call_args.kwargs['body'] + assert body['forget'] is True + + def test_disconnect_action(self, mock_client: MockClient): + _snapd_interfaces.disconnect('vlc', 'mount-observe') + body = mock_client.post.call_args.kwargs['body'] + assert body['action'] == 'disconnect' + + def test_disconnect_interfaces_unchanged_suppressed(self, mock_client: MockClient): + # The try/except in disconnect() suppresses _InterfacesUnchangedError + # to make disconnect symmetric with connect (both are no-ops when nothing changes). + mock_client.post.side_effect = _InterfacesUnchangedError( + 'nothing to do', + kind='interfaces-unchanged', + value='', + status_code=400, + status='Bad Request', + ) + _snapd_interfaces.disconnect('vlc', 'mount-observe') # Should not raise. + + def test_disconnect_interfaces_unchanged_suppressed_with_forget(self, mock_client: MockClient): + # _InterfacesUnchangedError is suppressed even when forget=True. + mock_client.post.side_effect = _InterfacesUnchangedError( + 'nothing to do', + kind='interfaces-unchanged', + value='', + status_code=400, + status='Bad Request', + ) + _snapd_interfaces.disconnect('vlc', 'mount-observe', forget=True) # Should not raise.