[DSEC-153] add check end-to-end test (#54323)

<!--Please give us some feedback on your experience writing this PR ! https://app.datadoghq.com/forms/43db4c02-6837-400c-8083-692e141b1b88 !-->

### What does this PR do?

Adds an end-to-end test for the `datasecurity` check: it drives a full check run through a test-only `mock` scan engine and asserts the emitted `sds-result` payload (matches, scanned columns, row count, status).

### Motivation

The scan path (`run_scan` / result building) had no test coverage (DSEC-153).

### Describe how you validated your changes

`cargo test` in the `datasecurity` crate.

### Additional Notes

Test-only: the `mock` engine (`platform: mock`) is compiled under `#[cfg(test)]` and returns caller-controlled `ScanData`; no production behavior changes.

Co-authored-by: aimene.belfodil <aimene.belfodil@datadoghq.com>

Author: aimenebelfodil

pkg/collector/sharedlibrary/rustchecks/checks/datasecurity/src/backend/mock.rs

@@ -0,0 +1,30 @@
+//! Test-only scan engine (`platform: mock`) returning caller-controlled data.
+
+use std::cell::RefCell;
+
+use anyhow::Result;
+
+use super::{ScanData, ScanEngine};
+use crate::config::SubTask;
+
+thread_local! {
+    static DATA: RefCell<ScanData> = RefCell::new(ScanData::default());
+}
+
+/// Sets the [`ScanData`] the mock engine returns from `fetch_data`.
+pub(crate) fn set_data(data: ScanData) {
+    DATA.with(|d| *d.borrow_mut() = data);
+}
+
+pub(super) struct MockEngine;
+pub(super) const ENGINE: MockEngine = MockEngine;
+
+impl ScanEngine for MockEngine {
+    fn name(&self) -> &'static str {
+        "mock"
+    }
+
+    fn fetch_data(&self, _sub_task: &SubTask) -> Result<ScanData> {
+        Ok(DATA.with(|d| d.borrow().clone()))
+    }
+}

pkg/collector/sharedlibrary/rustchecks/checks/datasecurity/src/backend/mod.rs

@@ -8,6 +8,9 @@ use crate::config::SubTask;
 #[cfg(feature = "engine-postgres")]
 mod postgres;
 
+#[cfg(test)]
+pub(crate) mod mock;
+
 /// One scanned column's name and its source data type (e.g. `text`, `varchar`).
 #[derive(Debug, Default, Clone, PartialEq, Eq)]
 pub struct ScannedColumn {
@@ -17,7 +20,7 @@ pub struct ScannedColumn {
 
 /// The result of running a sub task's query: the `{ column: [values] }` map fed
 /// to the scanner, plus metadata describing what was scanned.
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct ScanData {
     /// Column-oriented values consumed by the scanner.
     // TODO(dsec-173): return an `Event` (dd-sensitive-data-scanner) per backend
@@ -43,6 +46,8 @@ fn engines() -> &'static [&'static dyn ScanEngine] {
     &[
         #[cfg(feature = "engine-postgres")]
         &postgres::ENGINE,
+        #[cfg(test)]
+        &mock::ENGINE,
     ]
 }
 

pkg/collector/sharedlibrary/rustchecks/checks/datasecurity/src/check.rs

@@ -91,7 +91,6 @@ fn run_sub_task(
 
 /// Fetches the sub task's data and scans it, returning the matches and the
 /// scanned-table statistics.
-/// TODO(dsec-161): add tests for the scan.
 fn run_scan(scanner: &Scanner, sub_task: &SubTask) -> Result<ScanOutcome> {
     let data = backend::fetch_data(sub_task).context("fetching sub task data")?;
     let matches = scanner
@@ -103,3 +102,138 @@ fn run_scan(scanner: &Scanner, sub_task: &SubTask) -> Result<ScanOutcome> {
         scanned_row_count: data.scanned_row_count,
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use prost::Message;
+    use serde_json::json;
+    use shlib_core::stubs::AggregatorStub;
+
+    use crate::backend::{ScanData, ScannedColumn, mock};
+    use crate::constants::SDS_RESULT_EVENT_TYPE;
+    use crate::proto::{
+        PostgresScannedColumn, PostgresTable, Resource, ScanLocation, ScanMetadata, ScanResult,
+        ScanTaskMetadata, ScanningSource, SdsResultPayload, Status, TableMatch, scan_location,
+        scanning_source,
+    };
+
+    use super::check;
+
+    const INSTANCE: &str = r#"
+task_id: task-1
+scanning_rules:
+  - id: email
+    pattern: '[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]+'
+scan_data:
+  - sub_task_id: sub-1
+    query: SELECT email FROM users
+    timeout_seconds: 5
+    connection:
+      host: mock
+      dbname: app
+    entity:
+      platform: mock
+      database_cluster_name: cluster
+      database_instance_name: inst
+      database: app
+      schema: public
+      table: users
+"#;
+
+    #[test]
+    fn scans_data_and_emits_sds_result() {
+        // The mock engine returns this in place of a real query: two scanned
+        // columns, `email` (which matches) and `name` (which does not).
+        mock::set_data(ScanData {
+            columns: json!({
+                "email": ["alice@corp.io", "bob@corp.io hatem@corp.io"],
+                "name": ["alice", "bob"],
+            }),
+            scanned_columns: vec![
+                ScannedColumn {
+                    name: "email".to_string(),
+                    data_type: "text".to_string(),
+                },
+                ScannedColumn {
+                    name: "name".to_string(),
+                    data_type: "varchar".to_string(),
+                },
+            ],
+            scanned_row_count: 2,
+        });
+
+        let aggregator = AggregatorStub::new();
+        check(&aggregator.agent_check("{}", INSTANCE)).expect("check run failed");
+
+        // Exactly one sds-result event platform event is emitted.
+        let events = aggregator.event_platform_events();
+        assert_eq!(events.len(), 1);
+        assert_eq!(events[0].event_type, SDS_RESULT_EVENT_TYPE);
+
+        // The decoded payload matches in full (timestamp aside, it is clock-based).
+        let payload =
+            SdsResultPayload::decode(events[0].raw_event.as_slice()).expect("payload decodes");
+        assert!(payload.timestamp > 0, "timestamp should be populated");
+
+        assert_eq!(
+            payload,
+            SdsResultPayload {
+                timestamp: payload.timestamp,
+                resource: Some(Resource {
+                    r#type: "postgres_table".to_string(),
+                    name: "inst.app.public.users".to_string(),
+                }),
+                rule_ids: vec!["email".to_string()],
+                scanning_source: Some(ScanningSource {
+                    source: Some(scanning_source::Source::Agent(
+                        scanning_source::Agent::default()
+                    )),
+                }),
+                scan_results: vec![ScanResult {
+                    table_matches: vec![TableMatch {
+                        rule_id: "email".to_string(),
+                        column_name: "email".to_string(),
+                        count_matched_rows: 2,
+                        count_matches: 3,
+                        ..Default::default()
+                    }],
+                    location: Some(ScanLocation {
+                        scan_location: Some(scan_location::ScanLocation::PostgresTable(
+                            PostgresTable {
+                                database_cluster_name: "cluster".to_string(),
+                                database_instance_name: "inst".to_string(),
+                                database_host_name: "mock".to_string(),
+                                database_name: "app".to_string(),
+                                schema_name: "public".to_string(),
+                                table_name: "users".to_string(),
+                                scanned_row_count: 2,
+                                scanned_columns: vec![
+                                    PostgresScannedColumn {
+                                        name: "email".to_string(),
+                                        data_type: "text".to_string(),
+                                    },
+                                    PostgresScannedColumn {
+                                        name: "name".to_string(),
+                                        data_type: "varchar".to_string(),
+                                    },
+                                ],
+                                ..Default::default()
+                            }
+                        )),
+                        ..Default::default()
+                    }),
+                    scan_metadata: Some(ScanMetadata {
+                        scan_task_metadata: Some(ScanTaskMetadata {
+                            task_id: "task-1".to_string(),
+                            sub_task_id: "sub-1".to_string(),
+                            status: Status::Success as i32,
+                            ..Default::default()
+                        }),
+                    }),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }
+        );
+    }
+}