Pathfinder test runner

requirements-runners.txt

@@ -1,6 +1,4 @@
 ARS_Test_Runner==0.2.4
 # benchmarks-runner==0.1.3
 # ui-test-runner==0.0.2
-graph-validation-test-runners==0.1.5
-# we need to manually pin the right reasoner-validator version
-reasoner-validator==4.2.5
+# graph-validation-test-runners==0.1.5

requirements.txt

@@ -4,4 +4,5 @@ pydantic==2.7.1
 setproctitle==1.3.3
 slack_sdk==3.27.2
 tqdm==4.66.4
-translator-testing-model==0.3.2
+translator-testing-model==0.4.1
+reasoner-validator==4.2.5

test_harness/download.py

@@ -10,14 +10,18 @@
 from typing import List, Union, Dict
 import zipfile
 
-from translator_testing_model.datamodel.pydanticmodel import TestCase, TestSuite
+from translator_testing_model.datamodel.pydanticmodel import (
+    TestCase,
+    PathfinderTestCase,
+    TestSuite,
+)
 
 
 def download_tests(
     suite: Union[str, List[str]],
     url: Path,
     logger: logging.Logger,
-) -> Dict[str, TestCase]:
+) -> Dict[str, Union[TestCase, PathfinderTestCase]]:
     """Download tests from specified location."""
     assert Path(url).suffix == ".zip"
     logger.info(f"Downloading tests from {url}...")

test_harness/pathfinder_test_runner.py

@@ -0,0 +1,44 @@
+from typing import Dict, Union, List
+
+
+async def pathfinder_pass_fail_analysis(
+    report: Dict[str, any],
+    agent: str,
+    message: Dict[str, any],
+    path_nodes: List[List[str]],
+    minimum_required_path_nodes: int,
+) -> Dict[str, any]:
+    found_path_nodes = set()
+    unmatched_paths = set()
+    for analysis in message["results"][0]["analyses"]:
+        for path_bindings in analysis["path_bindings"].values():
+            for path_binding in path_bindings:
+                path_id = path_binding["id"]
+                matching_path_nodes = set()
+                for edge_id in message["auxiliary_graphs"][path_id]["edges"]:
+                    edge = message["knowledge_graph"]["edges"][edge_id]
+                    for node_curies in path_nodes:
+                        unhit_node = True
+                        for curie in node_curies:
+                            if curie in matching_path_nodes:
+                                unhit_node = False
+                        if unhit_node:
+                            if edge["subject"] in node_curies:
+                                matching_path_nodes.add(edge["subject"])
+                            if edge["object"] in node_curies:
+                                matching_path_nodes.add(edge["object"])
+                if len(matching_path_nodes) >= minimum_required_path_nodes:
+                    found_path_nodes.add(",".join(matching_path_nodes))
+                elif len(matching_path_nodes) > 0:
+                    unmatched_paths.add(",".join(matching_path_nodes))
+
+    if len(found_path_nodes) > 0:
+        report[agent]["status"] = "PASSED"
+        report[agent]["expected_nodes_found"] = "; ".join(found_path_nodes)
+    elif len(unmatched_paths) > 0:
+        report[agent]["status"] = "FAILED"
+        report[agent]["expected_nodes_found"] = "; ".join(unmatched_paths)
+    else:
+        report[agent]["status"] = "FAILED"
+
+    return report

test_harness/reporter.py

@@ -4,9 +4,14 @@
 import httpx
 import logging
 import os
-from typing import List
+from typing import List, Union
 
-from translator_testing_model.datamodel.pydanticmodel import TestCase, TestAsset
+from translator_testing_model.datamodel.pydanticmodel import (
+    TestCase,
+    PathfinderTestCase,
+    TestAsset,
+    PathfinderTestAsset,
+)
 
 
 class Reporter:
@@ -63,25 +68,36 @@ async def create_test_run(self, test_env, suite_name):
         self.test_run_id = res_json["id"]
         return self.test_run_id
 
-    async def create_test(self, test: TestCase, asset: TestAsset):
+    async def create_test(
+        self,
+        test: Union[TestCase, PathfinderTestCase],
+        asset: Union[TestAsset, PathfinderTestAsset],
+    ):
         """Create a test in the IR."""
         name = asset.name if asset.name else asset.description
-        res = await self.authenticated_client.post(
-            url=f"{self.base_path}/api/reporting/v1/test-runs/{self.test_run_id}/tests",
-            json={
-                "name": name,
-                "className": test.name,
-                "methodName": asset.name,
-                "startedAt": datetime.now().astimezone().isoformat(),
-                "labels": [
-                    {
-                        "key": "TestCase",
-                        "value": test.id,
-                    },
-                    {
-                        "key": "TestAsset",
-                        "value": asset.id,
-                    },
+        test_json = {
+            "name": name,
+            "className": test.name,
+            "methodName": asset.name,
+            "startedAt": datetime.now().astimezone().isoformat(),
+            "labels": [
+                {
+                    "key": "TestCase",
+                    "value": test.id,
+                },
+                {
+                    "key": "TestAsset",
+                    "value": asset.id,
+                },
+                {
+                    "key": "ExpectedOutput",
+                    "value": asset.expected_output,
+                },
+            ],
+        }
+        if isinstance(test, TestCase) and isinstance(asset, TestAsset):
+            test_json["labels"].extend(
+                [
                     {
                         "key": "InputCurie",
                         "value": asset.input_id,
@@ -90,12 +106,28 @@ async def create_test(self, test: TestCase, asset: TestAsset):
                         "key": "OutputCurie",
                         "value": asset.output_id,
                     },
+                ]
+            )
+        elif isinstance(test, PathfinderTestCase) and isinstance(
+            asset, PathfinderTestAsset
+        ):
+            test_json["labels"].extend(
+                [
+                    {
+                        "key": "SourceInputCurie",
+                        "value": asset.source_input_id,
+                    },
                     {
-                        "key": "ExpectedOutput",
-                        "value": asset.expected_output,
+                        "key": "TargetInputCurie",
+                        "value": asset.target_input_id,
                     },
-                ],
-            },
+                ]
+            )
+        else:
+            raise Exception
+        res = await self.authenticated_client.post(
+            url=f"{self.base_path}/api/reporting/v1/test-runs/{self.test_run_id}/tests",
+            json=test_json,
         )
         res.raise_for_status()
         res_json = res.json()

test_harness/result_collector.py

@@ -2,7 +2,12 @@
 
 import logging
 from typing import Union
-from translator_testing_model.datamodel.pydanticmodel import TestAsset, TestCase
+from translator_testing_model.datamodel.pydanticmodel import (
+    TestAsset,
+    PathfinderTestAsset,
+    TestCase,
+    PathfinderTestCase,
+)
 
 from test_harness.utils import get_tag
 
@@ -43,8 +48,8 @@ def __init__(self, logger: logging.Logger):
 
     def collect_result(
         self,
-        test: TestCase,
-        asset: TestAsset,
+        test: Union[TestCase, PathfinderTestCase],
+        asset: Union[TestAsset, PathfinderTestAsset],
         report: dict,
         parent_pk: Union[str, None],
         url: str,

test_harness/run.py

@@ -5,26 +5,31 @@
 import time
 from tqdm import tqdm
 import traceback
-from typing import Dict
+from typing import Dict, Union
 
 from ARS_Test_Runner.semantic_test import pass_fail_analysis
-from standards_validation_test_runner import StandardsValidationTest
+
+# from standards_validation_test_runner import StandardsValidationTest
 
 # from benchmarks_runner import run_benchmarks
 
-from translator_testing_model.datamodel.pydanticmodel import TestCase
+from translator_testing_model.datamodel.pydanticmodel import (
+    TestCase,
+    PathfinderTestCase,
+)
 
 from test_harness.runner.query_runner import QueryRunner
 from test_harness.reporter import Reporter
 from test_harness.slacker import Slacker
 from test_harness.result_collector import ResultCollector
 from test_harness.utils import get_tag, hash_test_asset
+from test_harness.pathfinder_test_runner import pathfinder_pass_fail_analysis
 
 
 async def run_tests(
     reporter: Reporter,
     slacker: Slacker,
-    tests: Dict[str, TestCase],
+    tests: Dict[str, Union[TestCase, PathfinderTestCase]],
     logger: logging.Logger = logging.getLogger(__name__),
     args: Dict[str, any] = {},
 ) -> Dict:
@@ -71,7 +76,7 @@ async def run_tests(
                 try:
                     test_id = await reporter.create_test(test, asset)
                     test_ids.append(test_id)
-                except Exception:
+                except Exception as e:
                     logger.error(f"Failed to create test: {test.id}")
                     continue
 
@@ -91,9 +96,24 @@ async def run_tests(
                         "pks": test_query["pks"],
                         "result": {},
                     }
+                    if isinstance(test, PathfinderTestCase):
+                        report["test_details"] = {
+                            "minimum_required_path_nodes": asset.minimum_required_path_nodes,
+                            "expected_path_nodes": "; ".join(
+                                [
+                                    ",".join(
+                                        [
+                                            normalized_curies[path_node_id]
+                                            for path_node_id in path_node.ids
+                                        ]
+                                    )
+                                    for path_node in asset.path_nodes
+                                ]
+                            ),
+                        }
                     for agent, response in test_query["responses"].items():
                         report["result"][agent] = {
-                            "trapi_validation": "NA",
+                            # "trapi_validation": "NA",
                         }
                         agent_report = report["result"][agent]
                         try:
@@ -117,28 +137,28 @@ async def run_tests(
                             logger.warning(
                                 f"Failed to parse basic response fields from {agent}: {e}"
                             )
-                        try:
-                            svt = StandardsValidationTest(
-                                test_asset=asset,
-                                environment=test.test_env,
-                                component=agent,
-                                trapi_version=args["trapi_version"],
-                                biolink_version="suppress",
-                                runner_settings="Inferred",
-                            )
-                            results = svt.test_case_processor(
-                                trapi_response=response["response"]
-                            )
-                            agent_report["trapi_validation"] = results[
-                                next(iter(results.keys()))
-                            ][agent]["status"]
-                            if agent_report["trapi_validation"] == "FAILED":
-                                agent_report["status"] = "FAILED"
-                                agent_report["message"] = "TRAPI Validation Error"
-                                continue
-                        except Exception as e:
-                            logger.warning(f"Failed to run TRAPI validation with {e}")
-                            agent_report["trapi_validation"] = "ERROR"
+                        # try:
+                        #     svt = StandardsValidationTest(
+                        #         test_asset=asset,
+                        #         environment=test.test_env,
+                        #         component=agent,
+                        #         trapi_version=args["trapi_version"],
+                        #         biolink_version="suppress",
+                        #         runner_settings="Inferred",
+                        #     )
+                        #     results = svt.test_case_processor(
+                        #         trapi_response=response["response"]
+                        #     )
+                        #     agent_report["trapi_validation"] = results[
+                        #         next(iter(results.keys()))
+                        #     ][agent]["status"]
+                        #     if agent_report["trapi_validation"] == "FAILED":
+                        #         agent_report["status"] = "FAILED"
+                        #         agent_report["message"] = "TRAPI Validation Error"
+                        #         continue
+                        # except Exception as e:
+                        #     logger.warning(f"Failed to run TRAPI validation with {e}")
+                        #     agent_report["trapi_validation"] = "ERROR"
                         try:
                             if (
                                 response["response"]["message"].get("results") is None
@@ -147,13 +167,28 @@ async def run_tests(
                                 agent_report["status"] = "DONE"
                                 agent_report["message"] = "No results"
                                 continue
-                            await pass_fail_analysis(
-                                report["result"],
-                                agent,
-                                response["response"]["message"]["results"],
-                                normalized_curies[asset.output_id],
-                                asset.expected_output,
-                            )
+                            if isinstance(test, PathfinderTestCase):
+                                await pathfinder_pass_fail_analysis(
+                                    report["result"],
+                                    agent,
+                                    response["response"]["message"],
+                                    [
+                                        [
+                                            normalized_curies[path_node_id]
+                                            for path_node_id in path_node.ids
+                                        ]
+                                        for path_node in asset.path_nodes
+                                    ],
+                                    asset.minimum_required_path_nodes,
+                                )
+                            else:
+                                await pass_fail_analysis(
+                                    report["result"],
+                                    agent,
+                                    response["response"]["message"]["results"],
+                                    normalized_curies[asset.output_id],
+                                    asset.expected_output,
+                                )
                         except Exception as e:
                             logger.error(
                                 f"Failed to run acceptance test analysis on {agent}: {e}"