Add script for updating test data

Internal-tag: [#64556]
Signed-off-by: bbrzyski <[email protected]>

docs/source/developers_guide/tests.md

@@ -72,3 +72,24 @@ python -m pytest --cov=topwrap --cov-report html
 ```
 
 Generated report is available at `htmlcov/index.html`
+
+## Updating kpm test data
+
+All kpm data from examples can be generated using nox.
+This is useful when changing topwrap functionality related to kpm in order to avoid manually changing every example's test data.
+You can either update only one part of examples data like specification or update everything (dataflows, specifications, designs).
+
+To update everything run:
+```bash
+nox -s update_test_data
+```
+
+To update only specifications run:
+```bash
+nox -s update_test_data -- specification
+```
+
+Possible options for `update_test_data` session:
+* `specification` - updates specifications
+* `dataflow` - updates dataflows
+* `design` - updates designs

noxfile.py

@@ -51,6 +51,16 @@ def tests(session: nox.Session) -> None:
     session.run("pytest", "-rs", "--cov-report", "html:cov_html", "--cov=topwrap", "tests")
 
 
+@nox.session
+def update_test_data(session: nox.Session) -> None:
+    session.install("-e", ".[tests]")
+    tests_to_update = (
+        ["dataflow", "specification", "design"] if not session.posargs else session.posargs
+    )
+    tests_to_update = [f"--{test_data}" for test_data in tests_to_update]
+    session.run("python3", "tests/update_test_data.py", *tests_to_update)
+
+
 def prepare_pyenv(session: nox.Session) -> dict:
     env = os.environ.copy()
     path = env.get("PATH")

tests/tests_kpm/common.py

@@ -2,12 +2,16 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import json
+from pathlib import Path
+
+import yaml
 
 from topwrap.design import DesignDescription
 
 AXI_NAME = "axi_bridge"
 PS7_NAME = "ps7"
 PWM_NAME = "litex_pwm_top"
+TEST_DATA_PATH = "tests/data/data_kpm/examples/"
 
 
 def read_json_file(json_file_path: str) -> dict:
@@ -18,3 +22,13 @@ def read_json_file(json_file_path: str) -> dict:
 
 def read_yaml_file(yaml_file_path: str) -> DesignDescription:
     return DesignDescription.load(yaml_file_path)
+
+
+def save_file_to_json(file_path: Path, file_name: str, file_content: dict):
+    with open(Path(file_path / file_name), "w") as json_file:
+        json.dump(file_content, json_file)
+
+
+def save_file_to_yaml(file_path: Path, file_name: str, file_content: dict):
+    with open(Path(file_path / file_name), "w") as yaml_file:
+        yaml_file.write(yaml.safe_dump(file_content, sort_keys=True))

tests/tests_kpm/conftest.py

@@ -8,16 +8,19 @@
 
 from topwrap.design import DesignDescription
 
-from .common import read_json_file, read_yaml_file
+from .common import TEST_DATA_PATH, read_json_file, read_yaml_file
 
 
-@pytest.fixture
-def ip_names() -> List[str]:
+def ip_names_data():
     return ["pwm", "hdmi", "hierarchy"]
 
 
 @pytest.fixture
-def pwm_ipcores_yamls() -> List[str]:
+def ip_names() -> List[str]:
+    return ip_names_data()
+
+
+def pwm_ipcores_yamls_data() -> List[str]:
     _pwm_yamls_prefix = "examples/pwm/ipcores/"
     return [
         "topwrap/ips/axi/axi_axil_adapter.yaml",
@@ -27,7 +30,11 @@ def pwm_ipcores_yamls() -> List[str]:
 
 
 @pytest.fixture
-def hdmi_ipcores_yamls() -> List[str]:
+def pwm_ipcores_yamls() -> List[str]:
+    return pwm_ipcores_yamls_data()
+
+
+def hdmi_ipcores_yamls_data() -> List[str]:
     _hdmi_yamls_prefix = "examples/hdmi/ipcores/"
     _axi_yamls_prefix = "topwrap/ips/axi/"
     return [
@@ -47,7 +54,11 @@ def hdmi_ipcores_yamls() -> List[str]:
 
 
 @pytest.fixture
-def hierarchy_ipcores_yamls() -> List[str]:
+def hdmi_ipcores_yamls() -> List[str]:
+    return hdmi_ipcores_yamls_data()
+
+
+def hierarchy_ipcores_yamls_data() -> List[str]:
     _hierarchy_yamls_prefix = "examples/hierarchy/repo/cores/"
     return [
         _hierarchy_yamls_prefix + "c_mod_1/c_mod_1.yaml",
@@ -62,90 +73,122 @@ def hierarchy_ipcores_yamls() -> List[str]:
 
 
 @pytest.fixture
-def all_yaml_files(
-    pwm_ipcores_yamls: List[str], hdmi_ipcores_yamls: List[str], hierarchy_ipcores_yamls: List[str]
-) -> Dict[str, List[str]]:
+def hierarchy_ipcores_yamls() -> List[str]:
+    return hierarchy_ipcores_yamls_data()
+
+
+def all_yaml_files_data() -> Dict[str, List[str]]:
     return {
-        "pwm": pwm_ipcores_yamls,
-        "hdmi": hdmi_ipcores_yamls,
-        "hierarchy": hierarchy_ipcores_yamls,
+        "pwm": pwm_ipcores_yamls_data(),
+        "hdmi": hdmi_ipcores_yamls_data(),
+        "hierarchy": hierarchy_ipcores_yamls_data(),
     }
 
 
 @pytest.fixture
-def all_specification_files(ip_names: List[str]) -> Dict[str, dict]:
+def all_yaml_files() -> Dict[str, List[str]]:
+    return all_yaml_files_data()
+
+
+def all_specification_files_data() -> Dict[str, dict]:
     return {
-        ip_name: read_json_file(
-            f"tests/data/data_kpm/examples/{ip_name}/specification_{ip_name}.json"
-        )
-        for ip_name in ip_names
+        ip_name: read_json_file(f"{TEST_DATA_PATH}{ip_name}/specification_{ip_name}.json")
+        for ip_name in ip_names_data()
     }
 
 
 @pytest.fixture
-def all_dataflow_files(ip_names: List[str]) -> Dict[str, dict]:
+def all_specification_files() -> Dict[str, dict]:
+    return all_specification_files_data()
+
+
+def all_dataflow_files_data() -> Dict[str, dict]:
     return {
-        ip_name: read_json_file(f"tests/data/data_kpm/examples/{ip_name}/{ip_name}_dataflow.json")
-        for ip_name in ip_names
+        ip_name: read_json_file(f"{TEST_DATA_PATH}{ip_name}/dataflow_{ip_name}.json")
+        for ip_name in ip_names_data()
     }
 
 
 @pytest.fixture
-def all_designs(ip_names: List[str]) -> Dict[str, DesignDescription]:
+def all_dataflow_files() -> Dict[str, dict]:
+    return all_dataflow_files_data()
+
+
+def all_designs_data() -> Dict[str, DesignDescription]:
     return {
-        ip_name: read_yaml_file(f"tests/data/data_kpm/examples/{ip_name}/test_project.yml")
-        for ip_name in ip_names
+        ip_name: read_yaml_file(f"{TEST_DATA_PATH}{ip_name}/project_{ip_name}.yml")
+        for ip_name in ip_names_data()
     }
 
 
 @pytest.fixture
-def all_encoded_design_files(all_designs: Dict[str, DesignDescription]) -> Dict[str, str]:
+def all_designs() -> Dict[str, DesignDescription]:
+    return all_designs_data()
+
+
+def all_examples_designs_data() -> Dict[str, DesignDescription]:
+    return {
+        ip_name: read_yaml_file(f"examples/{ip_name}/project.yml") for ip_name in ip_names_data()
+    }
+
+
+@pytest.fixture
+def all_examples_designs() -> Dict[str, DesignDescription]:
+    return all_examples_designs_data()
+
+
+def all_encoded_design_files_data() -> Dict[str, str]:
     return {
         test_name: b64encode(design.to_yaml().encode("utf-8")).decode("utf-8")
-        for test_name, design in all_designs.items()
+        for test_name, design in all_designs_data().items()
     }
 
 
 @pytest.fixture
-def pwm_specification(all_specification_files: Dict[str, dict]) -> dict:
-    return all_specification_files["pwm"]
+def all_encoded_design_files() -> Dict[str, str]:
+    return all_encoded_design_files_data()
+
+
+@pytest.fixture
+def pwm_specification() -> dict:
+    return all_specification_files_data()["pwm"]
 
 
 @pytest.fixture
-def hdmi_specification(all_specification_files: Dict[str, dict]) -> dict:
-    return all_specification_files["hdmi"]
+def hdmi_specification() -> dict:
+    return all_specification_files_data()["hdmi"]
 
 
 @pytest.fixture
-def hierarchy_specification(all_specification_files: Dict[str, dict]) -> dict:
-    return all_specification_files["hierarchy"]
+def hierarchy_specification() -> dict:
+    return all_specification_files_data()["hierarchy"]
 
 
 @pytest.fixture
-def pwm_dataflow(all_dataflow_files: Dict[str, dict]) -> dict:
-    return all_dataflow_files["pwm"]
+def pwm_dataflow() -> dict:
+    return all_dataflow_files_data()["pwm"]
 
 
 @pytest.fixture
-def hdmi_dataflow(all_dataflow_files: Dict[str, dict]) -> dict:
-    return all_dataflow_files["hdmi"]
+def hdmi_dataflow() -> dict:
+    return all_dataflow_files_data()["hdmi"]
 
 
 @pytest.fixture
-def hierarchy_dataflow(all_dataflow_files: Dict[str, dict]) -> dict:
-    return all_dataflow_files["hierarchy"]
+def hierarchy_dataflow() -> dict:
+    return all_dataflow_files_data()["hierarchy"]
 
 
 @pytest.fixture
-def pwm_design(all_designs: Dict[str, DesignDescription]) -> DesignDescription:
-    return all_designs["pwm"]
+def pwm_design() -> DesignDescription:
+    return all_designs_data()["pwm"]
 
 
 @pytest.fixture
-def hdmi_design(all_designs: Dict[str, DesignDescription]) -> DesignDescription:
-    return all_designs["hdmi"]
+def hdmi_design() -> DesignDescription:
+    return all_designs_data()["hdmi"]
 
 
 @pytest.fixture
-def hierarchy_design(all_designs: Dict[str, DesignDescription]) -> DesignDescription:
-    return all_designs["hierarchy"]
+def hierarchy_design() -> DesignDescription:
+    return all_designs_data()["hierarchy"]

tests/tests_kpm/test_kpm_client.py

@@ -2,7 +2,9 @@
 
 import pytest
 from deepdiff import DeepDiff
+from deepdiff.serialization import yaml
 from pipeline_manager_backend_communication.misc_structures import MessageType
+from pipeline_manager_backend_communication.utils import convert_message_to_string
 
 from tests.tests_build.test_design import Path
 from topwrap.kpm_common import RPCparams
@@ -58,25 +60,32 @@ def test_dataflow_run(self, all_yaml_files, all_dataflow_files, default_rpc_para
             }, f"Dataflow run returned {response_message} for test {test_name}"
 
     def test_dataflow_export(
-        self, all_yaml_files, all_dataflow_files, all_encoded_design_files, default_rpc_params
+        self, all_yaml_files, all_dataflow_files, all_designs, default_rpc_params
     ):
-        for (test_name, ip_core_yamls), dataflow_json, encoded_design in zip(
-            all_yaml_files.items(), all_dataflow_files.values(), all_encoded_design_files.values()
+        for (test_name, ip_core_yamls), dataflow_json, design in zip(
+            all_yaml_files.items(), all_dataflow_files.values(), all_designs.values()
         ):
             default_rpc_params.yamlfiles = ip_core_yamls
-
             response_message = RPCMethods(default_rpc_params).dataflow_export(dataflow_json)
+
+            response_design_dict = yaml.safe_load(
+                convert_message_to_string(response_message["content"], True, "application/x-yaml")
+            )
+            # Dump design and load back with yaml to avoid errors with different types like tuple vs list
+            design_loaded_yaml = yaml.safe_load(design.to_yaml())
+
+            design_diff = DeepDiff(design_loaded_yaml, response_design_dict, ignore_order=True)
             assert (
-                response_message["content"] == encoded_design
+                design_diff == {}
             ), f"Dataflow export returned different encoded file for test {test_name}"
 
     def test_dataflow_import(self, all_yaml_files, all_encoded_design_files, default_rpc_params):
-        for (test_name, ip_core_yamls), encoded_dataflow in zip(
+        for (test_name, ip_core_yamls), encoded_design in zip(
             all_yaml_files.items(), all_encoded_design_files.values()
         ):
             default_rpc_params.yamlfiles = ip_core_yamls
             response_message = RPCMethods(default_rpc_params).dataflow_import(
-                encoded_dataflow, "application/x-yaml", True
+                encoded_design, "application/x-yaml", True
             )
 
             # Checks if imported dataflow is correct are done in "test_kpm_import.py"