Implementing set_optoe_write_timeout API (#422)

Signed-off-by: Mihir Patel <[email protected]>

sonic_platform_base/sonic_xcvr/sfp_optoe_base.py

@@ -230,6 +230,15 @@ def set_optoe_write_max(self, write_max):
         except (OSError, IOError):
             pass
 
+    def set_optoe_write_timeout(self, write_timeout):
+        sys_path = self.get_eeprom_path()
+        sys_path = sys_path.replace("eeprom", "write_timeout")
+        try:
+            with open(sys_path, mode='w') as f:
+                f.write(str(write_timeout))
+        except (OSError, IOError):
+            pass
+
     def read_eeprom(self, offset, num_bytes):
         try:
             with open(self.get_eeprom_path(), mode='rb', buffering=0) as f:

tests/sonic_xcvr/test_sfp_optoe_base.py

@@ -1,3 +1,4 @@
+from unittest.mock import mock_open
 from mock import MagicMock 
 from mock import patch 
 import pytest 
@@ -79,3 +80,36 @@ def test_get_vdm_unfreeze_status(self, mock_response1, mock_response2, expected)
         result = self.sfp_optoe_api.get_vdm_unfreeze_status()
         assert result == expected
 
+    @patch("builtins.open", new_callable=mock_open)
+    @patch.object(SfpOptoeBase, 'get_eeprom_path')
+    def test_set_optoe_write_timeout_success(self, mock_get_eeprom_path, mock_open):
+        mock_get_eeprom_path.return_value = "/sys/bus/i2c/devices/1-0050/eeprom"
+        expected_path = "/sys/bus/i2c/devices/1-0050/write_timeout"
+        expected_timeout = 1
+
+        self.sfp_optoe_api.set_optoe_write_timeout(expected_timeout)
+
+        mock_open.assert_called_once_with(expected_path, mode='w')
+        mock_open().write.assert_called_once_with(str(expected_timeout))
+
+    @patch("builtins.open", new_callable=mock_open)
+    @patch.object(SfpOptoeBase, 'get_eeprom_path')
+    def test_set_optoe_write_timeout_ioerror(self, mock_get_eeprom_path, mock_open):
+        mock_get_eeprom_path.return_value = "/sys/bus/i2c/devices/1-0050/eeprom"
+        expected_timeout = 1
+        mock_open.side_effect = IOError
+
+        self.sfp_optoe_api.set_optoe_write_timeout(expected_timeout)
+
+        mock_open.assert_called()
+
+    @patch("builtins.open", new_callable=mock_open)
+    @patch.object(SfpOptoeBase, 'get_eeprom_path')
+    def test_set_optoe_write_timeout_oserror(self, mock_get_eeprom_path, mock_open):
+        mock_get_eeprom_path.return_value = "/sys/bus/i2c/devices/1-0050/eeprom"
+        expected_timeout = 1
+        mock_open.side_effect = OSError
+
+        self.sfp_optoe_api.set_optoe_write_timeout(expected_timeout)
+
+        mock_open.assert_called()