test(query_result): added metadata tests

Added tests around query response metadata:
1. execution info upon exception
2. metadata basic verification (state)
3. basic tracing verification

based on:
com.datastax.oss.driver.core.session.ExceptionIT
com.datastax.oss.driver.core.metadata.MetadataIT
com.datastax.oss.driver.core.cql.QueryTraceIT

Author: soyacz

scylla/tests/integration/query_result.rs

@@ -1,14 +1,26 @@
+use std::collections::HashSet;
+use std::env;
+use std::sync::Arc;
+
 use assert_matches::assert_matches;
 use futures::TryStreamExt;
-use scylla::errors::PagerExecutionError;
+use scylla::client::session_builder::SessionBuilder;
+use scylla::errors::{ExecutionError, PagerExecutionError, RequestAttemptError};
+use scylla::observability::history::{HistoryCollector, StructuredHistory};
+use scylla::observability::tracing::TracingInfo;
 use scylla::{
     batch::{Batch, BatchType},
     client::session::Session,
     query::Query,
 };
 use scylla_cql::frame::request::query::{PagingState, PagingStateResponse};
+use scylla_proxy::{Condition, RequestReaction, RequestRule, ShardAwareness};
+use uuid::Uuid;
 
-use crate::utils::{create_new_session_builder, setup_tracing, unique_keyspace_name, PerformDDL};
+use crate::utils::{
+    create_new_session_builder, setup_tracing, test_with_3_node_cluster, unique_keyspace_name,
+    PerformDDL,
+};
 
 const PAGE_SIZE: i32 = 100;
 const ROWS_PER_PARTITION: i32 = 1000;
@@ -295,3 +307,131 @@ async fn execute_iter_serialization_error() {
         Err(PagerExecutionError::SerializationError(_))
     )
 }
+
+#[tokio::test]
+#[ntest::timeout(60000)]
+#[cfg(not(scylla_cloud_tests))]
+async fn should_expose_execution_info_on_exceptions() {
+    setup_tracing();
+    let res = test_with_3_node_cluster(
+        ShardAwareness::Unaware,
+        |proxy_uris, translation_map, mut running_proxy| async move {
+            let session: Session = SessionBuilder::new()
+                .known_node(proxy_uris[0].as_str())
+                .address_translator(Arc::new(translation_map))
+                .build()
+                .await
+                .unwrap();
+
+            let forge_error_rule = RequestRule(Condition::True, RequestReaction::forge().invalid());
+            running_proxy.running_nodes[0]
+                .change_request_rules(Some(vec![forge_error_rule.clone()]));
+            running_proxy.running_nodes[1]
+                .change_request_rules(Some(vec![forge_error_rule.clone()]));
+            running_proxy.running_nodes[2]
+                .change_request_rules(Some(vec![forge_error_rule.clone()]));
+
+            let mut query = Query::from("select * from foo");
+            let history_listener = Arc::new(HistoryCollector::new());
+            query.set_history_listener(history_listener.clone());
+            let err = session
+                .query_unpaged(query, &[])
+                .await
+                .expect_err("expecting error");
+            let structured_history: StructuredHistory = history_listener.clone_structured_history();
+
+            assert_matches!(
+                err,
+                ExecutionError::LastAttemptError(RequestAttemptError::DbError(_, _))
+            );
+            assert_eq!(structured_history.requests.len(), 1);
+            running_proxy
+        },
+    )
+    .await;
+
+    match res {
+        Ok(()) => (),
+        Err(err) => panic!("{}", err),
+    }
+}
+
+#[tokio::test]
+#[ntest::timeout(60000)]
+#[cfg(not(scylla_cloud_tests))]
+async fn test_session_should_have_metadata() {
+    setup_tracing();
+    let session = create_new_session_builder().build().await.unwrap();
+    let state = session.get_cluster_state();
+    let keys = ["SCYLLA_URI", "SCYLLA_URI2", "SCYLLA_URI3"];
+    let expected_addresses: HashSet<String> = keys
+        .iter()
+        .map(|key| env::var(key).unwrap_or_else(|_| panic!("{} not set", key)))
+        .collect();
+
+    let got_addresses: HashSet<String> = state
+        .get_nodes_info()
+        .iter()
+        .map(|node| node.address.to_string())
+        .collect();
+
+    assert_eq!(
+        got_addresses, expected_addresses,
+        "Cluster node addresses do not match environment variables"
+    );
+}
+
+#[tokio::test]
+async fn test_should_not_have_tracing_id_when_tracing_disabled() {
+    setup_tracing();
+    let session = create_new_session_builder().build().await.unwrap();
+    let query: Query = Query::new("SELECT release_version FROM system.local");
+
+    let result = session.query_unpaged(query, &[]).await.unwrap();
+    let tracing_id: Option<Uuid> = result.tracing_id();
+    assert!(tracing_id.is_none());
+}
+
+#[tokio::test]
+async fn test_should_fetch_trace_when_tracing_enabled() {
+    let session = create_new_session_builder().build().await.unwrap();
+
+    let mut query = Query::from("SELECT release_version FROM system.local");
+    query.set_tracing(true);
+
+    let result = session.query_unpaged(query, &[]).await.unwrap();
+    let tracing_id: Option<Uuid> = result.tracing_id();
+    assert!(tracing_id.is_some());
+    let tracing_info: TracingInfo = session
+        .get_tracing_info(&tracing_id.unwrap())
+        .await
+        .unwrap();
+
+    // Verify trace information is present and has expected format
+    assert!(!tracing_info.events.is_empty());
+
+    // Check if the request type matches
+    assert_eq!(tracing_info.request.unwrap(), "Execute CQL3 query");
+
+    // Verify duration is positive
+    assert!(tracing_info.duration.unwrap() > 0);
+
+    // Verify started_at timestamp is present
+    assert!(tracing_info.started_at.unwrap().0 > 0);
+
+    // Check parameters
+    assert!(tracing_info
+        .parameters
+        .clone()
+        .unwrap()
+        .contains_key("consistency_level"));
+    assert!(tracing_info.parameters.unwrap().contains_key("query"));
+
+    // Check events
+    for event in tracing_info.events {
+        assert!(!event.activity.clone().unwrap().is_empty());
+        assert!(event.source.is_some());
+        assert!(event.source_elapsed.unwrap() >= 0);
+        assert!(!event.activity.unwrap().is_empty());
+    }
+}