re_redap_tests 0.31.2

Official test suite for the Rerun Data Protocol
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

use futures::TryStreamExt as _;
use itertools::Itertools as _;
use re_log_types::EntryId;
use re_protos::cloud::v1alpha1::rerun_cloud_service_server::RerunCloudService;
use re_protos::cloud::v1alpha1::{
    DeleteEntryRequest, FindEntriesRequest, GetTableSchemaRequest, ScanTableRequest,
};
use re_protos::{EntryName, cloud::v1alpha1::ext::EntryDetails};

use crate::tests::common::RerunCloudServiceExt as _;
use crate::{RecordBatchTestExt as _, SchemaTestExt as _};

/// We want to make sure that the "__entries" table is present and has the expected schema and data.
pub async fn list_entries_table(service: impl RerunCloudService) {
    let entries_table_id = entries_table_id(&service).await;

    let schema_request = GetTableSchemaRequest {
        table_id: Some(entries_table_id.into()),
    };

    let schema: arrow::datatypes::Schema = (&service
        .get_table_schema(tonic::Request::new(schema_request))
        .await
        .expect("Failed to get table schema")
        .into_inner()
        .schema
        .expect("Schema should be present"))
        .try_into()
        .expect("Failed to convert schema");

    insta::assert_snapshot!("entries_table_schema", schema.format_snapshot());

    let scan_request = ScanTableRequest {
        table_id: Some(entries_table_id.into()),
    };

    let table_resp: Vec<_> = service
        .scan_table(tonic::Request::new(scan_request))
        .await
        .expect("Failed to scan table")
        .into_inner()
        .try_collect()
        .await
        .expect("Failed to collect scan results");

    let batches = table_resp
        .into_iter()
        .map(|resp| {
            resp.dataframe_part
                .expect("Expected dataframe part")
                .try_into()
                .expect("Failed to decode dataframe")
        })
        .collect_vec();

    let batch =
        re_arrow_util::concat_polymorphic_batches(&batches).expect("Failed to concat batches");

    assert_eq!(batch.schema().fields(), schema.fields());

    let batch = batch.project_columns(&["name", "entry_kind"]);

    insta::assert_snapshot!("entries_table_data", batch.format_snapshot(false));
}

pub async fn entries_table_with_empty_dataset(service: impl RerunCloudService) {
    let dataset_name = "empty_dataset";
    let dataset_entry = service.create_dataset_entry_with_name(dataset_name).await;

    snapshot_entries_table(&service, "entries_table_with_empty_dataset").await;

    service
        .delete_entry(tonic::Request::new(DeleteEntryRequest {
            id: Some(dataset_entry.details.id.into()),
        }))
        .await
        .expect("Failed to delete entry");

    snapshot_entries_table(&service, "entries_table_with_empty_dataset_deleted").await;
}

async fn entries_table_id(service: &impl RerunCloudService) -> EntryId {
    let find_entries_table = FindEntriesRequest {
        filter: Some(re_protos::cloud::v1alpha1::EntryFilter {
            name: Some("__entries".to_owned()),
            ..Default::default()
        }),
    };

    let entries_resp = service
        .find_entries(tonic::Request::new(find_entries_table))
        .await
        .expect("Failed to find entries")
        .into_inner()
        .entries;

    assert_eq!(entries_resp.len(), 1);

    let entries: EntryDetails = entries_resp[0]
        .clone()
        .try_into()
        .expect("Failed to convert to EntryDetails");

    assert_eq!(entries.name, EntryName::entries_table());

    entries.id
}

async fn snapshot_entries_table(service: &impl RerunCloudService, snapshot_name: &str) {
    let entries_table_id = entries_table_id(service).await;

    let entries_resp: Vec<_> = service
        .scan_table(tonic::Request::new(ScanTableRequest {
            table_id: Some(entries_table_id.into()),
        }))
        .await
        .expect("Failed to scan table")
        .into_inner()
        .try_collect()
        .await
        .expect("Failed to collect scan results");

    let batches = entries_resp
        .into_iter()
        .map(|resp| {
            resp.dataframe_part
                .expect("Expected dataframe part")
                .try_into()
                .expect("Failed to decode dataframe")
        })
        .collect_vec();

    let batch =
        re_arrow_util::concat_polymorphic_batches(&batches).expect("Failed to concat batches");

    let batch = batch
        .project_columns(&["name", "entry_kind"])
        .auto_sort_rows()
        .unwrap();

    let mut settings = insta::Settings::clone_current();
    settings.add_filter(
        r"__bp_[0-9a-fA-F]{32}",
        "__bp_********************************",
    );
    settings.bind(|| {
        insta::assert_snapshot!(snapshot_name, batch.format_snapshot(false));
    });
}