forked from fetchai/cosmpy
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Tx generic signing & Tx module (the first take) (fetchai#12)
- Loading branch information
Showing
15 changed files
with
333 additions
and
70 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,15 @@ | ||
from abc import ABC, abstractmethod | ||
|
||
|
||
class Signer(ABC): | ||
@abstractmethod | ||
def sign( | ||
self, message: bytes, deterministic=False, canonicalise: bool = True | ||
) -> bytes: | ||
... | ||
|
||
@abstractmethod | ||
def sign_digest( | ||
self, digest: bytes, deterministic=False, canonicalise: bool = True | ||
) -> bytes: | ||
... |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,27 +1,30 @@ | ||
import urllib.error | ||
import urllib.request | ||
import requests | ||
|
||
|
||
class QueryRestClient: | ||
def __init__(self, rest_address: str): | ||
self._session = requests.session() | ||
self.rest_address = rest_address | ||
|
||
def get(self, request: str) -> str: | ||
url = self.rest_address + request | ||
|
||
try: | ||
with urllib.request.urlopen(url) as f: | ||
response = f.read().decode("utf-8") | ||
return response | ||
except urllib.error.HTTPError as e: | ||
raise RuntimeError( | ||
f"HTTPError when sending a get request.\n Request: {request}\n Response: {e.code}, {str(e.read().decode('utf-8'))})" | ||
) | ||
except urllib.error.URLError as e: | ||
def get(self, request: str) -> bytes: | ||
response = self._session.get(url=self.rest_address + request) | ||
if response.status_code != 200: | ||
raise RuntimeError( | ||
f"URLError when sending a get request.\n Request: {request}, Exception: {e})" | ||
f"Error when sending a query request.\n Request: {request}\n Response: {response.status_code}, {str(response.content)})" | ||
) | ||
except Exception as e: | ||
return response.content | ||
|
||
def post(self, url_path, json_request: dict) -> bytes: | ||
headers = {"Content-type": "application/json", "Accept": "application/json"} | ||
response = self._session.post( | ||
url=self.rest_address + url_path, json=json_request, headers=headers | ||
) | ||
|
||
if response.status_code != 200: | ||
raise RuntimeError( | ||
f"Exception during sending a get request.\n Request: {request}, Exception: {e})" | ||
f"Error when sending a query request.\n Request: {json_request}\n Response: {response.status_code}, {str(response.content)})" | ||
) | ||
return response.content | ||
|
||
def __del__(self): | ||
self._session.close() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,26 +1,45 @@ | ||
from cosm.query.rest_client import QueryRestClient | ||
from unittest import TestCase | ||
import mock | ||
from unittest.mock import Mock, patch | ||
from requests import Session, Response | ||
|
||
|
||
class QueryTests(TestCase): | ||
@mock.patch("urllib.request.urlopen", autospec=True) | ||
def test_get(self, mock_urlopen): | ||
client = QueryRestClient("address") | ||
client.get("/request") | ||
|
||
self.assertEqual(mock_urlopen.call_args_list[0].args, ("address/request",)) | ||
|
||
def test_get_url_error(self): | ||
client = QueryRestClient("https://127.0.0.1") | ||
try: | ||
client.get("/request") | ||
except RuntimeError as e: | ||
assert "URLError" in str(e) | ||
|
||
def test_get_value_error(self): | ||
client = QueryRestClient("address") | ||
try: | ||
client.get("/request") | ||
except RuntimeError as e: | ||
assert "unknown url type" in str(e) | ||
@patch("requests.session", spec=Session) | ||
def test_get_pass(self, session_mock): | ||
rest_address = "some url" | ||
client = QueryRestClient(rest_address) | ||
|
||
session_mock.assert_called_once_with() | ||
assert client.rest_address == rest_address | ||
|
||
request_url_path = "my weird url path" | ||
resp = Mock(spec=Response) | ||
resp.status_code = 200 | ||
resp.content = "dfdffdss".encode(encoding="utf8") | ||
|
||
session_mock.return_value.get.return_value = resp | ||
client.get(request_url_path) | ||
session_mock.return_value.get.assert_called_once_with( | ||
url=rest_address + request_url_path | ||
) | ||
|
||
@patch("requests.session", spec=Session) | ||
def test_get_error(self, session_mock): | ||
rest_address = "some url" | ||
client = QueryRestClient(rest_address) | ||
|
||
session_mock.assert_called_once_with() | ||
assert client.rest_address == rest_address | ||
|
||
request_url_path = "my weird url path" | ||
resp = Mock(spec=Response) | ||
resp.status_code = 400 | ||
resp.content = "dfdffdss".encode(encoding="utf8") | ||
|
||
session_mock.return_value.get.return_value = resp | ||
|
||
with self.assertRaises(Exception) as context: | ||
client.get(request_url_path) | ||
|
||
self.assertTrue("Error when sending a query request" in str(context.exception)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,24 @@ | ||
from cosm.crypto.interface import Signer | ||
from cosmos.tx.v1beta1.tx_pb2 import Tx, SignDoc | ||
|
||
|
||
def sign_transaction( | ||
tx: Tx, | ||
signer: Signer, | ||
chain_id: str, | ||
account_number: int, | ||
deterministic: bool = False, | ||
): | ||
sd = SignDoc() | ||
sd.body_bytes = tx.body.SerializeToString() | ||
sd.auth_info_bytes = tx.auth_info.SerializeToString() | ||
sd.chain_id = chain_id | ||
sd.account_number = account_number | ||
|
||
data_for_signing = sd.SerializeToString() | ||
|
||
# Generating deterministic signature: | ||
signature = signer.sign( | ||
data_for_signing, deterministic=deterministic, canonicalise=True | ||
) | ||
tx.signatures.extend([signature]) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,24 @@ | ||
from abc import ABC, abstractmethod | ||
import cosmos.tx.v1beta1.service_pb2 as svc | ||
|
||
|
||
class RPCInterface(ABC): | ||
# Simulate simulates executing a transaction for estimating gas usage. | ||
@abstractmethod | ||
def Simulate(self, request: svc.SimulateRequest) -> svc.SimulateResponse: | ||
... | ||
|
||
# GetTx fetches a tx by hash. | ||
@abstractmethod | ||
def GetTx(self, request: svc.GetTxRequest) -> svc.GetTxResponse: | ||
... | ||
|
||
# BroadcastTx broadcast transaction. | ||
@abstractmethod | ||
def BroadcastTx(self, request: svc.BroadcastTxRequest) -> svc.BroadcastTxResponse: | ||
... | ||
|
||
# GetTxsEvent fetches txs by event. | ||
@abstractmethod | ||
def GetTxsEvent(self, request: svc.GetTxsEventRequest) -> svc.GetTxsEventResponse: | ||
... |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,50 @@ | ||
from cosm.query.rest_client import QueryRestClient as RestClient | ||
from cosm.tx.interface import RPCInterface | ||
|
||
from cosmos.tx.v1beta1.service_pb2 import ( | ||
SimulateRequest, | ||
SimulateResponse, | ||
GetTxRequest, | ||
GetTxResponse, | ||
BroadcastTxRequest, | ||
BroadcastTxResponse, | ||
GetTxsEventRequest, | ||
GetTxsEventResponse, | ||
) | ||
from google.protobuf.json_format import MessageToDict, Parse | ||
from urllib.parse import urlencode | ||
|
||
|
||
class TxRestClient(RPCInterface): | ||
txs_url_path = "/cosmos/tx/v1beta1/txs" | ||
|
||
def __init__(self, rest_client: RestClient): | ||
self.rest_client = rest_client | ||
|
||
# Simulate simulates executing a transaction for estimating gas usage. | ||
def Simulate(self, request: SimulateRequest) -> SimulateResponse: | ||
json_request = MessageToDict(request) | ||
response = self.rest_client.post("/cosmos/tx/v1beta1/simulate", json_request) | ||
return Parse(response, SimulateResponse()) | ||
|
||
# GetTx fetches a tx by hash. | ||
def GetTx(self, request: GetTxRequest) -> GetTxResponse: | ||
json_request = MessageToDict(request) | ||
url_encoded_request = urlencode(json_request) | ||
response = self.rest_client.get( | ||
f"{self.txs_url_path}&{url_encoded_request}", | ||
) | ||
return Parse(response, GetTxResponse()) | ||
|
||
# BroadcastTx broadcast transaction. | ||
def BroadcastTx(self, request: BroadcastTxRequest) -> BroadcastTxResponse: | ||
json_request = MessageToDict(request) | ||
response = self.rest_client.post(self.txs_url_path, json_request) | ||
return Parse(response, BroadcastTxResponse()) | ||
|
||
# GetTxsEvent fetches txs by event. | ||
def GetTxsEvent(self, request: GetTxsEventRequest) -> GetTxsEventResponse: | ||
json_request = MessageToDict(request) | ||
url_encoded_request = urlencode(json_request) | ||
response = self.rest_client.get(f"{self.txs_url_path}&{url_encoded_request}") | ||
return Parse(response, GetTxsEventResponse()) |
Oops, something went wrong.