Merge pull request #1036 from IntersectMBO/chore/fix-backend-tests

Fix/Add tests for missing APIs

tests/govtool-backend/.env.example

@@ -1,7 +1,8 @@
-BASE_URL = `URL where the api is hosted`
+BASE_URL = "https://govtool.cardanoapi.io/api"
 RECORD_METRICS_API = `URL where metrics is posted`
 METRICS_API_SECRET= `api_secret`
 
 # required for setup
-KUBER_API_URL = ""
-KUBER_API_KEY = ""
+KUBER_API_URL = "https://kuber-govtool.cardanoapi.io"
+KUBER_API_KEY = "" # optional
+FAUCET_API_KEY= """

tests/govtool-backend/config.py

@@ -10,6 +10,6 @@
 dotenv.load_dotenv()
 
 RECORD_METRICS_API = os.getenv("RECORD_METRICS_API")
-METRICS_API_SECRET= os.getenv("METRICS_API_SECRET")
-KUBER_API_URL  = os.getenv("KUBER_API_URL")
-KUBER_API_KEY= os.getenv("KUBER_API_KEY")
+METRICS_API_SECRET = os.getenv("METRICS_API_SECRET")
+KUBER_API_URL = os.getenv("KUBER_API_URL")
+KUBER_API_KEY = os.getenv("KUBER_API_KEY")

tests/govtool-backend/lib/faucet_api.py

@@ -0,0 +1,41 @@
+import os
+from typing import TypedDict
+
+import requests
+
+
+class FaucetAmount(TypedDict):
+    lovelace: int
+
+
+class Transaction(TypedDict):
+    amount: FaucetAmount
+    txid: str
+    txin: str
+
+
+class CardanoFaucet:
+    def __init__(self, api_key: str, base_url: str = "https://faucet.sanchonet.world.dev.cardano.org"):
+        self.api_key = api_key
+        self.base_url = base_url
+
+    @staticmethod
+    def from_env():
+        api_key = os.getenv("FAUCET_API_KEY")
+        base_url = os.getenv("FAUCET_API_URL", "https://faucet.sanchonet.world.dev.cardano.org")
+        if not api_key:
+            raise ValueError("FAUCET_API_KEY environment variable not set.")
+        return CardanoFaucet(api_key, base_url)
+
+    def send_money(self, address: str, tx_type: str = "default") -> Transaction:
+        endpoint = f"{self.base_url}/send-money"
+        params = {"address": address, "api_key": self.api_key, "type": tx_type}
+        response = requests.get(endpoint, params=params)
+
+        if response.status_code == 200:
+            return response.json()
+        else:
+            response.raise_for_status()
+
+
+""

tests/govtool-backend/test_cases/govtool_api.py → tests/govtool-backend/lib/govtool_api.py

@@ -9,7 +9,7 @@
 from config import BUILD_ID
 
 
-class GovToolApi():
+class GovToolApi:
 
     def __init__(self, base_url: str):
         self._base_url = base_url
@@ -18,16 +18,15 @@ def __init__(self, base_url: str):
         self.requests_log = []
         self.tests_log = []
 
-    def __request(self, method: str, endpoint: str, param: Any | None = None,
-                  body: Any | None = None) -> Response:
-        endpoint = endpoint if endpoint.startswith('/') else '/' + endpoint
+    def __request(self, method: str, endpoint: str, param: Any | None = None, body: Any | None = None) -> Response:
+        endpoint = endpoint if endpoint.startswith("/") else "/" + endpoint
         full_url = self._base_url + endpoint
         full_url = full_url + "/" + param if param else full_url
-        start_time = int(time.time()*1000000)
+        start_time = int(time.time() * 1000000)
 
         response = self._session.request(method, full_url, json=body)
 
-        end_time = int(time.time()*1000000)
+        end_time = int(time.time() * 1000000)
         response_time = end_time - start_time
 
         try:
@@ -45,34 +44,57 @@ def __request(self, method: str, endpoint: str, param: Any | None = None,
             "response_json": response_json_str,
             "response_time": response_time,
             "start_date": int(start_time),
-            "build_id": BUILD_ID
+            "build_id": BUILD_ID,
         }
 
         self.requests_log.append(request_info)
 
-        assert 200 >= response.status_code <= 299, f"Expected {method}{endpoint} to succeed but got statusCode:{response.status_code} : body:{response.text}"
+        assert (
+            200 >= response.status_code <= 299
+        ), f"Expected {method}{endpoint} to succeed but got statusCode:{response.status_code} : body:{response.text}"
         return response
 
     def __get(self, endpoint: str, param: str | None = None) -> Response:
-        return self.__request('GET', endpoint, param)
+        return self.__request("GET", endpoint, param)
+
+    def __post(self, endpoint: str, param: str | None = None, body=None) -> Response:
+        return self.__request("POST", endpoint, param, body)
 
     def drep_list(self) -> Response:
-        return self.__get('/drep/list')
+        return self.__get("/drep/list")
+
+    def drep_info(self, drep_id) -> Response:
+        return self.__get("/drep/info", drep_id)
 
     def drep_getVotes(self, drep_id) -> Response:
-        return self.__get('/drep/getVotes', drep_id)
+        return self.__get("/drep/getVotes", drep_id)
 
     def drep_get_voting_power(self, drep_id) -> Response:
-        return self.__get('/drep/get-voting-power', drep_id)
+        return self.__get("/drep/get-voting-power", drep_id)
 
     def proposal_list(self) -> Response:
-        return self.__get('/proposal/list')
+        return self.__get("/proposal/list")
+
+    def get_proposal(self, id) -> Response:
+        return self.__get("/proposal/get", id)
 
     def ada_holder_get_current_delegation(self, stake_key: str) -> Response:
-        return self.__get('/ada-holder/get-current-delegation', stake_key)
+        return self.__get("/ada-holder/get-current-delegation", stake_key)
 
     def ada_holder_get_voting_power(self, stake_key) -> Response:
-        return self.__get('/ada-holder/get-voting-power', stake_key)
+        return self.__get("/ada-holder/get-voting-power", stake_key)
+
+    def epoch_params(self) -> Response:
+        return self.__get("/epoch/params")
+
+    def validate_metadata(self, metadata) -> Response:
+        return self.__post("/metadata/validate", body=metadata)
+
+    def network_metrics(self) -> Response:
+        return self.__get("/network/metrics")
+
+    def get_transaction_status(self, tx_id) -> Response:
+        return self.__get("/transaction/status", tx_id)
 
     def add_test_metrics(self, metrics: Metrics):
-        self.tests_log.append(metrics)
+        return self.tests_log.append(metrics)

tests/govtool-backend/models/TestData.py

@@ -1,22 +1,46 @@
-from typing import TypedDict
+from typing import TypedDict, Optional, List, Dict, Any
+
+
+class ProposalListResponse(TypedDict):
+    page: int
+    pageSize: int
+    total: int
+    elements: List["Proposal"]
+
+
+class GetProposalResponse(TypedDict):
+    votes: int
+    proposal: "Proposal"
 
 
 class Proposal(TypedDict):
     id: str
+    txHash: str
+    index: int
     type: str
-    details: str
+    details: Optional[dict]
     expiryDate: str
+    expiryEpochNo: int
+    createdDate: str
+    createdEpochNo: int
     url: str
     metadataHash: str
+    title: Optional[str]
+    about: Optional[str]
+    motivation: Optional[str]
+    rationale: Optional[str]
+    metadata: Optional[dict]
+    references: Optional[list]
     yesVotes: int
     noVotes: int
     abstainVotes: int
 
+
 class Drep(TypedDict):
     drepId: str
     url: str
     metadataHash: str
-    deposit : int
+    deposit: int
 
 
 class Delegation(TypedDict):
@@ -39,3 +63,71 @@ class Vote(TypedDict):
 class VoteonProposal(TypedDict):
     vote: Vote
     proposal: Proposal
+
+
+class DrepInfo(TypedDict):
+    isRegisteredAsDRep: bool
+    wasRegisteredAsDRep: bool
+    isRegisteredAsSoleVoter: bool
+    wasRegisteredAsSoleVoter: bool
+    deposit: int
+    url: str
+    dataHash: str
+    votingPower: Optional[int]
+    dRepRegisterTxHash: str
+    dRepRetireTxHash: Optional[str]
+    soleVoterRegisterTxHash: Optional[str]
+    soleVoterRetireTxHash: Optional[str]
+
+
+class EpochParam(TypedDict):
+    block_id: int
+    coins_per_utxo_size: int
+    collateral_percent: int
+    committee_max_term_length: int
+    committee_min_size: int
+    cost_model_id: int
+    decentralisation: int
+    drep_activity: int
+    drep_deposit: int
+    dvt_committee_no_confidence: float
+    dvt_committee_normal: float
+    dvt_hard_fork_initiation: float
+    dvt_motion_no_confidence: float
+    dvt_p_p_economic_group: float
+    dvt_p_p_gov_group: float
+    dvt_p_p_network_group: float
+    dvt_p_p_technical_group: float
+    dvt_treasury_withdrawal: float
+    dvt_update_to_constitution: float
+    epoch_no: int
+    extra_entropy: Optional[int]
+    gov_action_deposit: int
+    gov_action_lifetime: int
+    id: int
+    influence: float
+    key_deposit: int
+    max_bh_size: int
+    max_block_ex_mem: int
+    max_block_ex_steps: int
+    max_block_size: int
+    max_collateral_inputs: int
+    max_epoch: int
+    max_tx_ex_mem: int
+
+
+class TxStatus(TypedDict):
+    transactionConfirmed: bool
+
+
+class NetworkMetrics(TypedDict):
+    currentTime: str
+    currentEpoch: int
+    currentBlock: int
+    uniqueDelegators: int
+    totalDelegations: int
+    totalGovernanceActions: int
+    totalDRepVotes: int
+    totalRegisteredDReps: int
+    alwaysAbstainVotingPower: int
+    alwaysNoConfidenceVotingPower: int

tests/govtool-backend/setup.py

@@ -1,15 +1,11 @@
 import sys
 import requests
 import json
-from config import KUBER_API_URL, KUBER_API_KEY
 
+from lib.faucet_api import CardanoFaucet
+from lib.kuber_api import KuberApi
 
-if KUBER_API_URL is not None:
-    KUBER_API_URL = KUBER_API_URL[:-1] if KUBER_API_URL.endswith('/') else KUBER_API_URL
-    print(f"KUBER_API_URL: {KUBER_API_URL}")
-else:
-    print("KUBER_API_URL environment variable is not set.", file=sys.stderr)
-    sys.exit(1)
+kuber_api = KuberApi.from_env()
 
 # check fund for the main wallet
 main_wallet = {
@@ -100,6 +96,16 @@ def main():
             ada_wallets[0]["pay-skey"],
             ada_wallets[1]["stake-skey"],
             ada_wallets[1]["pay-skey"],
+            {
+                "type": "PaymentSigningKeyShelley_ed25519",
+                "description": "Payment Signing Key",
+                "cborHex": drep_wallets[0]["stake-skey"]["cborHex"],
+            },
+            {
+                "type": "PaymentSigningKeyShelley_ed25519",
+                "description": "Payment Signing Key",
+                "cborHex": drep_wallets[1]["stake-skey"]["cborHex"],
+            },
         ],
         "certificates": [
             {
@@ -123,7 +129,6 @@ def main():
         ],
         "proposals": [
             {
-                "deposit": 1000000000,
                 "refundAccount": {
                     "network": "Testnet",
                     "credential": {"key hash": ada_wallets[0]["stake-vkey"]},
@@ -136,22 +141,36 @@ def main():
             }
         ],
     }
-    kuber_url = KUBER_API_URL + "/api/v1/tx?submit=true"
-    print(json.dumps(kuber_json,indent=2))
+    print(json.dumps(kuber_json, indent=2))
     print("Submitting the above registration transaction..")
-    response = requests.post(
-        url=kuber_url, headers={"api-key": KUBER_API_KEY}, json=kuber_json
+    balance = kuber_api.get_balance(main_wallet["address"])
+    protocol_params = kuber_api.get_protocol_params()
+    total_locked = (
+            protocol_params["dRepDeposit"] * 2
+            + protocol_params["stakeAddressDeposit"] * 2
+            + protocol_params["govActionDeposit"]
     )
+    if balance < (total_locked + 10 * 10000000000000):
+        print("Loading balance to the bootstrap wallet")
+        faucet = CardanoFaucet.from_env()
+        result = faucet.send_money(main_wallet["address"])
+        if "error" in result:
+            print(result)
+            raise Exception("Failed to load balance from faucet")
+        kuber_api.wait_for_txout(result["txin"], log=True)
+    response = kuber_api.build_tx(kuber_json, submit=True)
+
     if response.status_code == 200:
         print("Transaction submitted", response.text)
+        data = response.json()
+        kuber_api.wait_for_txout(data["hash"] + '#0', log=True)
     else:
         print("Server Replied with Error [ StatusCode=", response.status_code, "]", response.reason, response.text)
-        if('DRepAlreadyRegistered' in response.text or 'StakeKeyRegisteredDELEG'):
+        if ("DRepAlreadyRegistered" in response.text) or ("StakeKeyRegisteredDELEG" in response.text):
             print("-----")
-            print("This might mean that you have already run the setup script.")
+            print("This probably means that you have already run the setup script.")
             print("-----")
             sys.exit(0)
-
         sys.exit(1)
 
     # vote from one of the dreps to the proposal
@@ -175,17 +194,16 @@ def main():
             },
         ],
     }
-    print(json.dumps(kuber_json,indent=2))
-    response = requests.post(
-        url=kuber_url, headers={"api-key": KUBER_API_KEY}, json=kuber_json
-    )
+    print(json.dumps(kuber_json, indent=2))
+    response = kuber_api.build_tx(kuber_json, submit=True)
     if response.status_code == 200:
         print("Transaction submitted", response.text)
+        data = response.json()
+        kuber_api.wait_for_txout(data["hash"] + '#0', log=True)
     else:
-        print("Server Replied with Error [ StatusCode=", response.status_code, "]", response.reason, response.text)
+        if "AlreadyRegistered" in response.text:
+            print("Server Replied with Error [ StatusCode=", response.status_code, "]", response.reason, response.text)
+            print("")
         sys.exit(1)
 
-    # write to the file in nice format
-\
-
 main()

tests/govtool-backend/test_cases/__init__.py

@@ -0,0 +1 @@
+from test_cases.fixtures import *