test: Add more destination tests

posthog/temporal/tests/batch_exports/test_bigquery_batch_export_workflow.py

@@ -453,6 +453,7 @@ async def test_bigquery_export_workflow(
 
         run = runs[0]
         assert run.status == "Completed"
+        assert run.records_completed == 100
 
         ingested_timestamp = frozen_time().replace(tzinfo=dt.timezone.utc)
         assert_clickhouse_records_in_bigquery(
@@ -566,6 +567,7 @@ class RefreshError(Exception):
     run = runs[0]
     assert run.status == "Failed"
     assert run.latest_error == "RefreshError: A useful error message"
+    assert run.records_completed == 0
 
 
 async def test_bigquery_export_workflow_handles_cancellation(ateam, bigquery_batch_export, interval):

posthog/temporal/tests/batch_exports/test_postgres_batch_export_workflow.py

@@ -385,6 +385,7 @@ async def test_postgres_export_workflow(
 
     run = runs[0]
     assert run.status == "Completed"
+    assert run.records_completed == 100
 
     await assert_clickhouse_records_in_postgres(
         postgres_connection=postgres_connection,
@@ -494,6 +495,7 @@ class InsufficientPrivilege(Exception):
     run = runs[0]
     assert run.status == "Failed"
     assert run.latest_error == "InsufficientPrivilege: A useful error message"
+    assert run.records_completed == 0
 
 
 async def test_postgres_export_workflow_handles_cancellation(ateam, postgres_batch_export, interval):

posthog/temporal/tests/batch_exports/test_redshift_batch_export_workflow.py

@@ -433,6 +433,7 @@ async def test_redshift_export_workflow(
 
     run = runs[0]
     assert run.status == "Completed"
+    assert run.records_completed == 100
 
     await assert_clickhouse_records_in_redshfit(
         redshift_connection=psycopg_connection,
@@ -559,3 +560,4 @@ class InsufficientPrivilege(Exception):
     run = runs[0]
     assert run.status == "Failed"
     assert run.latest_error == "InsufficientPrivilege: A useful error message"
+    assert run.records_completed == 0

posthog/temporal/tests/batch_exports/test_snowflake_batch_export_workflow.py

@@ -455,6 +455,7 @@ async def test_snowflake_export_workflow_exports_events(
 
     run = runs[0]
     assert run.status == "Completed"
+    assert run.records_completed == 10
 
 
 @pytest.mark.parametrize("interval", ["hour", "day"], indirect=True)
@@ -695,6 +696,7 @@ async def insert_into_snowflake_activity_mocked(_: SnowflakeInsertInputs) -> str
     run = runs[0]
     assert run.status == "FailedRetryable"
     assert run.latest_error == "ValueError: A useful error message"
+    assert run.records_completed == 0
 
 
 async def test_snowflake_export_workflow_handles_insert_activity_non_retryable_errors(ateam, snowflake_batch_export):