test: Add integration test

tests/e2e-pytest/helpers.py

@@ -0,0 +1,44 @@
+import copy
+import sys
+import pexpect
+import json
+
+
+RESOLVER_URL = "http://localhost:1313"
+PATH = "/1.0/identifiers/"
+
+TESTNET_DID = "did:cheqd:testnet:zFWM1mKVGGU2gHYuLAQcTJfZBebqBpGf"
+TESTNET_FRAGMENT = TESTNET_DID + "#key1"
+FAKE_TESTNET_DID = "did:cheqd:testnet:zF7rhDBfUt9d1gJPjx7s1JXfUY7oVWkY"
+FAKE_TESTNET_FRAGMENT = TESTNET_DID + "#fake_key"
+
+MAINNET_DID = "did:cheqd:mainnet:zF7rhDBfUt9d1gJPjx7s1JXfUY7oVWkY"
+MAINNET_FRAGMENT = MAINNET_DID + "#key1"
+FAKE_MAINNET_DID = "did:cheqd:mainnet:zFWM1mKVGGU2gHYuLAQcTJfZBebqBpGf"
+FAKE_MAINNET_FRAGMENT = MAINNET_DID + "#fake_key"
+
+
+IMPLICIT_TIMEOUT = 40
+ENCODING = "utf-8"
+READ_BUFFER = 60000
+
+
+def run(command, params, expected_output):
+    cli = pexpect.spawn(f"{command} {params}", encoding=ENCODING, timeout=IMPLICIT_TIMEOUT, maxread=READ_BUFFER)
+    cli.logfile = sys.stdout
+    cli.expect(expected_output)
+    return cli
+
+
+def run_interaction(cli, input_string, expected_output):
+    cli.sendline(input_string)
+    cli.expect(expected_output)
+
+
+def json_loads(s_to_load: str) -> dict:
+    s = copy.copy(s_to_load)
+    s = s.replace("\\", "")
+    s = s.replace("\"[", "[")
+    s = s.replace("]\"", "]")
+    return json.loads(s)
+

tests/e2e-pytest/requirements.txt

@@ -0,0 +1,3 @@
+pytest==6.0.2
+pytest-asyncio==0.16.0
+pexpect==4.8.0

tests/e2e-pytest/test_cosmos.py

@@ -0,0 +1,31 @@
+import pytest
+from helpers import run, TESTNET_DID, MAINNET_DID, TESTNET_FRAGMENT, MAINNET_FRAGMENT, \
+    FAKE_TESTNET_DID, FAKE_MAINNET_DID, FAKE_TESTNET_FRAGMENT, FAKE_MAINNET_FRAGMENT, RESOLVER_URL, PATH
+
+
+@pytest.mark.parametrize(
+    "did_url, expected_output",
+    [
+        (TESTNET_DID, fr"didDocument(.*?)\"id\":\"{TESTNET_DID}\"(.*?)didDocumentMetadata"
+                      r"(.*?)didResolutionMetadata"),
+        (MAINNET_DID, fr"didDocument(.*?)\"id\":\"{MAINNET_DID}\"(.*?)didDocumentMetadata"
+                      r"(.*?)didResolutionMetadata"),
+        (FAKE_TESTNET_DID, r"didDocument\":null,\"didDocumentMetadata\":\[\],"
+                           r"\"didResolutionMetadata(.*?)\"error\":\"notFound\""),
+        (FAKE_MAINNET_DID, r"didDocument\":null,\"didDocumentMetadata\":\[\],"
+                           r"\"didResolutionMetadata(.*?)\"error\":\"notFound\""),
+        ("did:wrong_method:MTMxDQKMTMxDQKMT", r"didDocument\":null,\"didDocumentMetadata\":\[\],"
+                                              r"\"didResolutionMetadata(.*?)\"error\":\"methodNotSupported\""),
+
+        (TESTNET_FRAGMENT, fr"\"contentStream\":(.*?)\"id\":\"{TESTNET_FRAGMENT}\"(.*?)contentMetadata"
+                           r"(.*?)dereferencingMetadata\""),
+        (MAINNET_FRAGMENT, fr"\"contentStream\":(.*?)\"id\":\"{MAINNET_FRAGMENT}\"(.*?)contentMetadata"
+                           r"(.*?)dereferencingMetadata\""),
+        (FAKE_TESTNET_FRAGMENT, r"\"contentStream\":null,\"contentMetadata\":\[\],"
+                                r"\"dereferencingMetadata(.*?)\"error\":\"FragmentNotFound\""),
+        (FAKE_MAINNET_FRAGMENT, r"\"contentStream\":null,\"contentMetadata\":\[\],"
+                                r"\"dereferencingMetadata(.*?)\"error\":\"FragmentNotFound\""),
+    ]
+)
+def test_resolution(did_url, expected_output):
+    run("curl", RESOLVER_URL + PATH + did_url.replace("#", "%23"), expected_output)