add report fetch mocked unit tests

lib/NarrativeService/reportfetcher.py

@@ -8,6 +8,11 @@ def __init__(self, ws_client: Workspace) -> None:
         self.ws_client = ws_client
 
     def find_report_from_object(self, upa: str) -> dict[str, list|str]:
+        """
+        Given the UPA of an object, this attempts to find any reports that it references.
+        If the object doesn't have any referencing reports, but it was a copy of another
+        object, this tries to find the copy's reports.
+        """
         # TODO: make sure upa's real.
 
         # first, fetch object references (without data)
@@ -16,7 +21,9 @@ def find_report_from_object(self, upa: str) -> dict[str, list|str]:
         # if we find at least one, return them
         # if we find 0, test if it's a copy, and search upstream.
         report_upas = [
-            ServiceUtils.object_info_to_object(ref_info)["ref"] for ref_info in ref_list if REPORT_TYPE in ref_info[2]
+            ServiceUtils.object_info_to_object(ref_info)["ref"]
+            for ref_info in ref_list
+            if len(ref_info) and REPORT_TYPE in ref_info[2]
         ]
         if len(report_upas):
             return self.build_output(upa, report_upas)
@@ -35,7 +42,23 @@ def find_report_from_copy_source(self, upa: str) -> dict[str, list|str]:
             return self.find_report_from_object(obj_data["copied"])
         return self.build_output(upa, [])
 
-    def build_output(self, upa: str, report_upas: list[str] | None=None, inaccessible: int=0, error: str | None=None) -> dict[str, list|str]:
+    def build_output(
+        self,
+        upa: str,
+        report_upas: list[str] | None=None,
+        inaccessible: int=0,
+        error: str | None=None
+    ) -> dict[str, list|str|int]:
+        """
+        Builds the output dictionary for the report fetching.
+        Definitely has keys:
+        report_upas - list[str] - list of report object UPAs, if they exist
+        object_upa - str - the UPA of the object referencing the reports.
+        Might have keys:
+        inaccessible - int (1) if the given object in object_upa was copied, and its source
+            (which might reference reports) is inaccessible
+        error - str - if an error occurred.
+        """
         if report_upas is None:
             report_upas = []
         ret_val = {

requirements-dev.txt

@@ -1,5 +1,6 @@
 coverage==7.5.3
 pytest-cov==5.0.0
 pytest==8.2.1
+pytest-mock==3.14.0
 requests-mock==1.12.1
 ruff==0.4.6

test/unit/test_reportfetcher.py

@@ -1,4 +1,67 @@
-from NarrativeService.reportfetcher import ReportFetcher
 import pytest
+from NarrativeService.reportfetcher import ReportFetcher
+
 from lib.installed_clients.WorkspaceClient import Workspace
 
+REPORT_OBJ_INFO = [
+    2,
+    "some_report",
+    "KBaseReport.Report-1.0",
+    "2024-05-30T15:22:20+0000",
+    1,
+    "some-user",
+    123,
+    "some_workspace",
+    123,
+    123,
+    None
+]
+
+def test_fetch_report_ok(mocker):
+    mock_ws = mocker.MagicMock()
+    mock_ws.list_referencing_objects.return_value = [[REPORT_OBJ_INFO]]
+    rf = ReportFetcher(mock_ws)
+    result = rf.find_report_from_object("123/1/1")
+    assert result == {
+        "report_upas": ["123/2/1"],
+        "object_upa": "123/1/1"
+    }
+
+def test_fetch_report_from_copy(mocker):
+    def list_ref_objects_effect(ref_list):
+        if ref_list[0]["ref"] == "123/5/1":
+            return [[REPORT_OBJ_INFO]]
+        return [[[]]]
+
+    mock_ws = mocker.MagicMock()
+    mock_ws.list_referencing_objects.side_effect = list_ref_objects_effect
+    mock_ws.get_objects2.return_value = { "data": [{"copied": "123/5/1"}] }
+
+    rf = ReportFetcher(mock_ws)
+    result = rf.find_report_from_object("123/1/1")
+    assert result == {
+        "report_upas": ["123/2/1"],
+        "object_upa": "123/5/1"
+    }
+
+def test_fetch_report_none(mocker):
+    mock_ws = mocker.MagicMock()
+    mock_ws.list_referencing_objects.return_value = [[[]]]
+    mock_ws.get_objects2.return_value = { "data": [{}] }
+    rf = ReportFetcher(mock_ws)
+    assert rf.find_report_from_object("123/1/1") == {
+        "report_upas": [],
+        "object_upa": "123/1/1"
+    }
+
+def test_fetch_report_copy_inaccessible(mocker):
+    mock_ws = mocker.MagicMock()
+    mock_ws.list_referencing_objects.return_value = [[[]]]
+    mock_ws.get_objects2.return_value = { "data": [{"copy_source_inaccessible": 1}] }
+    rf = ReportFetcher(mock_ws)
+    assert rf.find_report_from_object("123/1/1") == {
+        "report_upas": [],
+        "object_upa": "123/1/1",
+        "inaccessible": 1,
+        "error": "No report found. This object is a copy, and its source is inaccessible."
+    }