Merge pull request #325 from microbiomedata/317-watcher-get-401-unaut…

…horized-error-from-jobs-endpoint

Handle error responses from API token endpoint

.github/workflows/blt.yml

@@ -45,5 +45,5 @@ jobs:
 
     - name: Test with pytest
       run: |
-        poetry run pytest ./tests --junit-xml=pytest.xml --cov-report=term  \
+        poetry run pytest -m "not integration" ./tests --junit-xml=pytest.xml --cov-report=term  \
         --cov-report=xml --cov=nmdc_automation --local-badge-output-dir badges/

nmdc_automation/api/nmdcapi.py

@@ -14,7 +14,12 @@
 from datetime import datetime, timedelta, timezone
 from nmdc_automation.config import SiteConfig, UserConfig
 import logging
+from tenacity import retry, wait_exponential, stop_after_attempt
 
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+SECONDS_IN_DAY = 86400
 
 def _get_sha256(fn: Union[str, Path]) -> str:
     """
@@ -45,7 +50,7 @@ def expiry_dt_from_now(days=0, hours=0, minutes=0, seconds=0):
 
 class NmdcRuntimeApi:
     token = None
-    expires = 0
+    expires_at = 0
     _base_url = None
     client_id = None
     client_secret = None
@@ -63,15 +68,21 @@ def __init__(self, site_configuration: Union[str, Path, SiteConfig]):
     def refresh_token(func):
         def _get_token(self, *args, **kwargs):
             # If it expires in 60 seconds, refresh
-            if not self.token or self.expires + 60 > time():
+            if not self.token or self.expires_at + 60 > time():
                 self.get_token()
             return func(self, *args, **kwargs)
 
         return _get_token
 
+    @retry(
+        wait=wait_exponential(multiplier=4, min=8, max=120),
+        stop=stop_after_attempt(6),
+        reraise=True,
+    )
     def get_token(self):
         """
         Get a token using a client id/secret.
+        Retries up to 6 times with exponential backoff.
         """
         h = {
             "accept": "application/json",
@@ -83,17 +94,34 @@ def get_token(self):
             "client_secret": self.client_secret,
         }
         url = self._base_url + "token"
-        resp = requests.post(url, headers=h, data=data).json()
-        expt = resp["expires"]
-        self.expires = time() + expt["minutes"] * 60
 
-        self.token = resp["access_token"]
+        resp = requests.post(url, headers=h, data=data)
+        if not resp.ok:
+            logging.error(f"Failed to get token: {resp.text}")
+            resp.raise_for_status()
+        response_body = resp.json()
+
+        # Expires can be in days, hours, minutes, seconds - sum them up and convert to seconds
+        expires = 0
+        if "days" in response_body["expires"]:
+            expires += int(response_body["expires"]["days"]) * SECONDS_IN_DAY
+        if "hours" in response_body["expires"]:
+            expires += int(response_body["expires"]["hours"]) * 3600
+        if "minutes" in response_body["expires"]:
+            expires += int(response_body["expires"]["minutes"]) * 60
+        if "seconds" in response_body["expires"]:
+            expires += int(response_body["expires"]["seconds"])
+
+        self.expires_at = time() + expires
+
+        self.token = response_body["access_token"]
         self.header = {
             "accept": "application/json",
             "Content-Type": "application/json",
             "Authorization": "Bearer %s" % (self.token),
         }
-        return resp
+        logging.info(f"New token expires at {self.expires_at}")
+        return response_body
 
     def get_header(self):
         return self.header

nmdc_automation/workflow_automation/watch_nmdc.py

@@ -300,7 +300,7 @@ def get_unclaimed_jobs(self, allowed_workflows) -> List[WorkflowJob]:
             "workflow.id": {"$in": allowed_workflows},
             "claims": {"$size": 0}
         }
-        job_records =  self.runtime_api.list_jobs(filt=filt)
+        job_records = self.runtime_api.list_jobs(filt=filt)
 
         for job in job_records:
             jobs.append(WorkflowJob(self.config, workflow_state=job))

pyproject.toml

@@ -30,6 +30,7 @@ pytest-local-badge = "^1.0.3"
 pysam = "^0.22.1"
 importlib = "^1.0.4"
 tomli = "^2.0.2"
+tenacity = "^9.0.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.3.1"
@@ -43,3 +44,8 @@ style = "pep440"
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
+
+[tool.pytest.ini_options]
+markers = [
+    "integration: mark test as integration test",
+]

pytest.xml

@@ -1 +1,9 @@
-<?xml version="1.0" encoding="utf-8"?><testsuites><testsuite name="pytest" errors="0" failures="0" skipped="0" tests="95" time="15.756" timestamp="2024-11-26T12:32:34.300291" hostname="MBThornton-M92.local"><testcase classname="tests.test_config" name="test_config" time="0.046" /><testcase classname="tests.test_config" name="test_config_missing" time="0.001" /><testcase classname="tests.test_imports" name="test_gold_mapper_map_sequencing_data" time="0.061" /><testcase classname="tests.test_imports" name="test_gold_mapper_map_data_unique" time="0.061" /><testcase classname="tests.test_imports" name="test_gold_mapper_map_data_multiple" time="0.059" /><testcase classname="tests.test_imports" name="test_gold_mapper_map_workflow_executions" time="5.532" /><testcase classname="tests.test_models" name="test_workflow_process_factory" time="0.005" /><testcase classname="tests.test_models" name="test_workflow_process_factory_incorrect_id" time="1.213" /><testcase classname="tests.test_models" name="test_workflow_process_factory_data_generation_invalid_analyte_category" time="0.001" /><testcase classname="tests.test_models" name="test_workflow_process_factory_metagenome_assembly_with_invalid_execution_resource" time="0.001" /><testcase classname="tests.test_models" name="test_workflow_process_factory_mags_with_mags_list" time="0.001" /><testcase classname="tests.test_models" name="test_process_factory_with_db_record" time="0.001" /><testcase classname="tests.test_models" name="test_workflow_process_node[mags_analysis_record.json-nmdc:MagsAnalysis]" time="0.008" /><testcase classname="tests.test_models" name="test_workflow_process_node[metagenome_annotation_record.json-nmdc:MetagenomeAnnotation]" time="0.007" /><testcase classname="tests.test_models" name="test_workflow_process_node[metagenome_assembly_record.json-nmdc:MetagenomeAssembly]" time="0.007" /><testcase classname="tests.test_models" name="test_workflow_process_node[metatranscriptome_annotation_record.json-nmdc:MetatranscriptomeAnnotation]" time="0.007" /><testcase classname="tests.test_models" name="test_workflow_process_node[metatranscriptome_assembly_record.json-nmdc:MetatranscriptomeAssembly]" time="0.007" /><testcase classname="tests.test_models" name="test_workflow_process_node[metatranscriptome_expression_analysis_record.json-nmdc:MetatranscriptomeExpressionAnalysis]" time="0.007" /><testcase classname="tests.test_models" name="test_workflow_process_node[nucleotide_sequencing_record.json-nmdc:NucleotideSequencing]" time="0.071" /><testcase classname="tests.test_models" name="test_workflow_process_node[read_based_taxonomy_analysis_record.json-nmdc:ReadBasedTaxonomyAnalysis]" time="0.007" /><testcase classname="tests.test_models" name="test_workflow_process_node[read_qc_analysis_record.json-nmdc:ReadQcAnalysis]" time="0.007" /><testcase classname="tests.test_models" name="test_data_object_creation_from_records" time="0.145" /><testcase classname="tests.test_models" name="test_data_object_creation_from_db_records" time="0.165" /><testcase classname="tests.test_models" name="test_data_object_creation_invalid_data_object_type" time="0.001" /><testcase classname="tests.test_models" name="test_data_object_creation_invalid_data_category" time="0.001" /><testcase classname="tests.test_models" name="test_job_output_creation" time="0.001" /><testcase classname="tests.test_models" name="test_job_creation" time="0.001" /><testcase classname="tests.test_nmdcapi" name="test_basics" time="0.013" /><testcase classname="tests.test_nmdcapi" name="test_objects" time="0.012" /><testcase classname="tests.test_nmdcapi" name="test_list_funcs" time="0.006" /><testcase classname="tests.test_nmdcapi" name="test_update_op" time="0.004" /><testcase classname="tests.test_nmdcapi" name="test_jobs" time="0.006" /><testcase classname="tests.test_sched" name="test_scheduler_cycle[workflows.yaml]" time="0.042" /><testcase classname="tests.test_sched" name="test_scheduler_cycle[workflows-mt.yaml]" time="0.047" /><testcase classname="tests.test_sched" name="test_progress[workflows.yaml]" time="0.321" /><testcase classname="tests.test_sched" name="test_progress[workflows-mt.yaml]" time="0.193" /><testcase classname="tests.test_sched" name="test_multiple_versions" time="0.183" /><testcase classname="tests.test_sched" name="test_out_of_range" time="0.083" /><testcase classname="tests.test_sched" name="test_type_resolving" time="0.107" /><testcase classname="tests.test_sched" name="test_scheduler_add_job_rec[workflows.yaml]" time="0.011" /><testcase classname="tests.test_sched" name="test_scheduler_add_job_rec[workflows-mt.yaml]" time="0.009" /><testcase classname="tests.test_sched" name="test_scheduler_find_new_jobs" time="0.073" /><testcase classname="tests.test_watch_nmdc" name="test_file_handler_init_from_state_file" time="0.006" /><testcase classname="tests.test_watch_nmdc" name="test_file_handler_init_from_config_agent_state" time="0.002" /><testcase classname="tests.test_watch_nmdc" name="test_file_handler_init_default_state" time="0.002" /><testcase classname="tests.test_watch_nmdc" name="test_file_handler_read_state" time="0.001" /><testcase classname="tests.test_watch_nmdc" name="test_file_handler_write_state" time="0.005" /><testcase classname="tests.test_watch_nmdc" name="test_file_handler_get_output_path" time="0.001" /><testcase classname="tests.test_watch_nmdc" name="test_file_handler_write_metadata_if_not_exists" time="0.002" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_init" time="0.001" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_restore_from_state" time="0.001" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_job_checkpoint" time="0.001" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_save_checkpoint" time="0.002" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_find_job_by_opid" time="0.002" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_prepare_and_cache_new_job" time="0.001" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_prepare_and_cache_new_job_force" time="0.002" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_get_finished_jobs" time="0.002" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_process_successful_job" time="0.327" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_get_finished_jobs_1_failure" time="0.003" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_process_failed_job_1_failure" time="0.004" /><testcase classname="tests.test_watch_nmdc" name="test_job_manager_process_failed_job_2_failures" time="0.003" /><testcase classname="tests.test_watch_nmdc" name="test_claim_jobs" time="2.004" /><testcase classname="tests.test_watch_nmdc" name="test_runtime_manager_get_unclaimed_jobs" time="0.047" /><testcase classname="tests.test_watch_nmdc" name="test_reclaim_job" time="0.001" /><testcase classname="tests.test_watch_nmdc" name="test_watcher_restore_from_checkpoint" time="0.001" /><testcase classname="tests.test_wfutils" name="test_workflow_job" time="0.001" /><testcase classname="tests.test_wfutils" name="test_cromwell_job_runner" time="0.001" /><testcase classname="tests.test_wfutils" name="test_cromwell_job_runner_get_job_status" time="0.003" /><testcase classname="tests.test_wfutils" name="test_cromwell_job_runner_get_job_metadata" time="0.002" /><testcase classname="tests.test_wfutils" name="test_workflow_job_as_workflow_execution_dict" time="0.001" /><testcase classname="tests.test_wfutils" name="test_workflow_state_manager" time="0.001" /><testcase classname="tests.test_wfutils" name="test_workflow_manager_fetch_release_file_success" time="0.002" /><testcase classname="tests.test_wfutils" name="test_workflow_manager_fetch_release_file_failed_download" time="0.001" /><testcase classname="tests.test_wfutils" name="test_workflow_manager_fetch_release_file_failed_write" time="0.002" /><testcase classname="tests.test_wfutils" name="test_cromwell_runner_setup_inputs_and_labels" time="0.001" /><testcase classname="tests.test_wfutils" name="test_cromwell_runner_generate_submission_files" time="0.003" /><testcase classname="tests.test_wfutils" name="test_cromwell_runner_generate_submission_files_exception" time="0.003" /><testcase classname="tests.test_wfutils" name="test_cromwell_job_runner_submit_job_new_job" time="0.002" /><testcase classname="tests.test_wfutils" name="test_workflow_job_data_objects_and_execution_record_mags" time="0.004" /><testcase classname="tests.test_wfutils" name="test_workflow_job_from_database_job_record" time="0.001" /><testcase classname="tests.test_workflow_process" name="test_load_workflow_process_nodes[workflows.yaml]" time="0.025" /><testcase classname="tests.test_workflow_process" name="test_load_workflow_process_nodes[workflows-mt.yaml]" time="0.026" /><testcase classname="tests.test_workflow_process" name="test_load_workflow_process_nodes_with_obsolete_versions" time="0.017" /><testcase classname="tests.test_workflow_process" name="test_resolve_relationships" time="0.019" /><testcase classname="tests.test_workflow_process" name="test_load_workflow_process_nodes_does_not_load_metagenome_sequencing" time="0.014" /><testcase classname="tests.test_workflow_process" name="test_load_workflows[workflows.yaml]" time="0.004" /><testcase classname="tests.test_workflow_process" name="test_load_workflows[workflows-mt.yaml]" time="0.003" /><testcase classname="tests.test_workflow_process" name="test_get_required_data_objects_by_id[workflows.yaml]" time="0.012" /><testcase classname="tests.test_workflow_process" name="test_get_required_data_objects_by_id[workflows-mt.yaml]" time="0.013" /><testcase classname="tests.test_workflow_process" name="test_within_range" time="0.001" /><testcase classname="tests.test_jgi_file_staging.test_file_metadata" name="test_get_access_token" time="0.001" /><testcase classname="tests.test_jgi_file_staging.test_file_metadata" name="test_check_access_token" time="1.009" /><testcase classname="tests.test_jgi_file_staging.test_file_metadata" name="test_check_access_token_invalid" time="1.009" /><testcase classname="tests.test_jgi_file_staging.test_file_metadata" name="test_get_sequence_id" time="2.017" /><testcase classname="tests.test_jgi_file_staging.test_file_metadata" name="test_get_analysis_projects_from_proposal_id" time="0.034" /></testsuite></testsuites>
+<?xml version="1.0" encoding="utf-8"?><testsuites><testsuite name="pytest" errors="0" failures="1" skipped="0" tests="1" time="0.404" timestamp="2024-12-16T16:04:17.583510" hostname="MBThornton-M92.local"><testcase classname="tests.test_nmdcapi_integration" name="test_nmdcapi_basics" time="0.002"><failure message="AssertionError: Unimplemented&#10;assert False">requests_mock = &lt;requests_mock.mocker.Mocker object at 0x10c76b650&gt;, site_config_file = PosixPath('/Users/MBThornton/Documents/code/nmdc_automation/tests/site_configuration_test.toml')
+
+    @pytest.mark.integration
+    def test_nmdcapi_basics(requests_mock, site_config_file):
+&gt;       assert False, "Unimplemented"
+E       AssertionError: Unimplemented
+E       assert False
+
+tests/test_nmdcapi_integration.py:10: AssertionError</failure></testcase></testsuite></testsuites>

tests/conftest.py

@@ -67,32 +67,34 @@ def test_db():
     conn_str = os.environ.get("MONGO_URL", "mongodb://localhost:27017")
     return MongoClient(conn_str).test
 
-@fixture(autouse=True)
+@fixture(scope="function")
 def mock_api(monkeypatch, requests_mock, test_data_dir):
     monkeypatch.setenv("NMDC_API_URL", "http://localhost")
     monkeypatch.setenv("NMDC_CLIENT_ID", "anid")
     monkeypatch.setenv("NMDC_CLIENT_SECRET", "asecret")
     token_resp = {"expires": {"minutes": time()+60},
             "access_token": "abcd"
             }
-    requests_mock.post("http://localhost/token", json=token_resp)
+    requests_mock.post("http://localhost:8000/token", json=token_resp)
+
+
     resp = ["nmdc:dobj-01-abcd4321"]
     # mock mint responses in sequence
 
-    requests_mock.post("http://localhost/pids/mint", json=resp)
+    requests_mock.post("http://localhost:8000/pids/mint", json=resp)
     requests_mock.post(
-        "http://localhost/workflows/workflow_executions",
+        "http://localhost:8000/workflows/workflow_executions",
         json=resp
         )
-    requests_mock.post("http://localhost/pids/bind", json=resp)
+    requests_mock.post("http://localhost:8000/pids/bind", json=resp)
 
     rqcf = test_data_dir / "rqc_response2.json"
     rqc = json.load(open(rqcf))
     rqc_resp = {"resources": [rqc]}
-    requests_mock.get("http://localhost/jobs", json=rqc_resp)
+    requests_mock.get("http://localhost:8000/jobs", json=rqc_resp)
 
-    requests_mock.patch("http://localhost/operations/nmdc:1234", json={})
-    requests_mock.get("http://localhost/operations/nmdc:1234", json={'metadata': {}})
+    requests_mock.patch("http://localhost:8000/operations/nmdc:1234", json={})
+    requests_mock.get("http://localhost:8000/operations/nmdc:1234", json={'metadata': {}})
 
 
 @fixture(scope="session")

tests/site_configuration_test.toml

@@ -14,7 +14,7 @@ site = "Processing Site"
 
 [nmdc]
 url_root = "https://data.microbiomedata.org/data/"
-api_url = "http://localhost"
+api_url = "http://localhost:8000"
 
 [state]
 watch_state = "State File"