feat: add change_password for antminer stock firmware

pyasic/config/mining/__init__.py

@@ -268,19 +268,16 @@ def as_bosminer(self) -> dict:
         return cfg
 
     def as_boser(self) -> dict:
+        power_target = (
+            PowerTargetMode(power_target=Power(watt=self.power))
+            if self.power is not None
+            else PowerTargetMode()
+        )
         cfg: dict[str, Any] = {
             "set_performance_mode": SetPerformanceModeRequest(
                 save_action=SaveAction(SaveAction.SAVE_AND_APPLY),
                 mode=PerformanceMode(
-                    tuner_mode=TunerPerformanceMode(
-                        power_target=PowerTargetMode(
-                            power_target=(
-                                Power(watt=self.power)
-                                if self.power is not None
-                                else None
-                            )  # type: ignore[arg-type]
-                        )
-                    )
+                    tuner_mode=TunerPerformanceMode(power_target=power_target)
                 ),
             ),
         }
@@ -367,21 +364,18 @@ def as_bosminer(self) -> dict:
         return {"autotuning": conf}
 
     def as_boser(self) -> dict:
+        hashrate_target = (
+            HashrateTargetMode(
+                hashrate_target=TeraHashrate(terahash_per_second=float(self.hashrate))
+            )
+            if self.hashrate is not None
+            else HashrateTargetMode()
+        )
         cfg: dict[str, Any] = {
             "set_performance_mode": SetPerformanceModeRequest(
                 save_action=SaveAction(SaveAction.SAVE_AND_APPLY),
                 mode=PerformanceMode(
-                    tuner_mode=TunerPerformanceMode(
-                        hashrate_target=HashrateTargetMode(
-                            hashrate_target=TeraHashrate(
-                                terahash_per_second=(
-                                    float(self.hashrate)
-                                    if self.hashrate is not None
-                                    else None
-                                )  # type: ignore[arg-type]
-                            )
-                        )
-                    )
+                    tuner_mode=TunerPerformanceMode(hashrate_target=hashrate_target)
                 ),
             )
         }

pyasic/miners/backends/antminer.py

@@ -214,6 +214,10 @@ async def reboot(self) -> bool:
             return True
         return False
 
+    async def change_password(self, new_password: str) -> bool:
+        data = await self.web.change_password(new_password)
+        return data.get("stats") == "success"
+
     async def stop_mining(self) -> bool:
         cfg = await self.get_config()
         cfg.mining_mode = MiningModeConfig.sleep()

pyasic/miners/backends/btminer.py

@@ -922,7 +922,12 @@ async def get_config(self) -> MinerConfig:
         except LookupError:
             pass
 
-        if pools is not None and settings is not None and device_info is not None:
+        if (
+            pools is not None
+            and settings is not None
+            and device_info is not None
+            and miner_summary is not None
+        ):
             self.config = MinerConfig.from_btminer_v3(
                 rpc_pools=pools,
                 rpc_settings=settings,

pyasic/miners/base.py

@@ -185,6 +185,17 @@ async def set_power_limit(self, wattage: int) -> bool:
         """
         return False
 
+    async def change_password(self, new_password: str) -> bool:
+        """Change the miner's web password.
+
+        Parameters:
+            new_password: The new password to set.
+
+        Returns:
+            A boolean value of the success of changing the password.
+        """
+        return False
+
     async def upgrade_firmware(
         self,
         *,

pyasic/web/antminer.py

@@ -209,6 +209,25 @@ async def get_blink_status(self) -> dict:
         """
         return await self.send_command("get_blink_status")
 
+    async def change_password(self, new_password: str) -> dict:
+        """Change the web password on the miner.
+
+        Args:
+            new_password (str): The new password to set.
+
+        Returns:
+            dict: A dictionary response from the device after the command execution.
+        """
+        result = await self.send_command(
+            "passwd",
+            curPwd=self.pwd,
+            newPwd=new_password,
+            confirmPwd=new_password,
+        )
+        if result.get("stats") == "success":
+            self.pwd = new_password
+        return result
+
     async def set_network_conf(
         self,
         ip: str,

tests/local_tests/test_antminer_local.py

@@ -0,0 +1,101 @@
+import argparse
+import os
+import sys
+import unittest
+
+from pyasic.miners.base import BaseMiner
+from pyasic.miners.data import DataOptions
+from pyasic.miners.factory import MinerFactory
+
+
+class TestAntminerLocal(unittest.IsolatedAsyncioTestCase):
+    ip: str | None = None
+    password: str | None = None
+    miner: BaseMiner
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.ip = os.getenv("ANTMINER_IP")
+        cls.password = os.getenv("ANTMINER_PASSWORD", "root")
+        if not cls.ip:
+            raise unittest.SkipTest("Set ANTMINER_IP to run local Antminer tests")
+
+    async def asyncSetUp(self) -> None:
+        if self.ip is None:
+            self.skipTest("ANTMINER_IP not set")
+        factory = MinerFactory()
+        miner = await factory.get_miner(self.ip)  # type: ignore[func-returns-value]
+        if miner is None:
+            self.skipTest("Miner discovery failed; check IP/auth")
+        self.miner = miner
+        if self.password and hasattr(self.miner, "web") and self.miner.web is not None:
+            self.miner.web.pwd = self.password
+        return None
+
+    async def test_get_data_basics(self):
+        data = await self.miner.get_data(
+            include=[
+                DataOptions.HOSTNAME,
+                DataOptions.API_VERSION,
+                DataOptions.FW_VERSION,
+                DataOptions.HASHRATE,
+            ]
+        )
+        if data.hostname is None:
+            self.skipTest("Hostname not reported; skipping")
+        if data.hashrate is None:
+            self.skipTest("Hashrate not reported; skipping")
+
+        self.assertIsNotNone(data.hostname)
+        self.assertIsNotNone(data.hashrate)
+
+    async def test_get_config(self):
+        cfg = await self.miner.get_config()
+        self.assertIsNotNone(cfg)
+
+    async def test_change_password_roundtrip(self):
+        if not hasattr(self.miner, "change_password"):
+            self.skipTest("Miner does not support change_password")
+
+        original_password = self.password or "root"
+        temp_password = "test_pyasic_pwd"  # nosec B105 - test fixture
+
+        # Change to temp password
+        success = await self.miner.change_password(temp_password)
+        self.assertTrue(success, "Failed to change password to temp value")
+
+        # Verify API still works with new password
+        try:
+            info = await self.miner.web.get_system_info()
+            self.assertIsNotNone(info)
+        except Exception:
+            # Revert attempt even if verification failed
+            self.miner.web.pwd = temp_password
+            await self.miner.change_password(original_password)
+            self.fail("API call failed after password change")
+
+        # Change back to original
+        success = await self.miner.change_password(original_password)
+        self.assertTrue(success, "Failed to revert password to original")
+
+        # Verify API works with original password
+        info = await self.miner.web.get_system_info()
+        self.assertIsNotNone(info)
+
+
+def _main() -> None:
+    parser = argparse.ArgumentParser(description="Local Antminer smoke tests")
+    parser.add_argument("ip", nargs="?", help="Miner IP (overrides ANTMINER_IP)")
+    parser.add_argument("--password", help="Web password (default: root)", default=None)
+    args, unittest_args = parser.parse_known_args()
+
+    if args.ip:
+        os.environ["ANTMINER_IP"] = args.ip
+    if args.password:
+        os.environ["ANTMINER_PASSWORD"] = args.password
+
+    unittest.main(argv=[sys.argv[0]] + unittest_args)
+
+
+if __name__ == "__main__":
+    _main()

tests/test_antminer_change_password.py

@@ -0,0 +1,111 @@
+"""Tests for Antminer stock firmware password change."""
+
+import unittest
+from unittest.mock import AsyncMock, patch
+
+from pyasic.web.antminer import AntminerModernWebAPI
+
+
+class TestAntminerModernChangePassword(unittest.IsolatedAsyncioTestCase):
+    """Test AntminerModernWebAPI.change_password and the AntminerModern backend."""
+
+    async def test_web_api_success_updates_stored_password(self):
+        """Successful passwd.cgi response updates self.pwd."""
+        # Arrange
+        api = AntminerModernWebAPI("192.168.1.1")
+        api.pwd = "old_password"  # nosec B105 - test fixture
+
+        with patch.object(api, "send_command", new_callable=AsyncMock) as mock_send:
+            mock_send.return_value = {
+                "stats": "success",
+                "code": "M000",
+                "msg": "",
+            }
+
+            # Act
+            result = await api.change_password("new_password")
+
+            # Assert
+            mock_send.assert_awaited_once_with(
+                "passwd",
+                curPwd="old_password",
+                newPwd="new_password",
+                confirmPwd="new_password",
+            )
+            self.assertEqual(result["stats"], "success")
+            self.assertEqual(api.pwd, "new_password")
+
+    async def test_web_api_failure_preserves_stored_password(self):
+        """Failed passwd.cgi response leaves self.pwd unchanged."""
+        # Arrange
+        api = AntminerModernWebAPI("192.168.1.1")
+        api.pwd = "old_password"  # nosec B105 - test fixture
+
+        with patch.object(api, "send_command", new_callable=AsyncMock) as mock_send:
+            mock_send.return_value = {
+                "stats": "error",
+                "code": "E000",
+                "msg": "wrong password",
+            }
+
+            # Act
+            result = await api.change_password("new_password")
+
+            # Assert
+            self.assertEqual(result["stats"], "error")
+            self.assertEqual(api.pwd, "old_password")
+
+    async def test_backend_returns_true_on_success(self):
+        """AntminerModern.change_password returns True when the API succeeds."""
+        from pyasic.miners.backends.antminer import AntminerModern
+
+        # Arrange
+        miner = AntminerModern("192.168.1.1")
+
+        with patch.object(
+            miner.web, "change_password", new_callable=AsyncMock
+        ) as mock_change:
+            mock_change.return_value = {"stats": "success", "code": "M000", "msg": ""}
+
+            # Act
+            result = await miner.change_password("new_password")
+
+            # Assert
+            self.assertTrue(result)
+            mock_change.assert_awaited_once_with("new_password")
+
+    async def test_backend_returns_false_on_failure(self):
+        """AntminerModern.change_password returns False when the API fails."""
+        from pyasic.miners.backends.antminer import AntminerModern
+
+        # Arrange
+        miner = AntminerModern("192.168.1.1")
+
+        with patch.object(
+            miner.web, "change_password", new_callable=AsyncMock
+        ) as mock_change:
+            mock_change.return_value = {
+                "stats": "error",
+                "code": "E000",
+                "msg": "wrong password",
+            }
+
+            # Act
+            result = await miner.change_password("new_password")
+
+            # Assert
+            self.assertFalse(result)
+
+    async def test_base_miner_default_returns_false(self):
+        """BaseMiner.change_password returns False by default."""
+        from pyasic.miners.base import BaseMiner
+
+        # Act
+        result = await BaseMiner.change_password(BaseMiner, "anything")
+
+        # Assert
+        self.assertFalse(result)
+
+
+if __name__ == "__main__":
+    unittest.main()