Test sonic-xcvr with xcvr-emu

.github/workflows/test_sonic_xcvr_with_emu.yml

@@ -0,0 +1,20 @@
+name: Test sonic-xcvr with xcvr-emu
+
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+    branches:
+      - master
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: test sonic-xcvr with xcvr-emu
+      run: make test_sonic_xcvr_with_emu

Makefile

@@ -0,0 +1,13 @@
+DOCKER_CMD ?= docker
+SONIC_XCVR_IMAGE ?= sonic-xcvr
+
+.PHONY: test_sonic_xcvr_with_emu
+test_sonic_xcvr_with_emu:
+	$(DOCKER_CMD) build -f docker/Dockerfile -t $(SONIC_XCVR_IMAGE) .
+	$(DOCKER_CMD) run -it \
+	-v `pwd`/docker/pyproject.toml:/sonic_platform_base/pyproject.toml \
+	-v `pwd`/sonic_platform_base:/sonic_platform_base/sonic_platform_base \
+	-v `pwd`/docker/tests:/sonic_platform_base/tests \
+	-w /sonic_platform_base \
+	$(SONIC_XCVR_IMAGE) \
+	python -m pytest -v .

azure-pipelines.yml

@@ -77,7 +77,7 @@ steps:
     set -ex
     pip3 install ".[testing]"
     pip3 uninstall --yes sonic-platform-common
-    pytest
+    pytest --ignore docker
   displayName: 'Test Python 3'
 
 - task: PublishTestResults@2

docker/Dockerfile

@@ -0,0 +1,12 @@
+FROM ghcr.io/ishidawataru/xcvr-emu:main
+
+WORKDIR /sonic_platform_base
+
+RUN --mount=type=bind,source=docker/sonic_py_common,target=/sonic_py_common,rw \
+    cd /sonic_py_common && pip install .
+
+RUN mkdir sonic_platform_base && touch sonic_platform_base/__init__.py
+
+COPY docker/pyproject.toml .
+
+RUN pip install '.[dev]'

docker/pyproject.toml

@@ -0,0 +1,10 @@
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "sonic-platform-common"
+version = "1.0.0"
+
+[project.optional-dependencies]
+dev = ["pytest", "mock", "PyYAML", "pytest-asyncio"]

docker/sonic_py_common/pyproject.toml

@@ -0,0 +1,7 @@
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "sonic-py-common"
+version = "1.0.0"

docker/sonic_py_common/sonic_py_common/logger.py

@@ -0,0 +1,3 @@
+from unittest.mock import MagicMock
+
+Logger = MagicMock()

docker/tests/sonic_xcvr_with_emu/test_cmis.py

@@ -0,0 +1,83 @@
+import pytest
+
+from sonic_platform_base.sonic_xcvr.xcvr_eeprom import XcvrEeprom
+from sonic_platform_base.sonic_xcvr.api.public.cmis import CmisApi
+from sonic_platform_base.sonic_xcvr.mem_maps.public.cmis import CmisMemMap
+from sonic_platform_base.sonic_xcvr.codes.public.cmis import CmisCodes
+
+from xcvr_emu.transceiver import CMISTransceiver  # type: ignore
+from xcvr_emu.proto.emulator_pb2 import (  # type: ignore
+    ReadRequest,
+    WriteRequest,
+)
+
+
+class XcvrEmuEeprom(XcvrEeprom):
+    def __init__(self, config: dict):
+
+        codes = CmisCodes
+        mem_map = CmisMemMap(codes)
+
+        self.xcvr = CMISTransceiver(
+            0,
+            {
+                "present": True,
+                "defaults": config,
+            },
+        )
+
+        super().__init__(self._read, self._write, mem_map)
+
+    def _read(self, offset, num_bytes):
+        if not self.xcvr.present:
+            return None
+        # convert optoe offset to SFF page and offset
+        # optoe maps the SFF 2D address to a linear address
+        page = offset // 128
+        if page > 0:
+            page = page - 1
+
+        if offset > 128:
+            offset = (offset % 128) + 128
+
+        return self.xcvr.read(
+            ReadRequest(index=0, offset=offset, page=page, length=num_bytes)
+        )
+
+    def _write(self, offset, num_bytes, write_buffer):
+        assert len(write_buffer) <= num_bytes
+        # convert optoe offset to SFF page and offset
+        # optoe maps the SFF 2D address to a linear address
+        page = offset // 128
+        if page > 0:
+            page = page - 1
+
+        if offset > 128:
+            offset = (offset % 128) + 128
+
+        return self.xcvr.write(
+            WriteRequest(
+                index=0,
+                page=page,
+                offset=offset,
+                length=num_bytes,
+                data=bytes(write_buffer),
+            )
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "emu_response, expected", [("1234567890", "1234567890"), ("ABCD", "ABCD")]
+)
+async def test_get_model(emu_response, expected):
+    eeprom = XcvrEmuEeprom(
+        {
+            "VendorPN": emu_response,
+        },
+    )
+    api = CmisApi(eeprom)
+    result = api.get_model()
+    result = result.rstrip("\x00")
+    await eeprom.xcvr.plugout()
+    assert result == expected