fix: rm test_filter

.cloud-build/execute_changed_notebooks_cli.py

@@ -137,7 +137,11 @@ changed_notebooks = execute_changed_notebooks_helper.get_changed_notebooks(
 
 
 results_bucket = f"{args.artifacts_bucket}"
-results_file = f"{args.build_id}.json"
+# artifacts_bucket may get set by trigger to a full gs:// folder path
+if results_bucket.startswith("gs://"):
+    results_bucket = results_bucket[5:]
+results_bucket = results_bucket.split('/')[0]
+results_file = f"build_results/{args.build_id}.json"
 
 if args.test_percent == 100:
     notebooks = changed_notebooks
@@ -158,7 +162,6 @@ else:
         staging_bucket=args.staging_bucket,
         artifacts_bucket=args.artifacts_bucket,
         results_file=results_file,
-        accumulative_results=accumulative_results,
         should_parallelize=args.should_parallelize,
         timeout=args.timeout,
         variable_project_id=args.variable_project_id,

.cloud-build/execute_changed_notebooks_helper.py

@@ -95,8 +95,20 @@ def load_results(results_bucket: str,
     print("Loading existing accumulative results ...")
     accumulative_results = {}
     try:
-        content = util.download_blob_into_memory(results_bucket, results_file, download_as_text=True)
-        accumulative_results = json.loads(content)
+        client = storage.Client()
+        bucket = client.bucket(results_bucket)
+
+        build_results_dir = os.path.dirname(results_file)
+        blobs = client.list_blobs(results_bucket, prefix=build_results_dir)
+        for blob in blobs:
+            content = util.download_blob_into_memory(results_bucket, blob.name, download_as_text=True)
+            build_results = json.loads(content)
+            for notebook in build_results:
+                if notebook in accumulative_results:
+                    accumulative_results[notebook] += build_results[notebook]
+                else:
+                    accumulative_results[notebook] = build_results[notebook]
+
         print(accumulative_results)
     except Exception as e:
         print(e)
@@ -118,7 +130,15 @@ def select_notebook(changed_notebook: str,
         pass_count = 1
         fail_count = 0
 
-    return (random.randint(1, 100) * (1 + (fail_count / (pass_count + fail_count))) < test_percent)
+    inferred_failure_rate = fail_count / (pass_count + fail_count)
+
+    # If failure rate is high, the chance of testing should be higher
+    should_test_due_to_failure = random.uniform(0, 1) < inferred_failure_rate
+
+    # Additionally, only test a percentage of these
+    should_test_due_to_random_subset = random.uniform(0, 1) < test_percent
+
+    return should_test_due_to_failure and should_test_due_to_random_subset
 
 
 def _process_notebook(
@@ -379,39 +399,30 @@ def get_changed_notebooks(
     return notebooks
 
 def _save_results(results: List[NotebookExecutionResult],
-                  accumulative_results: Dict[str,Any],
                   artifacts_bucket: str,
                   results_file: str):
 
     artifacts_bucket = artifacts_bucket.replace("gs://", "").split('/')[0]
 
-    print("Updating accumulative results ...")
+    print("Updating build results ...")
+    build_results = {}
     for result in results:
-        if result.path in accumulative_results:
-            accumulative_results[result.path]['duration'] = result.duration.total_seconds()
-            accumulative_results[result.path]['start_time'] = str(result.start_time)
-            if result.is_pass:
-                accumulative_results[result.path]['passed'] += 1
-            else:
-                accumulative_results[result.path]['failed'] += 1
-            print(f"updating {result.path}")
+        if result.is_pass:
+            pass_count = 1
+            fail_count = 0
         else:
-            if result.is_pass:
-                pass_count = 1
-                fail_count = 0
-            else:
-                pass_count = 0
-                fail_count = 1
-            accumulative_results[result.path] = {
-                    'duration': result.duration.total_seconds(),
-                    'start_time': str(result.start_time),
-                    'passed': pass_count,
-                    'failed': fail_count
-                    }
-            print(f"adding {result.path}")
+            pass_count = 0
+            fail_count = 1
+        build_results[result.path] = {
+                'duration': result.duration.total_seconds(),
+                'start_time': str(result.start_time),
+                'passed': pass_count,
+                'failed': fail_count
+        }
+        print(f"adding {result.path}")
 
     print("Saving accumulative results ...")
-    content = json.dumps(accumulative_results)
+    content = json.dumps(build_results)
 
     client = storage.Client()
     bucket = client.get_bucket(artifacts_bucket)
@@ -425,7 +436,6 @@ def process_and_execute_notebooks(
     staging_bucket: str,
     artifacts_bucket: str,
     results_file: str,
-    accumulative_results: List[NotebookExecutionResult],
     should_parallelize: bool,
     timeout: int,
     variable_project_id: str,
@@ -455,8 +465,6 @@ def process_and_execute_notebooks(
             Required. The GCS staging bucket to write executed notebooks to.
         results_file (str):
             Required: The path to the artifacts bucket to save results 
-        accumulative_results (List):
-            Required: The in-memory previous accumulative notebook CI/CD test results.
         variable_project_id (str):
             Required. The value for PROJECT_ID to inject into notebooks.
         variable_region (str):
@@ -578,7 +586,6 @@ def process_and_execute_notebooks(
                 print(log_contents)
 
         _save_results(results_sorted, 
-                      accumulative_results,
                       artifacts_bucket, 
                       results_file)