WIP unit tests for Presentation class. Some refactoring of the class.

morrieinmaas · morrieinmaas · commit 4a27c36e9364 · 2021-04-09T04:15:24.000Z
diff --git a/.gitignore b/.gitignore
@@ -1,3 +1,4 @@
+.coverage
 .idea/
 .code
 dist
@@ -17,4 +18,4 @@ libs/om-aries-controller/demo/bob/image_received.png
 **/model.pt
 **/trained_model.pt
 **/untrained_model.pt
-**/part_trained.pt
+**/part_trained.pt
diff --git a/libs/aries-basic-controller/aries_basic_controller/models/presentation.py b/libs/aries-basic-controller/aries_basic_controller/models/presentation.py
@@ -3,103 +3,148 @@
 
 logger = logging.getLogger("aries_controller.models.presentation")
 
+
 class Presentation():
     def __init__(self, presentation_json):
-        # TODO Load verify var with json.loads() and test code
-        self.json_data = presentation_json
-        if self.is_verified() == "false":
-            logger.error('Presentation Object Verification Failed')
-            # TODO raise proper PresentationVerificationFailed exception
-            raise Exception('Presentation Object Verification Failed')
+        try:
+            self.json_data = json.loads(presentation_json)
+        except Exception:
+            logger.error("Failed to load presentation JSON")
+            raise
+        self.validate_presentation_json(presentation_json)
+        if not self.is_verified():
+            logger.error('Presentation Object Verification not verified')
+            raise Exception('Presentation Object Verification not verified')
 
     def get_self_attested(self):
-      if self.is_verified() == "false":
-          logger.debug('Verification Failed')
-      data = {}
-      for (name, val) in self.json_data['presentation']['requested_proof']['self_attested_attrs'].items():
-        data[name] = val['raw']
-      return json.dumps(data)
+        if self.is_verified():
+            data = {}
+            for (name, val) in self.json_data['presentation']['requested_proof']['self_attested_attrs'].items():
+                data[name] = val['raw']
+            return json.dumps(data)
 
     def get_revealed(self):
-      if self.is_verified() == "false":
-          logger.debug('Verification Failed')
-      data = {}
-      for (name, val) in self.json_data['presentation']['requested_proof']['revealed_attrs'].items():
-          data[name] = val['raw']
-
-      return json.dumps(data)
+        if self.is_verified():
+            data = {}
+            for (name, val) in self.json_data['presentation']['requested_proof']['revealed_attrs'].items():
+                data[name] = val['raw']
+            return json.dumps(data)
 
     def get_unrevealed_attrs(self):
-      if self.is_verified() == "false":
-          logger.debug('Verification Failed')
-      data = {}
-      for (name, val) in self.json_data['presentation']['requested_proof']['unrevealed_attrs'].items():
-        data[name] = val['raw']
-      return json.dumps(data)
+        if self.is_verified():
+            data = {}
+            for (name, val) in self.json_data['presentation']['requested_proof']['unrevealed_attrs'].items():
+                data[name] = val['raw']
+            return json.dumps(data)
 
     def get_predicates(self):
-      if self.is_verified() == "false":
-          logger.debug('Verification Failed')
-      data = {}
-      for (name, val) in self.json_data['presentation']['requested_proof']['predicates'].items():
-        data[name] = val['raw']
-      return json.dumps(data)
+        if self.is_verified():
+            data = {}
+            for (name, val) in self.json_data['presentation']['requested_proof']['predicates'].items():
+                data[name] = val['raw']
+            return json.dumps(data)
 
     def get_identifiers(self):
-        data = {}
-        identifiers = []
-        for index in range(len(self.json_data['presentation']['identifiers'])):
-            identifiers.extend([self.json_data['presentation']['identifiers'][index]])
-        data['identifiers'] = identifiers
-        return json.dumps(data)
+        if self.__has_identifiers():
+            data = {}
+            identifiers = []
+            for index in range(len(self.json_data['presentation']['identifiers'])):
+                identifiers.extend([self.json_data['presentation']['identifiers'][index]])
+            data['identifiers'] = identifiers
+            return json.dumps(data)
 
     def get_schemas(self):
-        data = {}
-        creds = []
-        for index in range(len(self.json_data['presentation']['identifiers'])):
-            for key in self.json_data['presentation']['identifiers'][index]:
-                if key == 'schema_id':
-                    creds.extend([self.json_data['presentation']['identifiers'][index][key]])
-        data['schema_id'] = creds
-        return json.dumps(data)
+        if self.__has_identifiers():
+            data = {}
+            creds = []
+            for index in range(len(self.json_data['presentation']['identifiers'])):
+                for key in self.json_data['presentation']['identifiers'][index]:
+                    if key == 'schema_id':
+                        creds.extend([self.json_data['presentation']['identifiers'][index][key]])
+            data['schema_id'] = creds
+            return json.dumps(data)
 
     def get_cred_def_ids(self):
-        data = {}
-        creds = []
-        for index in range(len(self.json_data['presentation']['identifiers'])):
-            for key in self.json_data['presentation']['identifiers'][index]:
-                if key == 'cred_def_id':
-                    creds.extend([self.json_data['presentation']['identifiers'][index][key]])
-        data['cred_def_id'] = creds
-        return json.dumps(data)
+        if self.__has_identifiers():
+            data = {}
+            creds = []
+            for index in range(len(self.json_data['presentation']['identifiers'])):
+                for key in self.json_data['presentation']['identifiers'][index]:
+                    if key == 'cred_def_id':
+                        creds.extend([self.json_data['presentation']['identifiers'][index][key]])
+            data['cred_def_id'] = creds
+            return json.dumps(data)
 
     def get_rev_reg_ids(self):
-        data = {}
-        creds = []
-        for index in range(len(self.json_data['presentation']['identifiers'])):
-            for key in self.json_data['presentation']['identifiers'][index]:
-                if key == 'rev_reg_id':
-                    creds.extend([self.json_data['presentation']['identifiers'][index][key]])
-        data['rev_reg_id'] = creds
-        return json.dumps(data)
+        if self.__has_identifiers():
+            data = {}
+            creds = []
+            for index in range(len(self.json_data['presentation']['identifiers'])):
+                for key in self.json_data['presentation']['identifiers'][index]:
+                    if key == 'rev_reg_id':
+                        creds.extend([self.json_data['presentation']['identifiers'][index][key]])
+            data['rev_reg_id'] = creds
+            return json.dumps(data)
 
     def get_role(self):
-      return(self.json_data['role'])
+        return(self.json_data['role'])
 
     def get_threadid(self):
-      return(self.json_data['thread_id'])
+        return(self.json_data['thread_id'])
 
     def get_presentation_request(self):
-      return(self.json_data['presentation_request'])
+        return(self.json_data['presentation_request'])
 
     def get_verified_state(self):
-      return self.json_data['state']
+        return self.json_data['state']
 
     def get_presxid(self):
-      return self.json_data['presentation_exchange_id']
-
-    def is_verified(self):
-      return self.json_data['verified']
+        return self.json_data['presentation_exchange_id']
 
     def from_conn_id(self):
-        return self.json_data['connection_id']
+        return self.json_data['connection_id']
+
+    def is_verified(self):
+        try:
+            assert self.json_data['verified'] == "true"
+        except AssertionError:
+            logger.debug('Verification Failed')
+            return False
+        return True
+
+    def validate_presentation_json(self, presentation_json):
+        try:
+            # Taken from sample response in swagger UI
+            # TODO Determine whether this is the minimal set of keys
+            presentation_keys = [
+                "auto_present",
+                "connection_id",
+                "created_at",
+                "error_msg",
+                "initiator",
+                "presentation",
+                "presentation_exchange_id",
+                "presentation_proposal_dict",
+                "presentation_request",
+                "presentation_request_dict",
+                "role",
+                "state",
+                "thread_id",
+                "trace",
+                "updated_at",
+                "verified"
+            ]
+            for key in presentation_keys:
+                assert key in json.loads(presentation_json),\
+                    f"Invalid presentation. Missing key {key}"
+        except AssertionError:
+            raise
+
+    def __has_identifiers(self):
+        try:
+            assert 'identifiers' in self.json_data['presentation'],\
+                "No key 'identifiers' in presentation"
+            return True
+        except AssertionError:
+            logger.warning("No key 'identifiers' in presentation")
+            raise
diff --git a/libs/aries-basic-controller/aries_basic_controller/tests/__init__.py b/libs/aries-basic-controller/aries_basic_controller/tests/__init__.py
diff --git a/libs/aries-basic-controller/aries_basic_controller/tests/test_presentation.py b/libs/aries-basic-controller/aries_basic_controller/tests/test_presentation.py
@@ -0,0 +1,116 @@
+import pytest
+import json
+import logging
+
+from ..models.presentation import Presentation
+
+LOGGER = logging.getLogger(__name__)
+
+presentation_keys = [
+                "auto_present",
+                "connection_id",
+                "created_at",
+                "error_msg",
+                "initiator",
+                "presentation",
+                "presentation_exchange_id",
+                "presentation_proposal_dict",
+                "presentation_request",
+                "presentation_request_dict",
+                "role",
+                "state",
+                "thread_id",
+                "trace",
+                "updated_at",
+                "verified"
+            ]
+
+# From sample response in Swagger UI
+pytest.valid_presentation = json.dumps({
+    "auto_present": "false",
+    "connection_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+    "created_at": "2021-04-08 09:04:01Z",
+    "error_msg": "Invalid structure",
+    "initiator": "self",
+    "presentation": {},
+    "presentation_exchange_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+    "presentation_proposal_dict": {},
+    "presentation_request": {},
+    "presentation_request_dict": {},
+    "role": "prover",
+    "state": "verified",
+    "thread_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+    "trace": "true",
+    "updated_at": "2021-04-08 09:04:01Z",
+    "verified": "true"
+    })
+
+pytest.unverified_presentation = json.dumps({
+    "auto_present": "false",
+    "connection_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+    "created_at": "2021-04-08 09:04:01Z",
+    "error_msg": "Invalid structure",
+    "initiator": "self",
+    "presentation": {},
+    "presentation_exchange_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+    "presentation_proposal_dict": {},
+    "presentation_request": {},
+    "presentation_request_dict": {},
+    "role": "prover",
+    "state": "verified",
+    "thread_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+    "trace": "true",
+    "updated_at": "2021-04-08 09:04:01Z",
+    "verified": "false"
+    })
+
+pytest.invalid_presentation = json.dumps({
+    "abc": "1234"
+    })
+
+pytest.not_json = 123456
+
+
+def test_init_valid():
+    pres = Presentation(pytest.valid_presentation)
+    assert pres.json_data == json.loads(pytest.valid_presentation)
+
+
+def test_init_invalid():
+    with pytest.raises(AssertionError,
+                       match="Missing key"):
+        Presentation(pytest.invalid_presentation)
+
+
+def test_init_unverified(caplog):
+    error_msg = "Presentation Object Verification not verified"
+    with pytest.raises(Exception,
+                       match=error_msg):
+        Presentation(pytest.unverified_presentation)
+    assert error_msg in caplog.text
+
+
+def test_init_not_json(caplog):
+    error_msg = "the JSON object must be str, bytes or bytearray, not int"
+    with pytest.raises(TypeError, match=error_msg):
+        Presentation(pytest.not_json)
+    assert "Failed to load presentation JSON" in caplog.text
+
+
+# check for each possibly missing key whether the correct error is raised
+@pytest.mark.parametrize("key", presentation_keys)
+def test_validate_presentation_json(key):
+    pres = pytest.valid_presentation
+    pres_obj = Presentation(pres)
+    with pytest.raises(AssertionError,
+                       match=f"Missing key {key}"):
+        del pres_obj.json_data[key]
+        pres_obj.validate_presentation_json(json.dumps(pres_obj.json_data))
+
+
+def test_has_no_identifiers(caplog):
+    error_msg = "No key 'identifiers' in presentation"
+    with pytest.raises(AssertionError,
+                       match=error_msg):
+        Presentation(pytest.valid_presentation).get_cred_def_ids()
+    assert error_msg in caplog.text
