v2.0.0.dev1

.github/workflows/pytest-integration.yml

@@ -22,5 +22,31 @@ jobs:
       - name: Install packages
         run: uv sync --all-extras --dev
 
+      - name: Start ejabberd
+        run: docker compose -f tests/xmpp/docker-compose.yml up -d
+
+      - name: Wait for ejabberd to be healthy
+        run: |
+          for i in $(seq 1 30); do
+            if docker compose -f tests/xmpp/docker-compose.yml ps | grep -q "healthy"; then
+              echo "ejabberd is healthy"
+              exit 0
+            fi
+            echo "Waiting for ejabberd... ($i/30)"
+            sleep 2
+          done
+          echo "ejabberd failed to become healthy"
+          docker compose -f tests/xmpp/docker-compose.yml logs
+          exit 1
+
       - name: Run integration tests
-        run: uv run pytest -m integration
+        env:
+          PYOBS_TEST_XMPP_HOST: localhost
+          PYOBS_TEST_XMPP_DOMAIN: localhost
+          PYOBS_TEST_XMPP_PORT: "5222"
+          PYOBS_TEST_XMPP_PASSWORD: pyobs
+        run: uv run pytest -m integration
+
+      - name: Stop ejabberd
+        if: always()
+        run: docker compose -f tests/xmpp/docker-compose.yml down

.github/workflows/lint.yml → .github/workflows/ruff.yml

@@ -1,8 +1,8 @@
-name: Lint
+name: ruff
 on: push
 
 jobs:
-  lint:
+  ruff:
     runs-on: ubuntu-latest
     timeout-minutes: 10
 
@@ -21,6 +21,3 @@ jobs:
 
       - name: Run ruff
         run: uv run ruff check pyobs/
-
-      - name: Run mypy
-        run: uv run mypy pyobs/

conftest.py

@@ -1,5 +1,7 @@
 import inspect
+import os
 from typing import Any
+
 import pytest
 
 
@@ -9,6 +11,11 @@ def pytest_addoption(parser: Any) -> None:
 
 def pytest_configure(config: Any) -> None:
     config.addinivalue_line("markers", "ssh: mark test as using SSH")
+    config.addinivalue_line("markers", "integration: mark test as an integration test")
+    config.addinivalue_line(
+        "markers",
+        "xmpp: mark test as requiring a live ejabberd server " "(set PYOBS_TEST_XMPP_HOST to enable)",
+    )
 
 
 def pytest_collection_modifyitems(config: Any, items: Any) -> None:
@@ -24,6 +31,13 @@ def pytest_collection_modifyitems(config: Any, items: Any) -> None:
             if "ssh" in item.keywords:
                 item.add_marker(nossh)
 
+    # do XMPP tests?
+    if not os.environ.get("PYOBS_TEST_XMPP_HOST"):
+        noxmpp = pytest.mark.skip(reason="XMPP testing disabled (set PYOBS_TEST_XMPP_HOST to enable)")
+        for item in items:
+            if "xmpp" in item.keywords:
+                item.add_marker(noxmpp)
+
 
 @pytest.fixture(scope="session", autouse=True)
 def download_IERS() -> None:

pyobs/application.py

@@ -98,7 +98,7 @@ def __init__(self, **kw: Any) -> None:
                     super().__init__(identifier="pyobs", **kw)
 
                 def _format_record(self, record: logging.LogRecord) -> list[tuple[str, Any]]:
-                    pairs = super()._format_record(record)
+                    pairs: list[tuple[str, Any]] = super()._format_record(record)
                     pairs.append(("PYOBS_MODULE", getattr(record, "pyobs_module", "")))
                     return pairs
 

pyobs/comm/comm.py

@@ -1,39 +1,39 @@
 from __future__ import annotations
 
 import asyncio
+import functools
 import inspect
 import logging
 from collections.abc import Callable, Coroutine
-from typing import TYPE_CHECKING, Any, TypeVar, overload
+from typing import TYPE_CHECKING, Any, overload
 
 import pyobs.interfaces
 from pyobs.events import Event, LogEvent, ModuleClosedEvent
 from pyobs.interfaces import Interface
 
 from .commlogging import CommLoggingHandler
-from .proxy import Proxy
+from .proxy import Proxy, ProxyType, _ProxyContext
 
 if TYPE_CHECKING:
     from pyobs.modules import Module
 
-log = logging.getLogger(__name__)
-
+StateCallback = Callable[[Any], None]
 
-ProxyType = TypeVar("ProxyType")
+log = logging.getLogger(__name__)
 
 
 class Comm:
     """Base class for all Comm modules in pyobs."""
 
     __module__ = "pyobs.comm"
 
-    def __init__(self, cache_proxies: bool = True):
+    def __init__(self) -> None:
         """Creates a comm module."""
 
         self._proxies: dict[str, Proxy] = {}
+        self._state_subscriptions: dict[str, list[tuple[type[Interface], StateCallback]]] = {}
         self._module: Module | None = None
         self._log_queue: asyncio.Queue[LogEvent] = asyncio.Queue()
-        self._cache_proxies = cache_proxies
         self._logging_task: asyncio.Task[Any] | None = None
         self._event_handlers: dict[type[Event], list[Callable[[Event, str], Coroutine[Any, Any, bool]]]] = {}
         self._closing = asyncio.Event()
@@ -117,7 +117,7 @@ async def _get_client(self, client: str) -> Module | Proxy | None:
             return None
 
         # if client doesn't exist or we disabled caching, fetch a new proxy
-        if client not in self._proxies or not self._cache_proxies:
+        if client not in self._proxies:
             # get interfaces
             try:
                 interfaces = await self.get_interfaces(client)
@@ -126,18 +126,20 @@ async def _get_client(self, client: str) -> Module | Proxy | None:
 
             # create new proxy
             proxy = Proxy(self, client, interfaces)
+
+            # subscribe to state
+            for interface in interfaces:
+                if getattr(interface, "state", None) is not None:
+                    await self.subscribe_state(client, interface, functools.partial(proxy.update_state, interface))
+
             self._proxies[client] = proxy
 
         # return proxy
         return self._proxies[client]
 
-    @overload
-    async def proxy(self, name_or_object: str | object, obj_type: type[ProxyType]) -> ProxyType: ...
-
-    @overload
-    async def proxy(self, name_or_object: str | object, obj_type: type[ProxyType] | None = None) -> Any: ...
-
-    async def proxy(self, name_or_object: str | object, obj_type: type[ProxyType] | None = None) -> Any | ProxyType:
+    async def _resolve_proxy(
+        self, name_or_object: str | object, obj_type: type[ProxyType] | None = None
+    ) -> Any | ProxyType:
         """Returns object directly if it is of given type. Otherwise get proxy of client with given name and check type.
 
         If name_or_object is an object:
@@ -184,16 +186,41 @@ async def proxy(self, name_or_object: str | object, obj_type: type[ProxyType] |
             # completely wrong...
             raise ValueError(f'Given parameter is neither a name nor an object of requested type "{obj_type}".')
 
-    async def safe_proxy(
+    async def _safe_resolve_proxy(
         self, name_or_object: str | object, obj_type: type[ProxyType] | None = None
     ) -> Any | ProxyType | None:
         """Calls proxy() in a safe way and returns None instead of raising an exception."""
 
         try:
-            return await self.proxy(name_or_object, obj_type)
+            return await self._resolve_proxy(name_or_object, obj_type)
         except ValueError:
             return None
 
+    @overload
+    def proxy(self, name_or_object: str | object, obj_type: type[ProxyType]) -> _ProxyContext[ProxyType]: ...
+    @overload
+    def proxy(self, name_or_object: str | object, obj_type: None = None) -> _ProxyContext[Any]: ...
+
+    def proxy(self, name_or_object: str | object, obj_type: type[ProxyType] | None = None) -> _ProxyContext[Any]:
+        """Returns a context manager; use as `async with self.proxy(...) as x:`."""
+        return _ProxyContext(self._resolve_proxy(name_or_object, obj_type))
+
+    @overload
+    def safe_proxy(
+        self, name_or_object: str | object, obj_type: type[ProxyType]
+    ) -> _ProxyContext[ProxyType | None]: ...
+    @overload
+    def safe_proxy(self, name_or_object: str | object, obj_type: None = None) -> _ProxyContext[Any]: ...
+
+    def safe_proxy(self, name_or_object: str | object, obj_type: type[ProxyType] | None = None) -> _ProxyContext[Any]:
+        """Same as proxy(), but yields None inside the block instead of raising."""
+        return _ProxyContext(self._safe_resolve_proxy(name_or_object, obj_type))
+
+    async def has_proxy(self, name_or_object: str | object, obj_type: type[Any] | None = None) -> bool:
+        """True if a proxy of the given type can currently be resolved. Doesn't keep a reference
+        to it, so doesn't need async with the way proxy()/safe_proxy() do."""
+        return await self._safe_resolve_proxy(name_or_object, obj_type) is not None
+
     async def _client_disconnected(self, event: Event, sender: str) -> bool:
         """Called when a client disconnects.
 
@@ -203,9 +230,15 @@ async def _client_disconnected(self, event: Event, sender: str) -> bool:
 
         """
 
-        # if a client disconnects, we remove its proxy
+        # if a client disconnects, clear its proxy state then evict it
         if sender in self._proxies:
+            self._proxies[sender].clear_state()
             del self._proxies[sender]
+
+        # tear down any state subscriptions held for that client
+        for interface, callback in self._state_subscriptions.pop(sender, []):
+            await self.unsubscribe_state(sender, interface, callback)
+
         return True
 
     @property
@@ -397,6 +430,67 @@ async def _register_events(
     ) -> None:
         pass
 
+    async def set_state(self, interface: type[Interface], state: Any) -> None:
+        """Publish state for this module.
+
+        Args:
+            interface: Interface type for the state.
+            state: State object to publish.
+        """
+        await self._set_state(interface, state)
+
+    async def _set_state(self, interface: type[Interface], state: Any) -> None:
+        pass
+
+    async def subscribe_state(
+        self,
+        module: str,
+        interface: type[Interface],
+        callback: StateCallback,
+    ) -> None:
+        """Subscribe to state updates for a given module and interface.
+
+        Delivers the current value immediately on subscribe.
+
+        Args:
+            module: Name of remote module.
+            interface: Interface type to subscribe to.
+            callback: Called with state object on each update.
+        """
+        self._state_subscriptions.setdefault(module, []).append((interface, callback))
+        await self._subscribe_state(module, interface, callback)
+
+    async def _subscribe_state(
+        self,
+        module: str,
+        interface: type[Interface],
+        callback: StateCallback,
+    ) -> None:
+        pass
+
+    async def unsubscribe_state(
+        self,
+        module: str,
+        interface: type[Interface],
+        callback: StateCallback,
+    ) -> None:
+        """Unsubscribe from state updates.
+
+        Args:
+            module: Name of remote module.
+            interface: Interface type to unsubscribe from.
+            callback: Callback that was registered.
+        """
+        await self._unsubscribe_state(module, interface, callback)
+
+    async def _unsubscribe_state(
+        self,
+        module: str,
+        interface: type[Interface],
+        callback: StateCallback,
+    ) -> None:
+        pass
+
     def _send_event_to_module(self, event: Event, from_client: str) -> None:
         """Send an event to all connected modules.