V0.1.0

Dockerfile

@@ -5,20 +5,19 @@ LABEL org.opencontainers.image.source /github.com/TranslatorSRI/TestHarness
 
 WORKDIR /app
 
-# make sure all is writeable for the nru USER later on
-RUN chmod -R 777 .
-
 # set up requirements
 COPY requirements.txt .
 COPY requirements-runners.txt .
 RUN pip install -r requirements.txt
 RUN pip install -r requirements-runners.txt
 
-# switch to the non-root user (nru). defined in the base image
-USER nru
-
 # set up source
 COPY . .
 
-# set up entrypoint
-ENTRYPOINT ["./main.sh"]
+# make sure all is writeable for the nru USER later on
+RUN chmod -R 777 .
+
+RUN pip install .
+
+# switch to the non-root user (nru). defined in the base image
+USER nru

requirements-runners.txt

@@ -1,2 +1,3 @@
-ARS_Test_Runner==0.0.2
-ui-test-runner==0.0.1
+ARS_Test_Runner==0.0.8
+benchmarks-runner==0.1.0
+ui-test-runner==0.0.2

requirements-test.txt

@@ -1,2 +1,3 @@
 pytest==7.4.2
+pytest-asyncio==0.23.3
 pytest-mock==3.11.1

requirements.txt

@@ -1,2 +1,5 @@
 httpx==0.25.0
 pydantic==1.10.13
+reasoner_pydantic==4.1.6
+tqdm==4.66.1
+translator-testing-model==0.2.3

setup.py

@@ -1,12 +1,13 @@
 """Setup file for SRI Test Harness package."""
+
 from setuptools import setup
 
 with open("README.md", encoding="utf-8") as readme_file:
     readme = readme_file.read()
 
 setup(
     name="sri-test-harness",
-    version="0.0.1",
+    version="0.1.0",
     author="Max Wang",
     author_email="max@covar.com",
     url="https://github.com/TranslatorSRI/TestHarness",

test_harness/download.py

@@ -1,20 +1,23 @@
 """Download tests."""
+
 import glob
 import httpx
 import io
 import json
 import logging
 from pathlib import Path
 import tempfile
-from typing import List, Union
+from typing import List, Union, Dict
 import zipfile
 
-from .models import TestCase, TestSuite
+from translator_testing_model.datamodel.pydanticmodel import TestCase, TestSuite
 
 
 def download_tests(
-    suite: Union[str, List[str]], url: Path, logger: logging.Logger
-) -> List[TestCase]:
+    suite: Union[str, List[str]],
+    url: Path,
+    logger: logging.Logger,
+) -> Dict[str, TestCase]:
     """Download tests from specified location."""
     assert Path(url).suffix == ".zip"
     logger.info(f"Downloading tests from {url}...")
@@ -28,34 +31,57 @@ def download_tests(
             zip_ref.extractall(tmpdir)
 
         # Find all json files in the downloaded zip
-        tests_paths = glob.glob(f"{tmpdir}/**/*.json", recursive=True)
+        # tests_paths = glob.glob(f"{tmpdir}/**/*.json", recursive=True)
+
+        tests_paths = glob.glob(f"{tmpdir}/*/test_suites/test_suite_output.json")
+
+        with open(tests_paths[0]) as f:
+            test_suite = TestSuite.parse_obj(json.load(f))
 
-        all_tests = []
-        suites = suite if type(suite) == list else [suite]
-        test_case_ids = []
+        # all_tests = []
+        # suites = suite if type(suite) == list else [suite]
+        # test_case_ids = []
 
-        logger.info(f"Reading in {len(tests_paths)} tests...")
+        # logger.info(f"Reading in {len(test_suite.test_cases)} 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
+        # for test_case in test_suite.test_cases:
+        #     try:
+        # test_suite = TestSuite.parse_obj(test_json)
+        # if test_suite.id in suites:
+        # if test_json["test_case_type"] == "acceptance":
+        #     # if suite is selected, grab all its test cases
+        #     # test_case_ids.extend(test_suite.case_ids)
+        #     all_tests.append(test_json)
+        #     continue
+        #     if test_json.get("test_env"):
+        #         # only grab Test Cases and not Test Assets
+        #         all_tests.append(test_json)
+        # except Exception as e:
+        #     # not a Test Suite
+        #     pass
+        # try:
+        #     # test_case = TestCase.parse_obj(test_json)
+        #     if test_json["test_case_type"] == "quantitative":
+        #         all_tests.append(test_json)
+        #         continue
+        #     # all_tests.append(test_json)
+        # except Exception as e:
+        #     # not a Test Case
+        #     print(e)
+        #     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
+    # tests = list(filter(lambda x: x in test_case_ids, all_tests))
+    # tests = [
+    #     test
+    #     for test in all_tests
+    #     for asset in test.test_assets
+    #     if asset.output_id
+    # ]
+    # for test in tests:
+    #     test.test_case_type = "acceptance"
+    # tests = all_tests
+    # tests = list(filter((lambda x: x for x in all_tests for asset in x.test_assets if asset.output_id), all_tests))
+    logger.info(f"Passing along {len(test_suite.test_cases)} tests")
+    return test_suite.test_cases

test_harness/logging.py → test_harness/logger.py

@@ -100,3 +100,5 @@ def setup_logger():
     logging.getLogger("httpcore").setLevel(logging.WARNING)
     logging.getLogger("httpx").setLevel(logging.WARNING)
     logging.getLogger("urllib3").setLevel(logging.WARNING)
+    logging.getLogger("matplotlib").setLevel(logging.WARNING)
+    logging.getLogger("root").setLevel(logging.WARNING)

test_harness/main.py

@@ -1,12 +1,16 @@
 """Translator SRI Automated Test Harness."""
+
 from argparse import ArgumentParser
+import asyncio
 import json
 from urllib.parse import urlparse
 from uuid import uuid4
 
-from .run import run_tests
-from .download import download_tests
-from .logging import get_logger, setup_logger
+from test_harness.run import run_tests
+from test_harness.download import download_tests
+from test_harness.logger import get_logger, setup_logger
+from test_harness.reporter import Reporter
+from test_harness.slacker import Slacker
 
 setup_logger()
 
@@ -18,7 +22,7 @@ def url_type(arg):
     raise TypeError("Invalid URL")
 
 
-def main(args):
+async def main(args):
     """Main Test Harness entrypoint."""
     qid = str(uuid4())[:8]
     logger = get_logger(qid, args["log_level"])
@@ -32,18 +36,29 @@ def main(args):
             "Please run this command with `-h` to see the available options."
         )
 
-    report = run_tests(tests, logger)
+    if len(tests) < 1:
+        return logger.warning("No tests to run. Exiting.")
+
+    # Create test run in the Information Radiator
+    reporter = Reporter(
+        base_url=args.get("reporter_url"),
+        refresh_token=args.get("reporter_access_token"),
+        logger=logger,
+    )
+    await reporter.get_auth()
+    await reporter.create_test_run()
+    slacker = Slacker()
+    report = await run_tests(reporter, slacker, tests, logger)
 
-    if args["save_to_dashboard"]:
-        logger.info("Saving to Testing Dashboard...")
-        raise NotImplementedError()
+    logger.info("Finishing up test run...")
+    await reporter.finish_test_run()
 
     if args["json_output"]:
-        logger.info("Saving report as JSON...")
+        # logger.info("Saving report as JSON...")
         with open("test_report.json", "w") as f:
             json.dump(report, f)
 
-    logger.info("All testing has completed!")
+    return logger.info("All tests have completed!")
 
 
 def cli():
@@ -78,6 +93,18 @@ def cli():
         help="Path to a file of tests to be run. This would be the same output from downloading the tests via `download_tests()`",
     )
 
+    parser.add_argument(
+        "--reporter_url",
+        type=url_type,
+        help="URL of the Testing Dashboard",
+    )
+
+    parser.add_argument(
+        "--reporter_access_token",
+        type=str,
+        help="Access token for authentication with the Testing Dashboard",
+    )
+
     parser.add_argument(
         "--save_to_dashboard",
         action="store_true",
@@ -95,11 +122,11 @@ def cli():
         type=str,
         choices=["ERROR", "WARNING", "INFO", "DEBUG"],
         help="Level of the logs.",
-        default="WARNING",
+        default="DEBUG",
     )
 
     args = parser.parse_args()
-    main(vars(args))
+    asyncio.run(main(vars(args)))
 
 
 if __name__ == "__main__":