Merge pull request #160 from ChannelFinder/refactor-cf-subpackage

Author: anderslindho

.github/workflows/server.yml

@@ -68,14 +68,21 @@ jobs:
     - name: Test unit tests
       run: |
         set -o pipefail
-        pytest tests/unit -v 2>&1 | tee pytest-unit.log
+        pytest tests/unit -v --cov=recceiver --cov-report=xml:coverage.xml 2>&1 | tee pytest-unit.log
     - name: Upload test log
       if: always()
       uses: actions/upload-artifact@v4
       with:
         name: pytest-unit-log
         path: server/pytest-unit.log
         retention-days: 14
+    - name: Upload coverage report
+      if: always()
+      uses: actions/upload-artifact@v4
+      with:
+        name: coverage-xml
+        path: server/coverage.xml
+        retention-days: 14
 
   test-integration:
     runs-on: ubuntu-latest

server/pyproject.toml

@@ -4,7 +4,7 @@ requires = [ "setuptools" ]
 
 [project]
 name = "recceiver"
-version = "1.9.3"
+version = "1.9.5"
 description = """\
   recCeiver is a server component of the recsync protocol. It receives record updates from recsync clients (e.g., \
   recCasters) and forwards them to a configurable backend such as ChannelFinder.\
@@ -36,11 +36,11 @@ dependencies = [
   "twisted>=22.10,<23; python_version<'3.8'",
   "twisted>=24.11,<24.12; python_version>='3.8'",
 ]
-optional-dependencies.test = [ "pytest>=8.3,<8.4", "testcontainers>=4.8.2,<4.9" ]
+optional-dependencies.test = [ "pytest>=8.3,<8.4", "pytest-cov>=6,<7", "testcontainers>=4.8.2,<4.9" ]
 urls.Repository = "https://github.com/ChannelFinder/recsync"
 
 [tool.setuptools]
-packages = [ "recceiver", "recceiver.protocol", "twisted.plugins" ]
+packages = [ "recceiver", "recceiver.cf", "recceiver.protocol", "twisted.plugins" ]
 include-package-data = true
 package-data.twisted = [ "plugins/recceiver_plugin.py" ]
 

server/recceiver/cf/__init__.py

@@ -0,0 +1,102 @@
+from typing import List
+
+try:
+    from typing import Protocol
+except ImportError:
+    from typing_extensions import Protocol  # type: ignore[assignment]
+
+from recceiver.cf.model import CFChannel, CFProperty, CFPropertyName, PVStatus
+
+# CF query URLs break above this length; names are pipe-joined and chunked to stay under it.
+_CF_NAME_QUERY_LIMIT = 600
+
+
+class ChannelFinderAdapter(Protocol):
+    """Typed boundary between CFProcessor and the ChannelFinder HTTP client.
+
+    All methods accept and return domain objects (CFChannel, CFProperty).
+    Dict serialisation is handled inside the implementation, not at callsites.
+    """
+
+    def find_by_ioc_id(self, iocid: str) -> List[CFChannel]:
+        """Return all channels registered under the given IOC ID."""
+        ...
+
+    def find_by_names(self, names: List[str]) -> List[CFChannel]:
+        """Return channels whose names are in the given list."""
+        ...
+
+    def find_active_for_recceiver(self, recceiverid: str) -> List[CFChannel]:
+        """Return all channels marked Active for the given recceiver."""
+        ...
+
+    def set_channels(self, channels: List[CFChannel]) -> None:
+        """Create or overwrite channels."""
+        ...
+
+    def update_property(self, prop: CFProperty, channel_names: List[str]) -> None:
+        """Update a single property value across the named channels."""
+        ...
+
+    def get_property_names(self) -> List[str]:
+        """Return the names of all property definitions registered in ChannelFinder."""
+        ...
+
+    def set_property(self, name: str, owner: str) -> None:
+        """Register a property definition if it does not already exist."""
+        ...
+
+
+class PyCFClientAdapter:
+    """Wraps pyCFClient's ChannelFinderClient to implement ChannelFinderAdapter."""
+
+    def __init__(self, client, size_limit: int = 0):
+        self._client = client
+        self._size_limit = size_limit
+
+    def _find(self, args: List) -> List[CFChannel]:
+        if self._size_limit > 0:
+            args = args + [("~size", self._size_limit)]
+        return [CFChannel.from_dict(ch) for ch in self._client.findByArgs(args)]
+
+    def find_by_ioc_id(self, iocid: str) -> List[CFChannel]:
+        return self._find([(CFPropertyName.IOC_ID.value, iocid)])
+
+    def find_by_names(self, names: List[str]) -> List[CFChannel]:
+        if not names:
+            return []
+        chunks, buf = [], ""
+        for name in names:
+            if not buf:
+                buf = name
+            elif len(buf) + len(name) < _CF_NAME_QUERY_LIMIT:
+                buf = buf + "|" + name
+            else:
+                chunks.append(buf)
+                buf = name
+        if buf:
+            chunks.append(buf)
+        results = []
+        for chunk in chunks:
+            results.extend(self._find([("~name", chunk)]))
+        return results
+
+    def find_active_for_recceiver(self, recceiverid: str) -> List[CFChannel]:
+        return self._find(
+            [
+                (CFPropertyName.PV_STATUS.value, PVStatus.ACTIVE.value),
+                (CFPropertyName.RECCEIVER_ID.value, recceiverid),
+            ]
+        )
+
+    def set_channels(self, channels: List[CFChannel]) -> None:
+        self._client.set(channels=[ch.as_dict() for ch in channels])
+
+    def update_property(self, prop: CFProperty, channel_names: List[str]) -> None:
+        self._client.update(property=prop.as_dict(), channelNames=channel_names)
+
+    def get_property_names(self) -> List[str]:
+        return [p["name"] for p in self._client.getAllProperties()]
+
+    def set_property(self, name: str, owner: str) -> None:
+        self._client.set(property={"name": name, "owner": owner})

server/recceiver/cf/adapter.py

@@ -0,0 +1,66 @@
+import socket
+from dataclasses import dataclass, fields
+from typing import Optional
+
+from recceiver.processors import ConfigAdapter
+
+RECCEIVERID_DEFAULT = socket.gethostname()
+DEFAULT_QUERY_LIMIT = 10_000
+
+
+@dataclass
+class CFConfig:
+    """Configuration options for the CF Processor."""
+
+    alias_enabled: bool = False
+    record_type_enabled: bool = False
+    environment_variables: str = ""
+    info_tags: str = ""
+    ioc_connection_info: bool = True
+    record_description_enabled: bool = False
+    clean_on_start: bool = True
+    clean_on_stop: bool = True
+    username: str = "cfstore"
+    env_owner_variable: str = "ENGINEER"
+    recceiver_id: str = RECCEIVERID_DEFAULT
+    timezone: Optional[str] = None
+    cf_query_limit: int = DEFAULT_QUERY_LIMIT
+    base_url: Optional[str] = None
+    cf_username: Optional[str] = None
+    cf_password: Optional[str] = None
+    verify_ssl: Optional[bool] = None
+    push_max_retries: int = 10
+    push_always_retry: bool = True
+
+    @classmethod
+    def loads(cls, conf: ConfigAdapter) -> "CFConfig":
+        """Load configuration from a ConfigAdapter instance."""
+        return CFConfig(
+            alias_enabled=conf.getboolean("alias", False),
+            record_type_enabled=conf.getboolean("recordType", False),
+            environment_variables=conf.get("environment_vars", ""),
+            info_tags=conf.get("infotags", ""),
+            ioc_connection_info=conf.getboolean("iocConnectionInfo", True),
+            record_description_enabled=conf.getboolean("recordDesc", False),
+            clean_on_start=conf.getboolean("cleanOnStart", True),
+            clean_on_stop=conf.getboolean("cleanOnStop", True),
+            username=conf.get("username", "cfstore"),
+            recceiver_id=conf.get("recceiverId", RECCEIVERID_DEFAULT),
+            timezone=conf.get("timezone", ""),
+            cf_query_limit=conf.get("findSizeLimit", DEFAULT_QUERY_LIMIT),
+            base_url=conf.get("baseUrl"),
+            cf_username=conf.get("cfUsername"),
+            cf_password=conf.get("cfPassword"),
+            verify_ssl=conf.getboolean("verifySSL"),
+            push_max_retries=conf.getint("pushMaxRetries", 10),
+            push_always_retry=conf.getboolean("pushAlwaysRetry", True),
+        )
+
+    def __repr__(self) -> str:
+        parts = []
+        for f in fields(self):
+            value = getattr(self, f.name)
+            if f.name == "cf_password":
+                value = "***" if value else None
+            parts.append(f"{f.name}={value!r}")
+        return f"CFConfig({', '.join(parts)})"

server/recceiver/cf/config.py

@@ -0,0 +1,111 @@
+import enum
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+class PVStatus(enum.Enum):
+    """Active/Inactive status values as used in the pvStatus CF property."""
+
+    ACTIVE = "Active"
+    INACTIVE = "Inactive"
+
+
+class CFPropertyName(enum.Enum):
+    """Canonical property names registered and managed in Channelfinder."""
+
+    HOSTNAME = "hostName"
+    IOC_NAME = "iocName"
+    IOC_ID = "iocid"
+    IOC_IP = "iocIP"
+    PV_STATUS = "pvStatus"
+    TIME = "time"
+    RECCEIVER_ID = "recceiverID"
+    ALIAS = "alias"
+    RECORD_TYPE = "recordType"
+    RECORD_DESC = "recordDesc"
+    CA_PORT = "caPort"
+    PVA_PORT = "pvaPort"
+
+
+@dataclass
+class CFProperty:
+    """A single named property attached to a Channelfinder channel."""
+
+    name: str
+    owner: str
+    value: Optional[str] = None
+
+    def as_dict(self) -> Dict[str, str]:
+        """Serialise to the dict shape expected by pyCFClient."""
+        return {"name": self.name, "owner": self.owner, "value": self.value or ""}
+
+    @classmethod
+    def from_dict(cls, prop_dict: Dict[str, str]) -> "CFProperty":
+        """Deserialise from the dict shape returned by pyCFClient."""
+        return cls(
+            name=prop_dict.get("name", ""),
+            owner=prop_dict.get("owner", ""),
+            value=prop_dict.get("value"),
+        )
+
+
+@dataclass
+class CFChannel:
+    """A Channelfinder channel with its associated properties."""
+
+    name: str
+    owner: str
+    properties: List[CFProperty]
+
+    def as_dict(self) -> Dict[str, Any]:
+        """Serialise to the dict shape expected by pyCFClient."""
+        return {
+            "name": self.name,
+            "owner": self.owner,
+            "properties": [p.as_dict() for p in self.properties],
+        }
+
+    @classmethod
+    def from_dict(cls, channel_dict: Dict[str, Any]) -> "CFChannel":
+        """Deserialise from the dict shape returned by pyCFClient."""
+        return cls(
+            name=channel_dict.get("name", ""),
+            owner=channel_dict.get("owner", ""),
+            properties=[CFProperty.from_dict(p) for p in channel_dict.get("properties", [])],
+        )
+
+
+@dataclass
+class IOCInfo:
+    """Runtime state for a connected IOC. The .id property is the primary key."""
+
+    host: str
+    hostname: str
+    ioc_name: str
+    ioc_ip: str
+    owner: str
+    time: str
+    port: int
+    channelcount: int = 0
+
+    @property
+    def id(self) -> str:
+        return f"{self.host}:{self.port}"
+
+
+@dataclass
+class RecordInfo:
+    """Per-record data extracted from a transaction before pushing to CF."""
+
+    pv_name: str
+    record_type: Optional[str] = None
+    info_properties: List[CFProperty] = field(default_factory=list)
+    aliases: List[str] = field(default_factory=list)
+
+
+class IOCMissingInfoError(Exception):
+    """Raised when an IOC is missing required information."""
+
+    def __init__(self, ioc_info: IOCInfo):
+        super().__init__(f"Missing hostName {ioc_info.hostname} or iocName {ioc_info.ioc_name}")
+        self.ioc_info = ioc_info