feat(snap): add snap interfaces module

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',

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 <snap>:<slot>`.
+        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 <snap>:<plug> <snap>:<slot>`.
+        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.

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