Feat/xcm_tests

makefile

@@ -70,6 +70,9 @@ test-eth-availability:
 test-new-assets:
 	$(TEST_RUN) "./tests/test_check_new_assets.py"
 
+test-xcm-delivery-factor:
+	$(TEST_RUN) "./tests/test_xcm_delivery_fee.py"
+
 allure:
 	allure serve $(ALLURE_DIR)
 

scripts/utils/chain_model.py

@@ -42,3 +42,6 @@ def init_properties(self):
         if (self.properties):
                 return self.substrate
         self.properties = get_properties(self.substrate)
+
+    def is_parachain(self) -> bool:
+        return self.parentId is not None

tests/conftest.py

@@ -1,12 +1,13 @@
 import pytest
+import allure
 
 from func_timeout import func_timeout, FunctionTimedOut
 from scripts.utils.chain_model import Chain
 from scripts.utils.metadata_interaction import get_properties
 from scripts.utils.substrate_interface import create_connection_by_url
 from tests.data.setting_data import collect_nodes_for_chains, get_substrate_chains
 
-FIXTURE_TIMEOUT = 10
+FIXTURE_TIMEOUT = 15
 
 chain_names = [
     f'Test for {task.name}'
@@ -17,11 +18,18 @@
     f'Test for {task["name"]}, url: {task["url"]}'
     for task in collect_nodes_for_chains(get_substrate_chains())
 ]
+
+class FixtureTimeoutError(Exception):
+    pass
+
 def execute_with_timeout(timeout: int, function, args: tuple, error_message: str):
     try:
         return func_timeout(timeout=timeout, func=function, args=args)
     except FunctionTimedOut:
-        pytest.fail(msg=error_message)
+        with allure.step("Function timed out"):
+            allure.attach(name="Error", body=error_message, attachment_type=allure.attachment_type.TEXT)
+        raise FixtureTimeoutError(error_message)
+
 
 @pytest.fixture(scope="module", params=collect_nodes_for_chains(get_substrate_chains()), ids=task_ids)
 def connection_by_url(request):

tests/data/setting_data.py

@@ -1,16 +1,18 @@
 import os
+import json
 
 from typing import List
 from scripts.utils.work_with_data import get_network_list
 from scripts.utils.chain_model import Chain
 
-network_file_path = os.getenv('CHAINS_JSON_PATH', "chains/v15/chains.json")
+network_file_path = os.getenv('CHAINS_JSON_PATH', "chains/v18/chains.json")
+xcm_file_path = os.getenv('DEV_XCM_JSON_PATH', "xcm/v6/transfers_dev.json")
 skipped_networks = ['Edgeware']
 network_list = get_network_list('/' + network_file_path)
 
 
 def get_substrate_chains() -> list[Chain]:
-    substrate_chains = []
+    substrate_chains: list[Chain] = []
     for data in network_list:
         if data.get('name') in skipped_networks or 'PAUSED' in data.get('name'):
             continue
@@ -26,6 +28,44 @@ def get_substrate_chains() -> list[Chain]:
     return substrate_chains
 
 
+def load_xcm_data():
+    with open(xcm_file_path) as f:
+        return json.load(f)
+
+
+def get_parachains() -> List[Chain]:
+    substrate_chains = get_substrate_chains()
+    parachains = [chain for chain in substrate_chains if chain.is_parachain()]
+    return parachains
+
+
+def get_relaychains() -> List[Chain]:
+    substrate_chains = get_substrate_chains()
+    relaychains = [
+        chain for chain in substrate_chains if not chain.is_parachain()]
+    return relaychains
+
+
+def get_parachains_with_xcm() -> List[Chain]:
+    substrate_parachains = get_parachains()
+    xcm_data = load_xcm_data()
+    chain_ids = [chain['chainId'] for chain in xcm_data['chains']]
+    parachains_with_xcm = [
+        chain for chain in substrate_parachains if chain.chainId in chain_ids]
+
+    return parachains_with_xcm
+
+
+def get_relaychains_with_xcm() -> List[Chain]:
+    substrate_relaychains = get_relaychains()
+    xcm_data = load_xcm_data()
+    chain_ids = [chain['chainId'] for chain in xcm_data['chains']]
+    relaychains_with_xcm = [
+        chain for chain in substrate_relaychains if chain.chainId in chain_ids]
+
+    return relaychains_with_xcm
+
+
 def get_ethereum_chain():
     eth_chains = []
     for data in network_list:

tests/test_xcm_delivery_fee.py

@@ -0,0 +1,27 @@
+import pytest
+from scripts.utils.chain_model import Chain
+from tests.data.setting_data import get_parachains_with_xcm, get_relaychains_with_xcm
+
+
+parachain_chain_names = [
+    f'Test for {task.name}'
+    for task in get_parachains_with_xcm()
+]
+
+relaychain_chain_names = [
+    f'Test for {task.name}'
+    for task in get_relaychains_with_xcm()
+]
+
+class TestXCMDeliveryFee:
+
+
+    @pytest.mark.parametrize('chain_model', get_parachains_with_xcm(), ids=parachain_chain_names, indirect=True)
+    def test_parachain_has_delivery_factor(self, chain_model: Chain):
+        data = chain_model.substrate.get_metadata_storage_function(module_name='XcmpQueue', storage_name='DeliveryFeeFactor')
+        assert data is None
+
+    @pytest.mark.parametrize('chain_model', get_relaychains_with_xcm(), ids=relaychain_chain_names, indirect=True)
+    def test_relaychain_has_delivery_factor(self, chain_model: Chain):
+        data = chain_model.substrate.get_metadata_storage_function(module_name='Dmp', storage_name='DeliveryFeeFactor')
+        assert data is None