Implement download

.gitignore

@@ -2,3 +2,4 @@
 __pycache__
 *.egg-info
 logs/
+Translator-Tests/

test_harness/download.py

@@ -1,11 +1,61 @@
 """Download tests."""
+import glob
+import httpx
+import io
+import json
 import logging
 from pathlib import Path
-from typing import List
+import tempfile
+from typing import List, Union
+import zipfile
 
-from .models import TestCase
+from .models import TestCase, TestSuite
 
 
-def download_tests(url: Path) -> List[TestCase]:
+def download_tests(
+    suite: Union[str, List[str]], url: Path, logger: logging.Logger
+) -> List[TestCase]:
     """Download tests from specified location."""
-    raise NotImplementedError()
+    assert Path(url).suffix == ".zip"
+    logger.info(f"Downloading tests from {url}...")
+    # download file from internet
+    with httpx.Client(follow_redirects=True) as client:
+        tests_zip = client.get(url)
+        tests_zip.raise_for_status()
+        # we already checked if zip before download, so now unzip
+    with tempfile.TemporaryDirectory() as tmpdir:
+        with zipfile.ZipFile(io.BytesIO(tests_zip.read())) as zip_ref:
+            zip_ref.extractall(tmpdir)
+
+        # Find all json files in the downloaded zip
+        tests_paths = glob.glob(f"{tmpdir}/**/*.json", recursive=True)
+
+        all_tests = []
+        suites = suite if type(suite) == list else [suite]
+        test_case_ids = []
+
+        logger.info(f"Reading in {len(tests_paths)} tests...")
+
+        # do the reading of the tests and make a tests list
+        for test_path in tests_paths:
+            with open(test_path, "r") as f:
+                test_json = json.load(f)
+                try:
+                    test_suite = TestSuite.parse_obj(test_json)
+                    if test_suite.id in suites:
+                        # if suite is selected, grab all its test cases
+                        test_case_ids.extend(test_suite.case_ids)
+                except Exception as e:
+                    # not a Test Suite
+                    pass
+                try:
+                    test_case = TestCase.parse_obj(test_json)
+                    all_tests.append(test_case)
+                except Exception as e:
+                    # not a Test Case
+                    pass
+
+    # only return the tests from the specified suites
+    tests = list(filter(lambda x: x in test_case_ids, all_tests))
+    logger.info(f"Passing along {len(tests)} tests")
+    return tests

test_harness/main.py

@@ -1,7 +1,7 @@
 """Translator SRI Automated Test Harness."""
 from argparse import ArgumentParser
 import json
-from pathlib import Path
+from urllib.parse import urlparse
 from uuid import uuid4
 
 from .run import run_tests
@@ -11,23 +11,28 @@
 setup_logger()
 
 
+def url_type(arg):
+    url = urlparse(arg)
+    if all((url.scheme, url.netloc)):
+        return arg
+    raise TypeError("Invalid URL")
+
+
 def main(args):
     """Main Test Harness entrypoint."""
     qid = str(uuid4())[:8]
     logger = get_logger(qid, args["log_level"])
     tests = []
     if "tests_url" in args:
-        logger.info("Downloading tests...")
-        tests = download_tests(args["tests_url"])
+        tests = download_tests(args["suite"], args["tests_url"], logger)
     elif "tests" in args:
         tests = args["tests"]
     else:
         return logger.error(
             "Please run this command with `-h` to see the available options."
         )
 
-    logger.info("Running tests...")
-    report = run_tests(tests)
+    report = run_tests(tests, logger)
 
     if args["save_to_dashboard"]:
         logger.info("Saving to Testing Dashboard...")
@@ -53,7 +58,16 @@ def cli():
     )
 
     download_parser.add_argument(
-        "tests_url", type=Path, help="URL to download in order to find the test files"
+        "suite",
+        type=str,
+        help="The name/id of the suite(s) to run. Once tests have been downloaded, the test cases in this suite(s) will be run.",
+    )
+
+    download_parser.add_argument(
+        "--tests_url",
+        type=url_type,
+        default="https://github.com/NCATSTranslator/Tests/archive/refs/heads/main.zip",
+        help="URL to download in order to find the test files",
     )
 
     run_parser = subparsers.add_parser("run", help="Run a given set of tests")

test_harness/models.py

@@ -1,5 +1,6 @@
 from enum import Enum
 from pydantic import BaseModel
+from typing import List, Optional
 
 
 class Type(str, Enum):
@@ -37,6 +38,7 @@ class TestCase(BaseModel):
     output_curie: str
     """
 
+    id: Optional[int]
     type: Type
     env: Env
     query_type: QueryType
@@ -45,6 +47,36 @@ class TestCase(BaseModel):
     output_curie: str
 
 
+class Tests(BaseModel):
+    """List of Test Cases."""
+
+    __root__: List[TestCase]
+
+    def __len__(self):
+        return len(self.__root__)
+
+    def __iter__(self):
+        return self.__root__.__iter__()
+
+    def __contains__(self, v):
+        return self.__root__.__contains__(v)
+
+    def __getitem__(self, i):
+        return self.__root__.__getitem__(i)
+
+
+class TestSuite(BaseModel):
+    """
+    Test Suite containing the ids of Test Cases.
+
+    id: int
+    case_ids: List[int]
+    """
+
+    id: int
+    case_ids: List[int]
+
+
 class TestResult(BaseModel):
     """Output of a Test."""
 

test_harness/run.py

@@ -1,22 +1,21 @@
 """Run tests through the Test Runners."""
 import logging
-from pydantic import validate_arguments
-from typing import List, Dict
+from tqdm import tqdm
+from typing import Dict
 
 from ui_test_runner import run_ui_test
 from ARS_Test_Runner.semantic_test import run_semantic_test as run_ars_test
 
-from .models import TestCase
+from .models import Tests
 
-logger = logging.getLogger(__name__)
 
-
-@validate_arguments
-def run_tests(tests: List[TestCase]) -> Dict:
+def run_tests(tests: Tests, logger: logging.Logger) -> Dict:
     """Send tests through the Test Runners."""
+    tests = Tests.parse_obj(tests)
+    logger.info(f"Running {len(tests)} tests...")
     full_report = {}
     # loop over all tests
-    for test in tests:
+    for test in tqdm(tests):
         # check if acceptance test
         if test.type == "acceptance":
             full_report[test.input_curie] = {}