Run black

test_harness/result_collector.py

@@ -1,4 +1,5 @@
 """The Collector of Results."""
+
 from translator_testing_model.datamodel.pydanticmodel import TestAsset, TestCase
 
 from .utils import get_tag
@@ -24,21 +25,27 @@ def __init__(self):
                 self.stats[agent][query_type] = {}
                 for result_type in self.result_types.values():
                     self.stats[agent][query_type][result_type] = 0
-        
+
         self.columns = ["name", "url", "pk", "TestCase", "TestAsset", *self.agents]
         header = ",".join(self.columns)
         self.csv = f"{header}\n"
-    
+
     def collect_result(self, test: TestCase, asset: TestAsset, result: dict, url: str):
         """Add a single result to the total output."""
         # add result to stats
         for agent in result["result"]:
             query_type = asset.expected_output
-            result_type = self.result_types.get(get_tag(result["result"][agent]), "Test Error")
+            result_type = self.result_types.get(
+                get_tag(result["result"][agent]), "Test Error"
+            )
             self.stats[agent][query_type][result_type] += 1
-        
+
         # add result to csv
-        agent_results = ",".join(get_tag(result["result"][agent]) for agent in self.agents)
+        agent_results = ",".join(
+            get_tag(result["result"][agent]) for agent in self.agents
+        )
         ars_pk = result["pks"].get("parent_pk", None)
         pk_url = f"https://arax.ncats.io/?r={ars_pk}" if ars_pk is not None else ""
-        self.csv += f"""{asset.name},{url},{pk_url},{test.id},{asset.id},{agent_results}\n"""
+        self.csv += (
+            f"""{asset.name},{url},{pk_url},{test.id},{asset.id},{agent_results}\n"""
+        )

test_harness/run.py

@@ -175,7 +175,12 @@ async def run_tests(
                             test_result["result"].get("ars", {}).get("status", "FAILED")
                         )
                     full_report[status] += 1
-                    collector.collect_result(test, asset, test_result, f"{reporter.base_path}/test-runs/{reporter.test_run_id}/tests/{test_id}")
+                    collector.collect_result(
+                        test,
+                        asset,
+                        test_result,
+                        f"{reporter.base_path}/test-runs/{reporter.test_run_id}/tests/{test_id}",
+                    )
                     if not err_msg and status != "SKIPPED":
                         # only upload ara labels if the test ran successfully
                         try:

test_harness/slacker.py

@@ -1,4 +1,5 @@
 """Slack notification integration class."""
+
 import httpx
 import json
 import os
@@ -10,7 +11,9 @@ class Slacker:
     """Slack notification poster."""
 
     def __init__(self, url=None, token=None, slack_channel=None):
-        self.channel = slack_channel if slack_channel is not None else os.getenv("SLACK_CHANNEL")
+        self.channel = (
+            slack_channel if slack_channel is not None else os.getenv("SLACK_CHANNEL")
+        )
         self.url = url if url is not None else os.getenv("SLACK_WEBHOOK_URL")
         slack_token = token if token is not None else os.getenv("SLACK_TOKEN")
         self.client = WebClient(slack_token)
@@ -37,7 +40,7 @@ async def post_notification(self, messages=[]):
                     "blocks": blocks,
                 },
             )
-    
+
     async def upload_test_results_file(self, filename, extension, results):
         """Upload a results file to Slack."""
         with tempfile.TemporaryDirectory() as td:
@@ -53,4 +56,3 @@ async def upload_test_results_file(self, filename, extension, results):
                 file=tmp_path,
                 initial_comment="Test Results:",
             )
-