update test regression

tests/conftest.py

@@ -2,6 +2,7 @@
 import pathlib
 import json
 import pytest
+import re
 
 try:
     os.environ['CMR_USER']
@@ -54,7 +55,7 @@ def pytest_generate_tests(metafunc):
 
         association_dir = 'uat' if metafunc.config.option.env == 'uat' else 'ops'
         associations = os.listdir(cmr_dirpath.joinpath(association_dir))
-        
+
         if 'collection_concept_id' in metafunc.fixturenames and associations is not None:
             metafunc.parametrize("collection_concept_id", associations)
     else:
@@ -69,27 +70,101 @@ def log_global_env_facts(record_testsuite_property, request):
     record_testsuite_property("env", request.config.getoption('env'))
 
 
+def get_error_message(report):
+
+    # If it's a regular test failure (not a skipped or xfailed test)
+    if hasattr(report, 'longreprtext'):
+        # Extract the short-form failure reason (in pytest >= 6)
+        error_message = report.longreprtext
+    else:
+        # Fallback if longreprtext is not available
+        if isinstance(report.longrepr, tuple):
+            error_message = report.longrepr[2]
+        else:
+            error_message = str(report.longrepr)
+
+    exception_pattern = r"E\s+(\w+):\s+\(([^,]+),\s+'(.+?)'\)"
+    match = re.search(exception_pattern, error_message)
+
+    if match:
+        exception_type = match.group(1)  # 'Exception'
+        exception_reason = match.group(2)  # 'Not Found'
+        exception_message = match.group(3)  # 'Error: EULA ... could not be found.'
+
+        # Combine all into one message
+        full_message = f"Exception Type: {exception_type}, Reason: {exception_reason}, Message: {exception_message}"
+        return full_message
+    else:
+        return "No exception found."
+
+
 def pytest_terminal_summary(terminalreporter, exitstatus, config):
 
     filtered_success, success, skipped, failed = [], [], [], []
+
     test_results = {'success': filtered_success, 'failed': failed, 'skipped': skipped}
 
-    # the fourth keyword is the collection concept id may change if we change the test inputs
-    skipped.extend([list(skip.keywords)[3] for skip in terminalreporter.stats.get('skipped', [])])
-    failed.extend([list(failed.keywords)[3] for failed in terminalreporter.stats.get('failed', [])])
-    success.extend([list(passed.keywords)[3] for passed in terminalreporter.stats.get('passed', [])])
+    failed_tests = terminalreporter.stats.get('failed', [])
+    skipped_tests = terminalreporter.stats.get('skipped', [])
+    success_tests = terminalreporter.stats.get('passed', [])
+
+    if failed_tests:
+        for report in failed_tests:
+
+            concept_id = list(report.keywords)[3]
 
-    # Have temporal and spatial test if failed either one don't put in success
+            # Extract the test name and exception message from the report
+            test_name = report.nodeid
+            test_type = None
 
-    # Convert lists to sets
-    fail_set = set(failed)
-    success_set = set(success)
+            if "spatial" in test_name:
+                test_type = "spatial"
+            elif "temporal" in test_name:
+                test_type = "temporal"
 
-    # Remove elements from success that are in fail
-    set_filtered_success = success_set - fail_set
+            full_message = get_error_message(report)
 
-    # Convert the set back to a list if needed
-    filtered_success.extend(list(set_filtered_success))
+            failed.append({
+                "concept_id": concept_id,
+                "test_type": test_type,
+                "message": full_message
+            })
+
+    if skipped_tests:
+        for report in skipped_tests:
+
+            concept_id = list(report.keywords)[3]
+
+            # Extract the test name and exception message from the report
+            test_name = report.nodeid
+            test_type = None
+
+            if "spatial" in test_name:
+                test_type = "spatial"
+            elif "temporal" in test_name:
+                test_type = "temporal"
+
+            # If it's a regular test failure (not a skipped or xfailed test)
+            if hasattr(report, 'longreprtext'):
+                # Extract the short-form failure reason (in pytest >= 6)
+                error_message = report.longreprtext
+            else:
+                # Fallback if longreprtext is not available
+                if isinstance(report.longrepr, tuple):
+                    error_message = report.longrepr[2]
+                else:
+                    error_message = str(report.longrepr)
+
+            error = "UNKNOWN"
+            if isinstance(report.longreprtext, str):
+                tuple_error = eval(report.longreprtext)
+                error = tuple_error[2]
+
+            skipped.append({
+                "concept_id": concept_id,
+                "test_type": test_type,
+                "message": error
+            })
 
     env = config.option.env
 
@@ -108,5 +183,3 @@ def pytest_terminal_summary(terminalreporter, exitstatus, config):
                     json.dump(tests, file)
 
 
-
-

tests/create_or_update_issue.py

@@ -168,7 +168,6 @@ def update_issue(repo_name, issue_number, issue_body, github_token):
 
     print(f"Issue updated successfully: {response.json()['html_url']}")
 
-
 def create_or_update_issue(repo_name, github_token, env):
 
     upper_env = env.upper()
@@ -185,41 +184,46 @@ def create_or_update_issue(repo_name, github_token, env):
 
     failed = results.get('failed', [])
     skipped = results.get('skipped',[])
-
+
+    failed_concept_ids = [collection.get('concept_id') for collection in failed]
+    skipped_concept_ids =  [collection.get('concept_id') for collection in skipped]
+
     no_associations = []
     failed_test = []
 
-    for collection in failed: 
-        if collection not in current_associations:
-            no_associations.append(collection)
+    for collection_concept_id in failed_concept_ids: 
+        if collection_concept_id not in current_associations:
+            no_associations.append(collection_concept_id)
         else:
-            failed_test.append(collection)
+            failed_test.append(collection_concept_id)
 
     providers = []
     issue_body = None
 
-    all_collections = failed + skipped
+    all_collections = failed_concept_ids + skipped_concept_ids
 
     if len(failed) > 0 or len(skipped) > 0:
 
-        for collection in failed:
+        for collection in failed_concept_ids:
             provider = collection.split('-')[1]
             if provider not in providers:
                 providers.append(provider)
-        for collection in skipped:
+        for collection in skipped_concept_ids:
             provider = collection.split('-')[1]
             if provider not in providers:
                 providers.append(provider)
 
         collection_names = get_collection_names(providers, env, all_collections)
         issue_body = datetime.now().strftime("Updated on %m-%d-%Y\n")
 
+        print(collection_names)
+
         if len(failed_test) > 0:
             issue_body += "\n FAILED: \n"
-            issue_body += "\n".join(f"{cid} ({collection_names.get(cid, '')})" for cid in failed_test)
+            issue_body += "\n".join(f"{cid.get('concept_id')} ({collection_names.get(cid.get('concept_id'), '')}) - {cid.get('test_type')} test -  {cid.get('message')}" for cid in failed)
         if len(skipped) > 0:
             issue_body += "\n SKIPPED: \n"
-            issue_body += "\n".join(f"{cid} ({collection_names.get(cid, '')})" for cid in skipped)
+            issue_body += "\n".join(f"{cid.get('concept_id')} ({collection_names.get(cid.get('concept_id'), '')}) - {cid.get('test_type')} test -  {cid.get('message')}" for cid in skipped)
         if len(no_associations) > 0:
             issue_body += "\n NO ASSOCIATIONS: \n"
             issue_body += "\n".join(f"{cid} ({collection_names.get(cid, '')})" for cid in no_associations)

tests/verify_collection.py

@@ -130,7 +130,7 @@ def granule_json(collection_concept_id: str, cmr_mode: str, bearer_token: str, r
     if 'items' in response_json and len(response_json['items']) > 0:
         return response_json['items'][0]
     elif cmr_mode == cmr.CMR_UAT:
-        pytest.skip(f"No granules found for UAT collection {collection_concept_id}. CMR search used was {cmr_url}")
+        pytest.fail(f"No granules found for UAT collection {collection_concept_id}. CMR search used was {cmr_url}")
     elif cmr_mode == cmr.CMR_OPS:
         pytest.fail(f"No granules found for OPS collection {collection_concept_id}. CMR search used was {cmr_url}")
 
@@ -155,7 +155,7 @@ def download_file(url):
     if granule_url:
         return download_file(granule_url)
     else:
-        pytest.skip(f"Unable to find download URL for {granule_json['meta']['concept-id']}")
+        pytest.fail(f"Unable to find download URL for {granule_json['meta']['concept-id']}")
 
 
 @pytest.fixture(scope="function")
@@ -167,8 +167,8 @@ def collection_variables(cmr_mode, collection_concept_id, env, bearer_token):
     collection_associations = collection_res.get("associations")
     variable_concept_ids = collection_associations.get("variables")
 
-    if variable_concept_ids is None and env == 'uat':
-        pytest.skip('There are no umm-v associated with this collection in UAT')
+    if variable_concept_ids is None:
+        pytest.fail(f'There are no umm-v associated with this collection in {env}')
 
     variables = []
     for i in range(0, len(variable_concept_ids), 40):