Complete the new TA pipeline implementations

database/models/reports.py

@@ -117,7 +117,7 @@ class Upload(CodecovBaseModel, MixinBaseClass):
     upload_type_id = Column(types.Integer)
 
     @cached_property
-    def flag_names(self):
+    def flag_names(self) -> list[str]:
         return [f.flag_name for f in self.flags]
 
 

database/tests/factories/reports.py

@@ -7,12 +7,14 @@
     Flake,
     RepositoryFlag,
     Test,
+    TestInstance,
     TestResultReportTotals,
 )
 from database.tests.factories.core import (
     CompareCommitFactory,
     ReportFactory,
     RepositoryFactory,
+    UploadFactory,
 )
 
 
@@ -43,6 +45,20 @@ class Meta:
     repository = factory.SubFactory(RepositoryFactory)
 
 
+class TestInstanceFactory(factory.Factory):
+    class Meta:
+        model = TestInstance
+
+    test = factory.SubFactory(TestFactory)
+    upload = factory.SubFactory(UploadFactory)
+
+    outcome = "pass"
+    duration_seconds = 1.5
+    commitid = factory.SelfAttribute("upload.report.commit.commitid")
+    branch = factory.SelfAttribute("upload.report.commit.branch")
+    repoid = factory.SelfAttribute("upload.report.commit.repository.repoid")
+
+
 class FlakeFactory(factory.Factory):
     class Meta:
         model = Flake

django_scaffold/settings.py

@@ -12,11 +12,23 @@
 if "timeseries" in DATABASES:
     DATABASES["timeseries"]["AUTOCOMMIT"] = False
 
+if "test_analytics" in DATABASES:
+    DATABASES["test_analytics"]["AUTOCOMMIT"] = False
+
 IS_DEV = os.getenv("RUN_ENV") == "DEV"
 IS_ENTERPRISE = os.getenv("RUN_ENV") == "ENTERPRISE"
 
 GCS_BUCKET_NAME = get_config("services", "minio", "bucket", default="codecov")
 
+BIGQUERY_WRITE_ENABLED = (
+    get_config("services", "bigquery", "write_enabled", default=False)
+    and "test_analytics" in DATABASES
+)
+
+BIGQUERY_READ_ENABLED = (
+    get_config("services", "bigquery", "read_enabled", default=False)
+    and "test_analytics" in DATABASES
+)
 # Application definition
 INSTALLED_APPS = [
     # dependencies
@@ -40,6 +52,7 @@
     "shared.django_apps.profiling",
     "shared.django_apps.reports",
     "shared.django_apps.staticanalysis",
+    "shared.django_apps.test_analytics",
 ]
 
 TELEMETRY_VANILLA_DB = "default"

protobuf/ta_testrun.proto

@@ -11,6 +11,7 @@ message TestRun {
         PASSED = 0;
         FAILED = 1;
         SKIPPED = 2;
+        FLAKY_FAILED = 3;
     }
 
     optional Outcome outcome = 6;
@@ -27,4 +28,8 @@ message TestRun {
 
     optional string filename = 14;
     optional string framework = 15;
+
+    optional int64 upload_id = 16;
+    optional bytes flags_hash = 17;
+    optional bytes test_id = 18;
 }

requirements.in

@@ -20,6 +20,7 @@ grpcio>=1.66.2
 httpx
 jinja2>=3.1.3
 lxml>=5.3.0
+mmh3>=5.0.1
 mock
 multidict>=6.1.0
 openai

requirements.txt

@@ -208,8 +208,10 @@ markupsafe==2.1.3
     # via jinja2
 minio==7.1.13
     # via shared
-mmh3==4.0.1
-    # via shared
+mmh3==5.0.1
+    # via
+    #   -r requirements.in
+    #   shared
 mock==4.0.3
     # via -r requirements.in
 monotonic==1.5

services/bigquery.py

@@ -1,5 +1,5 @@
 from types import ModuleType
-from typing import Dict, List, cast
+from typing import Dict, List, Sequence, cast
 
 import polars as pl
 from google.api_core import retry
@@ -108,7 +108,17 @@ def write(
 
         self.write_client.batch_commit_write_streams(batch_commit_write_streams_request)
 
-    def query(self, query: str, params: dict | None = None) -> List[Dict]:
+    def query(
+        self,
+        query: str,
+        params: Sequence[
+            bigquery.ScalarQueryParameter
+            | bigquery.RangeQueryParameter
+            | bigquery.ArrayQueryParameter
+            | bigquery.StructQueryParameter
+        ]
+        | None = None,
+    ) -> List[Dict]:
         """Execute a BigQuery SQL query and return results.
         Try not to write INSERT statements and use the write method instead.
 
@@ -125,9 +135,7 @@ def query(self, query: str, params: dict | None = None) -> List[Dict]:
         job_config = bigquery.QueryJobConfig()
 
         if params:
-            job_config.query_parameters = [
-                bigquery.ScalarQueryParameter(k, "STRING", v) for k, v in params.items()
-            ]
+            job_config.query_parameters = params
 
         row_iterator = self.client.query_and_wait(
             query, job_config=job_config, retry=retry.Retry(deadline=30)

services/processing/flake_processing.py

@@ -27,6 +27,17 @@ def process_flake_for_repo_commit(
         state__in=["processed", "v2_finished"],
     ).all()
 
+    process_flakes_for_uploads(repo_id, [upload for upload in uploads])
+
+    log.info(
+        "Successfully processed flakes",
+        extra=dict(repoid=repo_id, commit=commit_id),
+    )
+
+    return {"successful": True}
+
+
+def process_flakes_for_uploads(repo_id: int, uploads: list[ReportSession]):
     curr_flakes = fetch_curr_flakes(repo_id)
     new_flakes: dict[str, Flake] = dict()
 
@@ -92,13 +103,6 @@ def process_flake_for_repo_commit(
         upload.save()
         django_transaction.commit()
 
-    log.info(
-        "Successfully processed flakes",
-        extra=dict(repoid=repo_id, commit=commit_id),
-    )
-
-    return {"successful": True}
-
 
 def get_test_instances(
     upload: ReportSession,