ci: add coverage check

.github/workflows/ci.yaml

@@ -13,7 +13,7 @@ concurrency:
 jobs:
   unittest:
     name: unit tests
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-latest
     strategy:
       matrix:
         # don't go crazy with the Python versions as they eat up CI minutes
@@ -23,7 +23,7 @@ jobs:
       - uses: actions/checkout@v4
 
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python-version }}
 
@@ -34,10 +34,20 @@ jobs:
 
       - name: Test with pytest
         run: |
-          python -m pytest
+          python -m pytest --cov=chart_review --cov-report=xml
+
+      - name: Check coverage report
+        if: github.ref != 'refs/heads/main'
+        uses: orgoro/[email protected]
+        with:
+          coverageFile: coverage.xml
+          token: ${{ secrets.GITHUB_TOKEN }}
+          thresholdAll: .99
+          thresholdNew: 1
+          thresholdModified: 1
 
   lint:
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
 

chart_review/agree.py

@@ -62,30 +62,11 @@ def confusion_matrix(
             elif not truth_positive and not annotator_positive:
                 TN.append(key)
             else:
-                raise Exception("Guard: Impossible comparison of reviewers")
+                raise Exception("Guard: Impossible comparison of reviewers")  # pragma: no cover
 
     return {"TP": TP, "FN": FN, "FP": FP, "TN": TN}
 
 
-def append_matrix(first: dict, second: dict) -> dict:
-    """
-    Append two different confusion_matrix matrix dictionaries.
-
-    For example, (Annotator1 VS NLP) appended to (Annotator2 vs NLP).
-
-    TODO: Warning: assumes first and second have no overlapping NoteRange,
-          may not be applicable for other studies.
-
-    :param first: confusion_matrix matrix
-    :param second: confusion_matrix matrix
-    :return:
-    """
-    added = {}
-    for header in ["TP", "FP", "FN", "TN"]:
-        added[header] = first[header] + second[header]
-    return added
-
-
 def score_kappa(matrix: dict) -> float:
     """
     Computes Cohen kappa for pair-wise annotators.

chart_review/cli.py

@@ -30,8 +30,9 @@ def main_cli(argv: list[str] = None) -> None:
         args = parser.parse_args(argv)
         args.func(args)
     except Exception as exc:
-        sys.exit(str(exc))
+        print(str(exc), file=sys.stderr)
+        sys.exit(1)
 
 
 if __name__ == "__main__":
-    main_cli()
+    main_cli()  # pragma: no cover

chart_review/cohort.py

@@ -1,6 +1,5 @@
 from typing import Iterable
 
-from chart_review.common import guard_iter, guard_in
 from chart_review import agree, common, config, errors, external, simplify, types
 
 
@@ -86,13 +85,12 @@ def class_labels(self):
 
     def _select_labels(self, label_pick: str = None) -> Iterable[str]:
         if label_pick:
-            guard_in(label_pick, self.class_labels)
             return [label_pick]
         else:
             return self.class_labels
 
     def confusion_matrix(
-        self, truth: str, annotator: str, note_range: Iterable, label_pick: str = None
+        self, truth: str, annotator: str, note_range: types.NoteSet, label_pick: str = None
     ) -> dict:
         """
         This is the rollup of counting each symptom only once, not multiple times.
@@ -104,7 +102,6 @@ def confusion_matrix(
         :return: dict
         """
         labels = self._select_labels(label_pick)
-        note_range = set(guard_iter(note_range))
         return agree.confusion_matrix(
             self.annotations,
             truth,

chart_review/commands/accuracy.py

@@ -32,11 +32,9 @@ def print_accuracy(args: argparse.Namespace) -> None:
     annotator = args.annotator
 
     if truth not in reader.note_range:
-        print(f"Unrecognized annotator '{truth}'")
-        return
+        raise ValueError(f"Unrecognized annotator '{truth}'")
     if annotator not in reader.note_range:
-        print(f"Unrecognized annotator '{annotator}'")
-        return
+        raise ValueError(f"Unrecognized annotator '{annotator}'")
 
     # Grab the intersection of ranges
     note_range = set(reader.note_range[truth])

chart_review/common.py

@@ -58,65 +58,3 @@ def write_text(path: str, text: str) -> None:
 
     with open(path, "w") as f:
         f.write(text)
-
-
-def csv_map(csv_file: str) -> dict:
-    """
-    Get a simple map of csv file contents. For more complex handling, use pandas.
-    :param csv_file: path to CSV file with rows "key,value"
-    :return: dict {key:value}
-    """
-    res = {}
-    with open(csv_file, "r") as fp:
-        for line in fp.readlines():
-            line = line.replace('"', "").strip()
-            key, val = line.split(",")
-            res[key] = val
-    return res
-
-
-###############################################################################
-# Helper Functions: Pretty Print JSON, useful for Testing/Debugging
-###############################################################################
-def print_json(jsonable) -> None:
-    """
-    :param jsonable: dict or list of serializable object
-    """
-    if isinstance(jsonable, dict):
-        print(json.dumps(jsonable, indent=4))
-    elif isinstance(jsonable, list):
-        print(json.dumps(jsonable, indent=4))
-    else:
-        print(jsonable)
-
-
-def print_line(heading=None) -> None:
-    """
-    :param heading: optionally give a header to the line seperator
-    """
-    seperator = "##############################################################"
-    if heading:
-        print(f"\n{seperator}\n{heading}\n{seperator}")
-    else:
-        print(seperator)
-
-
-###############################################################################
-# Helper Functions: enum type smoothing
-###############################################################################
-def guard_iter(object) -> Iterable:
-    if isinstance(object, Enum):
-        return guard_iter(object.value)
-    elif isinstance(object, EnumMeta):
-        return guard_iter(object.value)
-    elif isinstance(object, Iterable):
-        return object
-    else:
-        raise Exception(f"expected Iterable|Enum but got {type(object)}")
-
-
-def guard_in(entry, strict: Iterable):
-    if entry in strict:
-        return entry
-    else:
-        raise Exception(f"expected entry {entry} to be in {strict}")

pyproject.toml

@@ -39,6 +39,7 @@ line-length = 100
 tests = [
     "ddt",
     "pytest",
+    "pytest-cov",
 ]
 dev = [
     "bandit[toml]",

tests/base.py

@@ -30,3 +30,10 @@ def capture_stdout() -> io.StringIO:
         stdout = io.StringIO()
         with contextlib.redirect_stdout(stdout):
             yield stdout
+
+    @staticmethod
+    @contextlib.contextmanager
+    def capture_stderr() -> io.StringIO:
+        stderr = io.StringIO()
+        with contextlib.redirect_stderr(stderr):
+            yield stderr

tests/data/external/enc.csv

@@ -1,4 +1,4 @@
-BLARG_ID,SYMPTOM
+ENC_ID,SYMPTOM
 "ABC-Enc","happy"
 "ABC","ignored"
 "ABC-Enc","tired"

tests/data/ignore/config.yaml

@@ -6,3 +6,4 @@ ignore:
   - 3-Enc
   - Encounter/Anon-4-Enc
   - 5
+  - Encounter/Unknown  # just test we don't blow up

tests/test_accuracy.py

@@ -164,3 +164,17 @@ def test_verbose_cvs(self):
             ],
             stdout.splitlines(),
         )
+
+    def test_bad_truth(self):
+        """Verify that we do something suitable for bad arguments"""
+        # Truth
+        with self.capture_stderr() as stderr:
+            with self.assertRaises(SystemExit):
+                self.run_cli("accuracy", "nope1", "nope2", path=f"{self.DATA_DIR}/cold")
+        self.assertEqual("Unrecognized annotator 'nope1'\n", stderr.getvalue())
+
+        # Annotator
+        with self.capture_stderr() as stderr:
+            with self.assertRaises(SystemExit):
+                self.run_cli("accuracy", "jill", "nope2", path=f"{self.DATA_DIR}/cold")
+        self.assertEqual("Unrecognized annotator 'nope2'\n", stderr.getvalue())

tests/test_cohort.py

@@ -67,3 +67,15 @@ def test_non_existent_ids(self):
             reader = cohort.CohortReader(config.ProjectConfig(tmpdir))
 
             self.assertEqual({"bob": {1, 3}}, reader.note_range)
+
+    def test_unknown_annotators_are_ignored(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            common.write_json(f"{tmpdir}/config.json", {"annotators": {"bob": 1}})
+            common.write_json(
+                f"{tmpdir}/labelstudio-export.json",
+                [
+                    {"id": 1, "annotations": [{"completed_by": 1}, {"completed_by": 2}]},
+                ],
+            )
+            reader = cohort.CohortReader(config.ProjectConfig(tmpdir))
+            self.assertEqual({"bob": {1}}, reader.note_range)