bhtune_test_support/
mock_bridge.rs

1//! One mock gRPC `Bridge` for driver, CLI, and server tests.
2//!
3//! The field set is the union of the three copies this replaced. Defaults match the CLI and
4//! server mocks: browse pages are complete, and capabilities advertise the 0.4.0/protocol-2
5//! surface. Gateway info stays empty, so a caller that classifies an empty metadata response
6//! as unknown still does. A read response whose single value has tag id `ignored` is expanded
7//! to the requested tag ids; that is the CLI behavior, and it is a no-op unless a test sets
8//! that exact tag id.
9
10use std::net::SocketAddr;
11use std::sync::atomic::{AtomicU32, Ordering};
12use std::sync::{Arc, Mutex, PoisonError};
13
14use opcda_bridge_proto::bridge::bridge_server::{Bridge, BridgeServer};
15use opcda_bridge_proto::bridge::{
16    BrowsePage, BrowseRequest, CloseBrowseSessionRequest, ControlSearchIndexRequest,
17    GetCapabilitiesRequest, GetCapabilitiesResponse, GetGatewayInfoRequest, GetGatewayInfoResponse,
18    GetSearchIndexStatusRequest, ListServersRequest, ListServersResponse, ReadRequest,
19    ReadResponse, RefreshSearchIndexRequest, SearchEvent, SearchIndexRequest, SearchIndexResponse,
20    SearchIndexStatus, SearchRequest, TagValue as ProtoTagValue, WriteRequest, WriteResponse,
21};
22use tokio::sync::{mpsc, oneshot};
23use tokio::task::JoinHandle;
24use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream};
25use tonic::transport::Server;
26use tonic::{Request, Response, Status};
27
28/// Configurable mock `Bridge` service.
29///
30/// Every field is public so a test can set one value and fill the rest with
31/// `..Default::default()`. Counters and request logs are shared `Arc`s so a test can
32/// observe them after the service has moved into the server task.
33#[derive(Clone)]
34pub struct MockBridgeService {
35    /// Returned by `read` unless [`Self::read_error`] or [`Self::fail_read_from_call`] fires.
36    /// A single value whose `tag_id` is `ignored` is cloned once per requested tag id.
37    pub read_response: ReadResponse,
38    /// Returned by `write` unless [`Self::write_error`] is set. `success: false` is a rejected
39    /// write, not an RPC error.
40    pub write_response: WriteResponse,
41    /// Returned by `browse` when the request has no page token, and as the fallback when a
42    /// continuation page was not configured.
43    pub browse_response: BrowsePage,
44    /// Returned by `browse` when the request carries a page token.
45    pub browse_continuation_response: Option<BrowsePage>,
46    /// Returned by `list_servers` unless [`Self::list_servers_error`] is set.
47    pub list_servers_response: ListServersResponse,
48    /// Returned by `get_capabilities` unless [`Self::capabilities_error`] is set.
49    pub capabilities_response: GetCapabilitiesResponse,
50    /// Returned by `get_gateway_info` unless [`Self::gateway_info_error`] is set.
51    pub gateway_info_response: GetGatewayInfoResponse,
52    /// Events forwarded, in order, by `search`.
53    pub search_events: Vec<SearchEvent>,
54    /// Returned by the indexed-search status, refresh, and control RPCs on success.
55    pub search_index_status_response: SearchIndexStatus,
56    /// Returned by `search_index` on success.
57    pub search_index_response: SearchIndexResponse,
58    /// RPC error for `search`, checked before any event is sent.
59    pub search_error: Option<Status>,
60    /// 1-based `read` call number at which later reads fail. Earlier reads still succeed.
61    pub fail_read_from_call: Option<u32>,
62    /// Count of `read` calls, including calls that then fail.
63    pub read_calls: Arc<AtomicU32>,
64    /// Count of `write` calls, including calls that then fail.
65    pub write_calls: Arc<AtomicU32>,
66    /// Count of `close_browse_session` calls that did not return [`Self::close_error`].
67    pub close_browse_session_calls: Arc<AtomicU32>,
68    /// RPC error for `list_servers`.
69    pub list_servers_error: Option<Status>,
70    /// RPC error for `browse`, checked before page-token selection.
71    pub browse_error: Option<Status>,
72    /// RPC error for `get_capabilities`.
73    pub capabilities_error: Option<Status>,
74    /// RPC error for `get_gateway_info`.
75    pub gateway_info_error: Option<Status>,
76    /// Recorded `get_search_index_status` requests, including calls that then fail.
77    pub search_index_status_requests: Arc<Mutex<Vec<GetSearchIndexStatusRequest>>>,
78    /// RPC error for `get_search_index_status`, returned after the request is recorded.
79    pub search_index_status_error: Option<Status>,
80    /// Recorded `search_index` requests, including calls that then fail.
81    pub search_index_requests: Arc<Mutex<Vec<SearchIndexRequest>>>,
82    /// RPC error for `search_index`, returned after the request is recorded.
83    pub search_index_error: Option<Status>,
84    /// Recorded `refresh_search_index` requests, including calls that then fail.
85    pub refresh_search_index_requests: Arc<Mutex<Vec<RefreshSearchIndexRequest>>>,
86    /// RPC error for `refresh_search_index`, returned after the request is recorded.
87    pub refresh_search_index_error: Option<Status>,
88    /// Recorded `control_search_index` requests, including calls that then fail.
89    pub control_search_index_requests: Arc<Mutex<Vec<ControlSearchIndexRequest>>>,
90    /// RPC error for `control_search_index`, returned after the request is recorded.
91    pub control_search_index_error: Option<Status>,
92    /// RPC error for `read`. Checked before [`Self::fail_read_from_call`], so both can be set.
93    pub read_error: Option<Status>,
94    /// RPC error for `write`. Distinct from a `success: false` [`Self::write_response`].
95    pub write_error: Option<Status>,
96    /// RPC error for `close_browse_session`. A failed close is not counted.
97    pub close_error: Option<Status>,
98}
99
100impl Default for MockBridgeService {
101    fn default() -> Self {
102        Self {
103            read_response: ReadResponse::default(),
104            write_response: WriteResponse::default(),
105            browse_response: BrowsePage {
106                complete: true,
107                ..Default::default()
108            },
109            browse_continuation_response: None,
110            list_servers_response: ListServersResponse::default(),
111            capabilities_response: GetCapabilitiesResponse {
112                application_version: "0.4.0".to_string(),
113                protocol_version: "2".to_string(),
114                max_page_size: 1000,
115                supports_browse_sessions: true,
116                supports_search: true,
117                supports_indexed_search: true,
118                indexed_search_protocol_version: "1".to_string(),
119                max_indexed_search_results: 50,
120                ..Default::default()
121            },
122            gateway_info_response: GetGatewayInfoResponse::default(),
123            search_events: Vec::new(),
124            search_index_status_response: SearchIndexStatus::default(),
125            search_index_response: SearchIndexResponse::default(),
126            search_error: None,
127            fail_read_from_call: None,
128            read_calls: Arc::new(AtomicU32::new(0)),
129            write_calls: Arc::new(AtomicU32::new(0)),
130            close_browse_session_calls: Arc::new(AtomicU32::new(0)),
131            list_servers_error: None,
132            browse_error: None,
133            capabilities_error: None,
134            gateway_info_error: None,
135            search_index_status_requests: Arc::new(Mutex::new(Vec::new())),
136            search_index_status_error: None,
137            search_index_requests: Arc::new(Mutex::new(Vec::new())),
138            search_index_error: None,
139            refresh_search_index_requests: Arc::new(Mutex::new(Vec::new())),
140            refresh_search_index_error: None,
141            control_search_index_requests: Arc::new(Mutex::new(Vec::new())),
142            control_search_index_error: None,
143            read_error: None,
144            write_error: None,
145            close_error: None,
146        }
147    }
148}
149
150impl MockBridgeService {
151    /// Fails `read` on the 1-based call `n` and every call after it.
152    #[must_use]
153    pub fn failing_read_from_call(mut self, n: u32) -> Self {
154        self.fail_read_from_call = Some(n);
155        self
156    }
157}
158
159#[tonic::async_trait]
160impl Bridge for MockBridgeService {
161    async fn get_gateway_info(
162        &self,
163        _request: Request<GetGatewayInfoRequest>,
164    ) -> Result<Response<GetGatewayInfoResponse>, Status> {
165        if let Some(status) = self.gateway_info_error.clone() {
166            return Err(status);
167        }
168        Ok(Response::new(self.gateway_info_response.clone()))
169    }
170
171    async fn get_capabilities(
172        &self,
173        _request: Request<GetCapabilitiesRequest>,
174    ) -> Result<Response<GetCapabilitiesResponse>, Status> {
175        if let Some(status) = self.capabilities_error.clone() {
176            return Err(status);
177        }
178        Ok(Response::new(self.capabilities_response.clone()))
179    }
180
181    async fn list_servers(
182        &self,
183        _request: Request<ListServersRequest>,
184    ) -> Result<Response<ListServersResponse>, Status> {
185        if let Some(status) = self.list_servers_error.clone() {
186            return Err(status);
187        }
188        Ok(Response::new(self.list_servers_response.clone()))
189    }
190
191    async fn browse(
192        &self,
193        request: Request<BrowseRequest>,
194    ) -> Result<Response<BrowsePage>, Status> {
195        if let Some(status) = self.browse_error.clone() {
196            return Err(status);
197        }
198        let response = if request.into_inner().page_token.is_some() {
199            self.browse_continuation_response
200                .clone()
201                .unwrap_or_else(|| self.browse_response.clone())
202        } else {
203            self.browse_response.clone()
204        };
205        Ok(Response::new(response))
206    }
207
208    async fn close_browse_session(
209        &self,
210        _request: Request<CloseBrowseSessionRequest>,
211    ) -> Result<Response<()>, Status> {
212        if let Some(status) = self.close_error.clone() {
213            return Err(status);
214        }
215        self.close_browse_session_calls
216            .fetch_add(1, Ordering::SeqCst);
217        Ok(Response::new(()))
218    }
219
220    type SearchStream = ReceiverStream<Result<SearchEvent, Status>>;
221
222    async fn search(
223        &self,
224        _request: Request<SearchRequest>,
225    ) -> Result<Response<Self::SearchStream>, Status> {
226        if let Some(status) = self.search_error.clone() {
227            return Err(status);
228        }
229        let (tx, rx) = mpsc::channel(4);
230        drop(tokio::spawn(forward_search_events(
231            self.search_events.clone(),
232            tx,
233        )));
234        Ok(Response::new(ReceiverStream::new(rx)))
235    }
236
237    async fn get_search_index_status(
238        &self,
239        request: Request<GetSearchIndexStatusRequest>,
240    ) -> Result<Response<SearchIndexStatus>, Status> {
241        record(&self.search_index_status_requests, request.into_inner());
242        if let Some(status) = self.search_index_status_error.clone() {
243            return Err(status);
244        }
245        Ok(Response::new(self.search_index_status_response.clone()))
246    }
247
248    async fn refresh_search_index(
249        &self,
250        request: Request<RefreshSearchIndexRequest>,
251    ) -> Result<Response<SearchIndexStatus>, Status> {
252        record(&self.refresh_search_index_requests, request.into_inner());
253        if let Some(status) = self.refresh_search_index_error.clone() {
254            return Err(status);
255        }
256        Ok(Response::new(self.search_index_status_response.clone()))
257    }
258
259    async fn control_search_index(
260        &self,
261        request: Request<ControlSearchIndexRequest>,
262    ) -> Result<Response<SearchIndexStatus>, Status> {
263        record(&self.control_search_index_requests, request.into_inner());
264        if let Some(status) = self.control_search_index_error.clone() {
265            return Err(status);
266        }
267        Ok(Response::new(self.search_index_status_response.clone()))
268    }
269
270    async fn search_index(
271        &self,
272        request: Request<SearchIndexRequest>,
273    ) -> Result<Response<SearchIndexResponse>, Status> {
274        record(&self.search_index_requests, request.into_inner());
275        if let Some(status) = self.search_index_error.clone() {
276            return Err(status);
277        }
278        Ok(Response::new(self.search_index_response.clone()))
279    }
280
281    async fn read(&self, request: Request<ReadRequest>) -> Result<Response<ReadResponse>, Status> {
282        let call = self.read_calls.fetch_add(1, Ordering::SeqCst) + 1;
283        if let Some(status) = self.read_error.clone() {
284            return Err(status);
285        }
286        if let Some(fail_from) = self.fail_read_from_call
287            && call >= fail_from
288        {
289            return Err(Status::unavailable("mock bridge: simulated read failure"));
290        }
291        let request = request.into_inner();
292        let mut response = self.read_response.clone();
293        if response.values.len() == 1 && response.values[0].tag_id == "ignored" {
294            let template = response.values[0].clone();
295            response.values = request
296                .tag_ids
297                .into_iter()
298                .map(|tag_id| ProtoTagValue {
299                    tag_id,
300                    ..template.clone()
301                })
302                .collect();
303        }
304        Ok(Response::new(response))
305    }
306
307    async fn write(
308        &self,
309        _request: Request<WriteRequest>,
310    ) -> Result<Response<WriteResponse>, Status> {
311        self.write_calls.fetch_add(1, Ordering::SeqCst);
312        if let Some(status) = self.write_error.clone() {
313            return Err(status);
314        }
315        Ok(Response::new(self.write_response.clone()))
316    }
317}
318
319/// A running mock server and the signal that stops it.
320///
321/// Keep this value alive for as long as the test needs the server. Dropping it closes the
322/// shutdown channel, which stops the server, but only [`Self::shutdown`] waits for the
323/// server task to finish. Binding the handle to `_` drops it immediately.
324#[must_use = "dropping the handle stops the mock server"]
325pub struct MockServerHandle {
326    shutdown: oneshot::Sender<()>,
327    task: JoinHandle<()>,
328}
329
330impl MockServerHandle {
331    /// Stops the server and waits until its task has exited.
332    ///
333    /// A panicked or failed server task is not re-raised here. The test that needed the
334    /// server observes that failure through its own connection or assertion.
335    pub async fn shutdown(self) {
336        let _ = self.shutdown.send(());
337        let _ = self.task.await;
338    }
339}
340
341/// Binds `service` to an ephemeral loopback port.
342///
343/// The returned address is `127.0.0.1:<port>`. The handle must stay alive until the test
344/// is done with the server; call [`MockServerHandle::shutdown`] to wait for it to exit.
345pub async fn start_mock_server(
346    service: MockBridgeService,
347) -> std::io::Result<(String, MockServerHandle)> {
348    let addr = SocketAddr::from(([127, 0, 0, 1], 0));
349    let listener = tokio::net::TcpListener::bind(addr).await?;
350    let port = listener.local_addr()?.port();
351    let (shutdown_tx, shutdown_rx) = oneshot::channel();
352    let task = tokio::spawn(async move {
353        let _ = Server::builder()
354            .add_service(BridgeServer::new(service))
355            .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async {
356                let _ = shutdown_rx.await;
357            })
358            .await;
359    });
360    Ok((
361        format!("127.0.0.1:{port}"),
362        MockServerHandle {
363            shutdown: shutdown_tx,
364            task,
365        },
366    ))
367}
368
369async fn forward_search_events(
370    events: Vec<SearchEvent>,
371    tx: mpsc::Sender<Result<SearchEvent, Status>>,
372) {
373    for event in events {
374        if tx.send(Ok(event)).await.is_err() {
375            break;
376        }
377    }
378}
379
380fn record<T>(recorder: &Mutex<Vec<T>>, value: T) {
381    recorder
382        .lock()
383        .unwrap_or_else(PoisonError::into_inner)
384        .push(value);
385}
386
387#[cfg(test)]
388mod tests {
389    use super::{
390        BrowsePage, BrowseRequest, CloseBrowseSessionRequest, ControlSearchIndexRequest,
391        GetCapabilitiesRequest, GetCapabilitiesResponse, GetGatewayInfoRequest,
392        GetGatewayInfoResponse, GetSearchIndexStatusRequest, ListServersRequest,
393        ListServersResponse, MockBridgeService, PoisonError, ReadRequest, ReadResponse,
394        RefreshSearchIndexRequest, SearchEvent, SearchIndexRequest, SearchIndexStatus,
395        SearchRequest, Status, WriteRequest, WriteResponse, forward_search_events, record,
396        start_mock_server,
397    };
398    use std::sync::atomic::Ordering;
399    use std::sync::{Arc, Mutex};
400    use tokio::sync::mpsc;
401    use tonic::Request;
402
403    use opcda_bridge_proto::bridge::bridge_server::Bridge;
404
405    fn configured_error() -> Status {
406        Status::unavailable("configured")
407    }
408
409    #[tokio::test]
410    async fn list_servers_returns_an_empty_response_by_default() {
411        let service = MockBridgeService::default();
412        let response = service
413            .list_servers(Request::new(ListServersRequest::default()))
414            .await
415            .unwrap();
416        assert!(response.into_inner().servers.is_empty());
417    }
418
419    #[tokio::test]
420    async fn list_servers_returns_the_configured_response() {
421        let service = MockBridgeService {
422            list_servers_response: ListServersResponse {
423                servers: vec!["Matrikon.OPC.Simulation.1".to_string()],
424            },
425            ..Default::default()
426        };
427        let response = service
428            .list_servers(Request::new(ListServersRequest::default()))
429            .await
430            .unwrap();
431        assert_eq!(
432            response.into_inner().servers,
433            vec!["Matrikon.OPC.Simulation.1".to_string()]
434        );
435    }
436
437    #[tokio::test]
438    async fn browse_returns_the_configured_page_without_a_token() {
439        let service = MockBridgeService {
440            browse_response: BrowsePage {
441                session_id: "session".to_string(),
442                complete: true,
443                ..Default::default()
444            },
445            browse_continuation_response: Some(BrowsePage {
446                session_id: "not-this".to_string(),
447                ..Default::default()
448            }),
449            ..Default::default()
450        };
451        let page = service
452            .browse(Request::new(BrowseRequest::default()))
453            .await
454            .unwrap()
455            .into_inner();
456        assert_eq!(page.session_id, "session");
457    }
458
459    #[tokio::test]
460    async fn browse_uses_the_continuation_page_when_one_is_configured() {
461        let service = MockBridgeService {
462            browse_continuation_response: Some(BrowsePage {
463                session_id: "continued".to_string(),
464                complete: false,
465                ..Default::default()
466            }),
467            ..Default::default()
468        };
469        let page = service
470            .browse(Request::new(BrowseRequest {
471                page_token: Some("token".into()),
472                ..Default::default()
473            }))
474            .await
475            .unwrap()
476            .into_inner();
477        assert_eq!(page.session_id, "continued");
478        assert!(!page.complete);
479    }
480
481    #[tokio::test]
482    async fn browse_falls_back_when_a_token_has_no_continuation_page() {
483        let service = MockBridgeService::default();
484        let page = service
485            .browse(Request::new(BrowseRequest {
486                page_token: Some("token".into()),
487                ..Default::default()
488            }))
489            .await
490            .unwrap()
491            .into_inner();
492        assert!(page.complete);
493    }
494
495    #[tokio::test]
496    async fn default_capabilities_match_the_cli_and_server_mocks() {
497        let service = MockBridgeService::default();
498        let capabilities = service
499            .get_capabilities(Request::new(GetCapabilitiesRequest::default()))
500            .await
501            .unwrap()
502            .into_inner();
503        assert_eq!(
504            capabilities,
505            GetCapabilitiesResponse {
506                application_version: "0.4.0".to_string(),
507                protocol_version: "2".to_string(),
508                max_page_size: 1000,
509                supports_browse_sessions: true,
510                supports_search: true,
511                supports_indexed_search: true,
512                indexed_search_protocol_version: "1".to_string(),
513                max_indexed_search_results: 50,
514                ..Default::default()
515            }
516        );
517        assert!(service.gateway_info_response.application_version.is_empty());
518    }
519
520    #[tokio::test]
521    async fn gateway_info_returns_the_configured_response() {
522        let service = MockBridgeService {
523            gateway_info_response: GetGatewayInfoResponse {
524                application_version: "test-gateway".to_string(),
525                ..Default::default()
526            },
527            ..Default::default()
528        };
529        let response = service
530            .get_gateway_info(Request::new(GetGatewayInfoRequest::default()))
531            .await
532            .unwrap();
533        assert_eq!(response.into_inner().application_version, "test-gateway");
534    }
535
536    #[tokio::test]
537    async fn configured_rpc_errors_win_over_the_success_response() {
538        let error = configured_error();
539        let service = MockBridgeService {
540            gateway_info_error: Some(error.clone()),
541            capabilities_error: Some(error.clone()),
542            list_servers_error: Some(error.clone()),
543            browse_error: Some(error.clone()),
544            search_error: Some(error.clone()),
545            read_error: Some(error.clone()),
546            write_error: Some(error.clone()),
547            close_error: Some(error.clone()),
548            fail_read_from_call: Some(1),
549            ..Default::default()
550        };
551        assert_eq!(
552            service
553                .get_gateway_info(Request::new(GetGatewayInfoRequest::default()))
554                .await
555                .unwrap_err()
556                .message(),
557            "configured"
558        );
559        assert_eq!(
560            service
561                .get_capabilities(Request::new(GetCapabilitiesRequest::default()))
562                .await
563                .unwrap_err()
564                .message(),
565            "configured"
566        );
567        assert_eq!(
568            service
569                .list_servers(Request::new(ListServersRequest::default()))
570                .await
571                .unwrap_err()
572                .message(),
573            "configured"
574        );
575        assert_eq!(
576            service
577                .browse(Request::new(BrowseRequest {
578                    page_token: Some("token".into()),
579                    ..Default::default()
580                }))
581                .await
582                .unwrap_err()
583                .message(),
584            "configured"
585        );
586        assert_eq!(
587            service
588                .search(Request::new(SearchRequest::default()))
589                .await
590                .unwrap_err()
591                .message(),
592            "configured"
593        );
594        assert_eq!(
595            service
596                .read(Request::new(ReadRequest::default()))
597                .await
598                .unwrap_err()
599                .message(),
600            "configured"
601        );
602        assert_eq!(service.read_calls.load(Ordering::SeqCst), 1);
603        assert_eq!(
604            service
605                .write(Request::new(WriteRequest::default()))
606                .await
607                .unwrap_err()
608                .message(),
609            "configured"
610        );
611        assert_eq!(
612            service
613                .close_browse_session(Request::new(CloseBrowseSessionRequest::default()))
614                .await
615                .unwrap_err()
616                .message(),
617            "configured"
618        );
619        assert_eq!(service.close_browse_session_calls.load(Ordering::SeqCst), 0);
620    }
621
622    #[tokio::test]
623    async fn close_browse_session_counts_only_successful_calls() {
624        let service = MockBridgeService::default();
625        service
626            .close_browse_session(Request::new(CloseBrowseSessionRequest::default()))
627            .await
628            .unwrap();
629        assert_eq!(service.close_browse_session_calls.load(Ordering::SeqCst), 1);
630    }
631
632    #[tokio::test]
633    async fn read_fails_from_the_configured_call_and_expands_ignored_tag_ids() {
634        let service = MockBridgeService {
635            read_response: ReadResponse {
636                values: vec![opcda_bridge_proto::bridge::TagValue {
637                    tag_id: "ignored".to_string(),
638                    value: "10.0".to_string(),
639                    quality: "Good".to_string(),
640                    timestamp: String::new(),
641                }],
642            },
643            ..Default::default()
644        }
645        .failing_read_from_call(2);
646        let first = service
647            .read(Request::new(ReadRequest {
648                tag_ids: vec!["PV".to_string(), "MV".to_string()],
649                ..Default::default()
650            }))
651            .await
652            .unwrap()
653            .into_inner();
654        assert_eq!(
655            first
656                .values
657                .iter()
658                .map(|value| value.tag_id.as_str())
659                .collect::<Vec<_>>(),
660            vec!["PV", "MV"]
661        );
662        assert!(
663            first
664                .values
665                .iter()
666                .all(|value| value.value == "10.0" && value.quality == "Good")
667        );
668        let empty = service
669            .read(Request::new(ReadRequest::default()))
670            .await
671            .unwrap_err();
672        assert_eq!(empty.message(), "mock bridge: simulated read failure");
673        let later = service
674            .read(Request::new(ReadRequest::default()))
675            .await
676            .unwrap_err();
677        assert_eq!(later.message(), "mock bridge: simulated read failure");
678    }
679
680    #[tokio::test]
681    async fn read_leaves_a_real_tag_id_unchanged() {
682        let service = MockBridgeService {
683            read_response: ReadResponse {
684                values: vec![opcda_bridge_proto::bridge::TagValue {
685                    tag_id: "Area.PV".to_string(),
686                    value: "1".to_string(),
687                    quality: "Good".to_string(),
688                    timestamp: String::new(),
689                }],
690            },
691            ..Default::default()
692        };
693        let response = service
694            .read(Request::new(ReadRequest {
695                tag_ids: vec!["Other".to_string()],
696                ..Default::default()
697            }))
698            .await
699            .unwrap()
700            .into_inner();
701        assert_eq!(response.values[0].tag_id, "Area.PV");
702    }
703
704    #[tokio::test]
705    async fn write_returns_a_rejected_write_without_an_rpc_error() {
706        let service = MockBridgeService {
707            write_response: WriteResponse {
708                tag_id: "Area.MV".to_string(),
709                success: false,
710                error: Some("tag is read-only".to_string()),
711            },
712            ..Default::default()
713        };
714        let response = service
715            .write(Request::new(WriteRequest::default()))
716            .await
717            .unwrap()
718            .into_inner();
719        assert!(!response.success);
720        assert_eq!(response.error.as_deref(), Some("tag is read-only"));
721    }
722
723    #[tokio::test]
724    async fn indexed_search_records_the_request_before_returning_an_error() {
725        let status_requests = Arc::new(Mutex::new(Vec::new()));
726        let query_requests = Arc::new(Mutex::new(Vec::new()));
727        let refresh_requests = Arc::new(Mutex::new(Vec::new()));
728        let control_requests = Arc::new(Mutex::new(Vec::new()));
729        let error = configured_error();
730        let service = MockBridgeService {
731            search_index_status_requests: Arc::clone(&status_requests),
732            search_index_status_error: Some(error.clone()),
733            search_index_requests: Arc::clone(&query_requests),
734            search_index_error: Some(error.clone()),
735            refresh_search_index_requests: Arc::clone(&refresh_requests),
736            refresh_search_index_error: Some(error.clone()),
737            control_search_index_requests: Arc::clone(&control_requests),
738            control_search_index_error: Some(error),
739            ..Default::default()
740        };
741        assert!(
742            service
743                .get_search_index_status(Request::new(GetSearchIndexStatusRequest::default()))
744                .await
745                .is_err()
746        );
747        assert!(
748            service
749                .search_index(Request::new(SearchIndexRequest::default()))
750                .await
751                .is_err()
752        );
753        assert!(
754            service
755                .refresh_search_index(Request::new(RefreshSearchIndexRequest::default()))
756                .await
757                .is_err()
758        );
759        assert!(
760            service
761                .control_search_index(Request::new(ControlSearchIndexRequest::default()))
762                .await
763                .is_err()
764        );
765        assert_eq!(status_requests.lock().unwrap().len(), 1);
766        assert_eq!(query_requests.lock().unwrap().len(), 1);
767        assert_eq!(refresh_requests.lock().unwrap().len(), 1);
768        assert_eq!(control_requests.lock().unwrap().len(), 1);
769    }
770
771    #[tokio::test]
772    async fn indexed_search_returns_the_configured_success_response() {
773        let service = MockBridgeService {
774            search_index_status_response: SearchIndexStatus {
775                server: "S1".to_string(),
776                ..Default::default()
777            },
778            ..Default::default()
779        };
780        let status = service
781            .get_search_index_status(Request::new(GetSearchIndexStatusRequest::default()))
782            .await
783            .unwrap()
784            .into_inner();
785        assert_eq!(status.server, "S1");
786        let refreshed = service
787            .refresh_search_index(Request::new(RefreshSearchIndexRequest::default()))
788            .await
789            .unwrap()
790            .into_inner();
791        assert_eq!(refreshed.server, "S1");
792        let controlled = service
793            .control_search_index(Request::new(ControlSearchIndexRequest::default()))
794            .await
795            .unwrap()
796            .into_inner();
797        assert_eq!(controlled.server, "S1");
798        assert!(
799            service
800                .search_index(Request::new(SearchIndexRequest::default()))
801                .await
802                .is_ok()
803        );
804        assert_eq!(
805            service.search_index_status_requests.lock().unwrap().len(),
806            1
807        );
808        assert_eq!(
809            service.refresh_search_index_requests.lock().unwrap().len(),
810            1
811        );
812        assert_eq!(
813            service.control_search_index_requests.lock().unwrap().len(),
814            1
815        );
816        assert_eq!(service.search_index_requests.lock().unwrap().len(), 1);
817    }
818
819    #[tokio::test]
820    async fn search_forwards_events_and_stops_when_the_receiver_is_dropped() {
821        let (tx, mut rx) = mpsc::channel(4);
822        forward_search_events(vec![SearchEvent::default()], tx).await;
823        assert!(rx.recv().await.unwrap().is_ok());
824
825        let (dropped_tx, dropped_rx) = mpsc::channel(4);
826        drop(dropped_rx);
827        forward_search_events(
828            vec![SearchEvent::default(), SearchEvent::default()],
829            dropped_tx,
830        )
831        .await;
832    }
833
834    #[tokio::test]
835    async fn search_stream_can_be_collected_from_the_rpc() {
836        let service = MockBridgeService {
837            search_events: vec![SearchEvent::default()],
838            ..Default::default()
839        };
840        let mut receiver = service
841            .search(Request::new(SearchRequest::default()))
842            .await
843            .unwrap()
844            .into_inner()
845            .into_inner();
846        assert!(receiver.recv().await.unwrap().is_ok());
847        assert!(receiver.recv().await.is_none());
848    }
849
850    #[test]
851    fn record_recovers_a_poisoned_request_log() {
852        let recorder = Mutex::new(Vec::<u32>::new());
853        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
854            let _guard = recorder.lock().unwrap();
855            panic!("poison the request log");
856        }));
857        assert!(recorder.is_poisoned());
858        record(&recorder, 7);
859        let values = recorder.lock().unwrap_or_else(PoisonError::into_inner);
860        assert_eq!(*values, vec![7]);
861    }
862
863    #[tokio::test]
864    async fn start_mock_server_shuts_down_gracefully() {
865        let (_host, server) = start_mock_server(MockBridgeService::default())
866            .await
867            .unwrap();
868        server.shutdown().await;
869    }
870}
bhtune_test_support/mock_bridge.rs

bhtune_test_support/
mock_bridge.rs
