chore: tests

Author: jmoreira-valory

operate/cli.py

@@ -44,7 +44,7 @@
 from operate.account.user import UserAccount
 from operate.constants import KEY, KEYS, OPERATE_HOME, SERVICES
 from operate.ledger.profiles import DEFAULT_NEW_SAFE_FUNDS_AMOUNT
-from operate.migration import FormatMigrator
+from operate.migration import MigrationManager
 from operate.operate_types import Chain, DeploymentStatus, LedgerType
 from operate.quickstart.analyse_logs import analyse_logs
 from operate.quickstart.claim_staking_rewards import claim_staking_rewards
@@ -96,8 +96,8 @@ def __init__(
         )
         self.password: t.Optional[str] = os.environ.get("OPERATE_USER_PASSWORD")
 
-        fm = FormatMigrator(self._path, self.logger)
-        fm.migrate_user_json()
+        mm = MigrationManager(self._path, self.logger)
+        mm.migrate_account_user()
 
     def create_user_account(self, password: str) -> UserAccount:
         """Create a user account."""

operate/migration.py

@@ -25,7 +25,7 @@
 from pathlib import Path
 
 
-class FormatMigrator:
+class MigrationManager:
     """FormatMigrator"""
 
     # TODO Backport here migration for services/config.json, etc.
@@ -40,11 +40,10 @@ def __init__(
         self._path = home
         self.logger = logger
 
-    def migrate_user_json(self) -> None:
+    def migrate_account_user(self) -> None:
         """Migrates user.json"""
-        self.logger.info("[FORMAT MIGRATOR] Migrating user.json")
-        path = self._path / "user.json"
 
+        path = self._path / "user.json"
         if not path.exists():
             return
 
@@ -57,3 +56,5 @@ def migrate_user_json(self) -> None:
         new_data = {"password_hash": data["password_sha"]}
         with open(path, "w", encoding="utf-8") as f:
             json.dump(new_data, f, indent=4)
+
+        self.logger.info("[MIGRATION MANAGER] Migrated user.json.")

tests/conftest.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+# ------------------------------------------------------------------------------
+#
+#   Copyright 2025 Valory AG
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+#
+# ------------------------------------------------------------------------------
+
+"""
+Fixtures for pytest
+
+The conftest.py file serves as a means of providing fixtures for an entire
+directory. Fixtures defined in a conftest.py can be used by any test in that
+package without needing to import them (pytest will automatically discover them).
+
+See https://docs.pytest.org/en/stable/reference/fixtures.html
+"""
+
+import random
+import string
+
+import pytest
+
+
+def random_string(length: int = 8) -> str:
+    """random_string"""
+    chars = string.ascii_letters + string.digits
+    return "".join(random.choices(chars, k=length))  # nosec B311
+
+
+@pytest.fixture
+def password() -> str:
+    """password fixture"""
+    return random_string(16)

tests/test_operate_cli.py

@@ -19,31 +19,29 @@
 
 """Tests for operate.cli module."""
 
+import hashlib
+import json
 import random
 import string
 from pathlib import Path
 
 import pytest
+from argon2 import PasswordHasher, Type
+from argon2.exceptions import InvalidHashError
 from web3 import Web3
 
 from operate.cli import OperateApp
+from operate.constants import OPERATE_HOME
 from operate.operate_types import LedgerType
 
 
 ROOT_PATH = Path(__file__).resolve().parent
-OPERATE = ".operate_test"
 
 MSG_NEW_PASSWORD_MISSING = "You must provide a new password"  # nosec
 MSG_INVALID_PASSWORD = "Password is not valid"  # nosec
 MSG_INVALID_MNEMONIC = "Seed phrase is not valid"  # nosec
 
 
-def random_string(length: int = 8) -> str:
-    """random_string"""
-    chars = string.ascii_letters + string.digits
-    return "".join(random.choices(chars, k=length))  # nosec B311
-
-
 def random_mnemonic(num_words: int = 12) -> str:
     """Generate a random BIP-39 mnemonic"""
     w3 = Web3()
@@ -52,7 +50,7 @@ def random_mnemonic(num_words: int = 12) -> str:
     return mnemonic
 
 
-class TestOperate:
+class TestOperateApp:
     """Tests for operate.cli.OperateApp class."""
 
     def test_update_password(
@@ -62,16 +60,16 @@ def test_update_password(
         """Test operate.update_password() and operate.update_password_with_mnemonic()"""
 
         operate = OperateApp(
-            home=tmp_path / OPERATE,
+            home=tmp_path / OPERATE_HOME,
         )
         operate.setup()
         password1 = random_string()
         operate.create_user_account(password=password1)
         operate.password = password1
         wallet_manager = operate.wallet_manager
-        _, mnemonic = wallet_manager.create(LedgerType.ETHEREUM)
-        num_words = len(mnemonic)
-        mnemonic = " ".join(mnemonic)
+        _, mnemonic_list = wallet_manager.create(LedgerType.ETHEREUM)
+        num_words = len(mnemonic_list)
+        mnemonic = " ".join(mnemonic_list)
 
         password2 = random_string()
         password3 = "!@#Test$%^"
@@ -139,3 +137,46 @@ def test_update_password(
         invalid_mnemonic = random_mnemonic(num_words=15)
         with pytest.raises(ValueError, match=rf"^{MSG_INVALID_MNEMONIC}"):
             operate.update_password_with_mnemonic(invalid_mnemonic, password2)
+
+    def test_migrate_account(
+        self,
+        tmp_path: Path,
+        password: str,
+    ) -> None:
+        operate_home_path = tmp_path / OPERATE_HOME
+        operate = OperateApp(
+            home=operate_home_path,
+        )
+
+        # Artificially create an old-format user.json
+        sha256 = hashlib.sha256()
+        sha256.update(password.encode())
+        password_sha = sha256.hexdigest()
+        data = {"password_sha": password_sha}
+        user_json_path = operate_home_path / "user.json"
+        user_json_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+
+        operate = OperateApp(
+            home=operate_home_path,
+        )
+
+        data = json.loads(user_json_path.read_text(encoding="utf-8"))
+
+        assert operate.user_account
+        assert "password_hash" in data
+        assert "password_sha" not in data
+        assert password_sha == data["password_hash"]
+        ph = PasswordHasher(type=Type.ID)
+        with pytest.raises(InvalidHashError):
+            ph.verify(data["password_hash"], password)
+
+        operate.user_account.is_valid(password)
+        data = json.loads(user_json_path.read_text(encoding="utf-8"))
+
+        assert operate.user_account
+        assert operate.user_account.is_valid(password)
+        assert "password_hash" in data
+        assert "password_sha" not in data
+        assert password_sha != data["password_hash"]
+        ph = PasswordHasher(type=Type.ID)
+        assert ph.verify(data["password_hash"], password)

tests/test_services_manage.py

@@ -19,8 +19,6 @@
 
 """Tests for services.service module."""
 
-import random
-import string
 import typing as t
 from pathlib import Path
 
@@ -38,14 +36,6 @@
 OPERATE = ".operate_test"
 
 
-@pytest.fixture
-def random_string() -> str:
-    """random_string"""
-    length = 8
-    chars = string.ascii_letters + string.digits
-    return "".join(random.choices(chars, k=length))  # nosec B311
-
-
 def get_template(**kwargs: t.Any) -> ServiceTemplate:
     """get_template"""
 
@@ -108,15 +98,14 @@ def test_service_manager_partial_update(
         update_description: bool,
         update_hash: bool,
         tmp_path: Path,
-        random_string: str,
+        password: str,
     ) -> None:
         """Test operate.service_manager().update()"""
 
         operate = OperateApp(
             home=tmp_path / OPERATE,
         )
         operate.setup()
-        password = random_string
         operate.create_user_account(password=password)
         operate.password = password
         service_manager = operate.service_manager()
@@ -200,15 +189,14 @@ def test_service_manager_update(
         update_description: bool,
         update_hash: bool,
         tmp_path: Path,
-        random_string: str,
+        password: str,
     ) -> None:
         """Test operate.service_manager().update()"""
 
         operate = OperateApp(
             home=tmp_path / OPERATE,
         )
         operate.setup()
-        password = random_string
         operate.create_user_account(password=password)
         operate.password = password
         service_manager = operate.service_manager()