test(bigquery): simplify scripting system test to reduce flakiness (#…

…9458)

No need to filter by time in the scripting system tests. Rather, we only
care about filtering by parent job when testing those features.

This addresses flakiness encountered:

* https://source.cloud.google.com/results/invocations/a29867f9-6492-4477-85d4-d9de8d9ff85b/targets/cloud-devrel%2Fclient-libraries%2Fgoogle-cloud-python%2Fpresubmit%2Fbigquery/log
* https://source.cloud.google.com/results/invocations/cdb77d4a-b0e3-4339-ab55-ae08fb2b297d/targets/cloud-devrel%2Fclient-libraries%2Fgoogle-cloud-python%2Fpresubmit%2Fbigquery/log

bigquery/tests/system.py

@@ -432,7 +432,7 @@ def test_list_tables(self):
         self.assertGreater(len(list(iterator)), 0)
 
     def test_listing_scripting_jobs(self):
-        # run an SQL script
+        # Run a SQL script.
         sql_script = """
             -- Declare a variable to hold names as an array.
             DECLARE top_names ARRAY<STRING>;
@@ -453,31 +453,14 @@ def test_listing_scripting_jobs(self):
             FROM `bigquery-public-data.samples.shakespeare`
             );
         """
-        test_start = datetime.datetime.utcnow()
-        query_job = Config.CLIENT.query(sql_script, project=Config.CLIENT.project)
-        query_job.result()
-
-        # fetch jobs created by the SQL script, sort them into parent and
-        # child jobs
-        script_jobs = list(Config.CLIENT.list_jobs(min_creation_time=test_start))
-
-        parent_jobs = []
-        child_jobs = []
+        parent_job = Config.CLIENT.query(sql_script, project=Config.CLIENT.project)
+        parent_job.result()
 
-        for job in script_jobs:
-            if job.num_child_jobs > 0:
-                parent_jobs.append(job)
-            else:
-                child_jobs.append(job)
-
-        assert len(parent_jobs) == 1  # also implying num_child_jobs > 0
-        assert len(child_jobs) == parent_jobs[0].num_child_jobs
+        # Fetch jobs created by the SQL script.
+        child_jobs = list(Config.CLIENT.list_jobs(parent_job=parent_job))
 
-        # fetch jobs using the parent job filter, verify that results are as expected
-        fetched_jobs = list(Config.CLIENT.list_jobs(parent_job=parent_jobs[0]))
-        assert sorted(job.job_id for job in fetched_jobs) == sorted(
-            job.job_id for job in child_jobs
-        )
+        assert parent_job.num_child_jobs > 0
+        assert len(child_jobs) == parent_job.num_child_jobs
 
     def test_update_table(self):
         dataset = self.temp_dataset(_make_dataset_id("update_table"))