Add some tests / Rebase

Signed-off-by: jamshale <[email protected]>

aries_cloudagent/anoncreds/default/legacy_indy/registry.py

@@ -240,13 +240,6 @@ async def register_schema(
                     )
                 )
             except LedgerObjectAlreadyExistsError as err:
-                indy_schema = err.obj
-                schema = AnonCredsSchema(
-                    name=indy_schema["name"],
-                    version=indy_schema["version"],
-                    attr_names=indy_schema["attrNames"],
-                    issuer_id=indy_schema["id"].split(":")[0],
-                )
                 raise AnonCredsSchemaAlreadyExists(err.message, err.obj_id, schema)
             except (AnonCredsIssuerError, LedgerError) as err:
                 raise AnonCredsRegistrationError("Failed to register schema") from err

aries_cloudagent/anoncreds/default/legacy_indy/tests/test_registry.py

@@ -1,10 +1,28 @@
 """Test LegacyIndyRegistry."""
 
-import pytest
+import json
 import re
-from ..registry import LegacyIndyRegistry
+
+import pytest
+from asynctest import TestCase
 from base58 import alphabet
 
+from aries_cloudagent.anoncreds.base import AnonCredsSchemaAlreadyExists
+from aries_cloudagent.anoncreds.models.anoncreds_schema import (
+    AnonCredsSchema,
+    SchemaResult,
+)
+from aries_cloudagent.askar.profile_anon import AskarAnoncredsProfile
+from aries_cloudagent.connections.models.conn_record import ConnRecord
+from aries_cloudagent.core.in_memory.profile import InMemoryProfile
+from aries_cloudagent.ledger.error import LedgerObjectAlreadyExistsError
+from aries_cloudagent.protocols.endorse_transaction.v1_0.manager import (
+    TransactionManager,
+)
+from aries_cloudagent.tests import mock
+
+from .. import registry as test_module
+
 B58 = alphabet if isinstance(alphabet, str) else alphabet.decode("ascii")
 INDY_DID = rf"^(did:sov:)?[{B58}]{{21,22}}$"
 INDY_SCHEMA_ID = rf"^[{B58}]{{21,22}}:2:.+:[0-9.]+$"
@@ -33,23 +51,191 @@
     "WgWxqztrNooG92RXvxSTWv:4:WgWxqztrNooG92RXvxSTWv:3:CL:20:tag:CL_ACCUM:0"
 )
 
+mock_schema = AnonCredsSchema(
+    issuer_id="did:indy:sovrin:SGrjRL82Y9ZZbzhUDXokvQ",
+    attr_names=["name", "age", "vmax"],
+    name="test_schema",
+    version="1.0",
+)
 
-@pytest.fixture
-def registry():
-    """Registry fixture"""
-    yield LegacyIndyRegistry()
 
+@pytest.mark.anoncreds
+class TestLegacyIndyRegistry(TestCase):
+    def setUp(self):
+        self.profile = InMemoryProfile.test_profile(
+            settings={"wallet-type": "askar-anoncreds"},
+            profile_class=AskarAnoncredsProfile,
+        )
+        self.registry = test_module.LegacyIndyRegistry()
 
-@pytest.mark.indy
-class TestLegacyIndyRegistry:
-    @pytest.mark.asyncio
-    async def test_supported_did_regex(self, registry: LegacyIndyRegistry):
+    async def test_supported_did_regex(self):
         """Test the supported_did_regex."""
 
-        assert registry.supported_identifiers_regex == SUPPORTED_ID_REGEX
-        assert bool(registry.supported_identifiers_regex.match(TEST_INDY_DID))
-        assert bool(registry.supported_identifiers_regex.match(TEST_INDY_DID_1))
-        assert bool(registry.supported_identifiers_regex.match(TEST_INDY_SCHEMA_ID))
+        assert self.registry.supported_identifiers_regex == SUPPORTED_ID_REGEX
+        assert bool(self.registry.supported_identifiers_regex.match(TEST_INDY_DID))
+        assert bool(self.registry.supported_identifiers_regex.match(TEST_INDY_DID_1))
+        assert bool(
+            self.registry.supported_identifiers_regex.match(TEST_INDY_SCHEMA_ID)
+        )
         assert bool(
-            registry.supported_identifiers_regex.match(TEST_INDY_REV_REG_DEF_ID)
+            self.registry.supported_identifiers_regex.match(TEST_INDY_REV_REG_DEF_ID)
+        )
+
+    async def test_register_schema_no_endorsement(self):
+        self.profile.inject_or = mock.MagicMock(
+            return_value=mock.CoroutineMock(
+                send_schema_anoncreds=mock.CoroutineMock(return_value=1)
+            )
+        )
+
+        result = await self.registry.register_schema(self.profile, mock_schema, {})
+
+        assert isinstance(result, SchemaResult)
+
+    @mock.patch.object(
+        ConnRecord,
+        "retrieve_by_id",
+        return_value=mock.CoroutineMock(
+            metadata_get=mock.CoroutineMock(return_value={"endorser_did": "test_did"})
+        ),
+    )
+    async def test_register_schema_with_author_role(self, mock_endorser_conn_record):
+        self.profile.inject_or = mock.MagicMock(
+            return_value=mock.CoroutineMock(
+                send_schema_anoncreds=mock.CoroutineMock(
+                    return_value=(
+                        "test_schema_id",
+                        {
+                            "signed_txn": "test_signed_txn",
+                        },
+                    )
+                )
+            )
         )
+        self.profile.settings.set_value("endorser.author", True)
+
+        result = await self.registry.register_schema(
+            self.profile, mock_schema, {"endorser_connection_id": "test_connection_id"}
+        )
+
+        assert result["sent"] is not None
+        assert result["txn"] is not None
+        assert mock_endorser_conn_record.called
+
+    @mock.patch.object(
+        ConnRecord,
+        "retrieve_by_id",
+        return_value=mock.CoroutineMock(
+            metadata_get=mock.CoroutineMock(return_value={"endorser_did": "test_did"})
+        ),
+    )
+    async def test_register_schema_already_exists(self, mock_endorser_conn_record):
+        self.profile.inject_or = mock.MagicMock(
+            return_value=mock.CoroutineMock(
+                send_schema_anoncreds=mock.CoroutineMock(
+                    side_effect=LedgerObjectAlreadyExistsError(
+                        "test", "test", mock_schema
+                    )
+                )
+            )
+        )
+        self.profile.settings.set_value("endorser.author", True)
+
+        with self.assertRaises(AnonCredsSchemaAlreadyExists):
+            await self.registry.register_schema(
+                self.profile,
+                mock_schema,
+                {"endorser_connection_id": "test_connection_id"},
+            )
+
+    @mock.patch.object(
+        ConnRecord,
+        "retrieve_by_id",
+        return_value=mock.CoroutineMock(
+            metadata_get=mock.CoroutineMock(return_value={"endorser_did": "test_did"})
+        ),
+    )
+    async def test_register_schema_with_create_trasaction_param(
+        self, mock_endorser_conn_record
+    ):
+        self.profile.inject_or = mock.MagicMock(
+            return_value=mock.CoroutineMock(
+                send_schema_anoncreds=mock.CoroutineMock(
+                    return_value=(
+                        "test_schema_id",
+                        {
+                            "signed_txn": "test_signed_txn",
+                        },
+                    )
+                )
+            )
+        )
+
+        result = await self.registry.register_schema(
+            self.profile,
+            mock_schema,
+            {
+                "endorser_connection_id": "test_connection_id",
+                "create_transaction_for_endorser": True,
+            },
+        )
+
+        assert result["sent"] is not None
+        assert result["txn"] is not None
+        assert mock_endorser_conn_record.called
+
+    @mock.patch.object(
+        ConnRecord,
+        "retrieve_by_id",
+        return_value=mock.CoroutineMock(
+            metadata_get=mock.CoroutineMock(return_value={"endorser_did": "test_did"})
+        ),
+    )
+    @mock.patch.object(
+        TransactionManager,
+        "create_request",
+    )
+    async def test_register_schema_with_author_role_and_create_request(
+        self, mock_create_request, mock_endorser_conn_record
+    ):
+        class MockTransaction:
+            def __init__(self, txn):
+                self.txn = txn
+
+            def serialize(self):
+                return json.dumps(self.txn)
+
+        self.profile.inject_or = mock.MagicMock(
+            return_value=mock.CoroutineMock(
+                send_schema_anoncreds=mock.CoroutineMock(
+                    return_value=(
+                        "test_schema_id",
+                        {
+                            "signed_txn": "test_signed_txn",
+                        },
+                    )
+                )
+            )
+        )
+
+        mock_create_request.return_value = (
+            MockTransaction({"test": "test"}),
+            "transaction_request",
+        )
+
+        self.profile.settings.set_value("endorser.author", True)
+        self.profile.settings.set_value("endorser.auto_request", True)
+
+        mock_outbound_handler = mock.CoroutineMock()
+
+        result = await self.registry.register_schema(
+            self.profile,
+            mock_schema,
+            {"endorser_connection_id": "test_connection_id"},
+            mock_outbound_handler,
+        )
+
+        assert result["sent"] is not None
+        assert result["txn"] is not None
+        assert mock_create_request.called
+        assert mock_outbound_handler.called