ic_testkit/pocket_ic/
nonblocking.rs

1// Copyright DFINITY Foundation. Licensed under Apache-2.0.
2use crate::pocket_ic::common::rest::{
3    AutoProgressConfig, BlobCompression, BlobId, CanisterHttpRequest, CreateHttpGatewayResponse,
4    CreateInstanceResponse, HttpGatewayBackend, HttpGatewayConfig, HttpGatewayInfo, HttpsConfig,
5    InstanceConfig, InstanceId, MockCanisterHttpResponse, RawAddCycles, RawCanisterCall,
6    RawCanisterHttpRequest, RawCanisterId, RawCanisterResult, RawCanisterSnapshotDownload,
7    RawCanisterSnapshotId, RawCanisterSnapshotUpload, RawCycles, RawEffectivePrincipal,
8    RawIngressStatusArgs, RawMessageId, RawMockCanisterHttpResponse, RawPrincipalId,
9    RawSetStableMemory, RawStableMemory, RawSubnetId, RawTickConfigs, RawTime, SubnetId, Topology,
10};
11#[cfg(windows)]
12use crate::pocket_ic::wsl_path;
13use crate::pocket_ic::{ClientError, Operation, RequestContext, http::Exchange};
14use crate::pocket_ic::{
15    CreateCanisterParams, CreateCanisterPlacement, IngressStatusResult, PocketIcBuilder,
16    RejectResponse, TickConfigs, Time,
17};
18use backon::{BackoffBuilder, ExponentialBuilder};
19use candid::{
20    Principal, decode_args, encode_args,
21    utils::{ArgumentDecoder, ArgumentEncoder},
22};
23use ic_management_canister_types::{
24    CanisterId, CanisterIdRecord, CanisterInstallMode, CanisterLogRecord, CanisterSettings,
25    CanisterStatusResult, ChunkHash, DeleteCanisterSnapshotArgs, FetchCanisterLogsResult,
26    InstallChunkedCodeArgs, InstallCodeArgs, LoadCanisterSnapshotArgs,
27    ProvisionalCreateCanisterWithCyclesArgs, Snapshot, StoredChunksResult,
28    TakeCanisterSnapshotArgs, UpdateSettingsArgs, UploadChunkArgs, UploadChunkResult,
29};
30use ic_platform::certification::cbor::decode_certificate;
31use ic_platform::certification::{Certificate, Label, LookupResult};
32use ic_platform::ic_protocol::Envelope;
33use ic_platform::ic_protocol::EnvelopeContent::ReadState;
34use ic_platform::ic_protocol::{
35    ReadStateResponse, SubnetMetrics, decode_read_state_response, decode_subnet_metrics,
36    encode_envelope,
37};
38use reqwest::{StatusCode, Url};
39use serde::{Serialize, de::DeserializeOwned};
40use sha2::{Digest, Sha256};
41use std::future::Future;
42use std::net::IpAddr;
43use std::path::PathBuf;
44use std::time::Duration;
45use tracing::{debug, instrument};
46
47// wait time between polling requests
48
49mod snapshot;
50
51const LOCALHOST: &str = "localhost";
52
53// The minimum joint size of a canister's WASM
54// and its initial argument blob for which
55// we install the canister WASM as a sequence of chunks.
56// The size is chosen to be smaller than the maximum
57// ingress message size of 2 MiB (2,097,152 B)
58// on application subnet (which have the tighest
59// ingress message size limit)
60// to account for a few additional constant-size fields
61// in the management canister payload to install code
62// and the overhead of candid encoding.
63const INSTALL_CHUNKED_CODE_THRESHOLD: usize = 2_000_000; // 2 MB
64
65// The maximum size of one WASM chunk when installing a canister WASM
66// as a sequence of chunks. This constant is specified
67// in the IC protocol.
68const INSTALL_CODE_CHUNK_SIZE: usize = 1 << 20; // 1 MiB
69
70enum HttpMethod {
71    Get,
72    Post,
73}
74
75/// Main entry point for interacting with PocketIC.
76pub struct PocketIc {
77    /// The unique ID of this PocketIC instance.
78    pub instance_id: InstanceId,
79    // how long a get/post request may retry or poll
80    max_request_time_ms: Option<u64>,
81    http_gateway: Option<HttpGatewayInfo>,
82    server_url: Url,
83    reqwest_client: reqwest::Client,
84    // the instance should only be deleted when dropping this handle if this handle owns the instance
85    owns_instance: bool,
86}
87
88impl PocketIc {
89    /// Creates a new PocketIC instance with a single application subnet on the server.
90    /// The caller supplies an already admitted server endpoint.
91    pub async fn new(server_url: Url) -> Result<Self, ClientError> {
92        PocketIcBuilder::new()
93            .with_server_url(server_url)
94            .with_application_subnet()
95            .build()
96            .await
97    }
98
99    /// Creates a PocketIC handle to an existing instance on a running server.
100    /// Note that this handle does not extend the lifetime of the existing instance,
101    /// i.e., the existing instance is deleted and this handle stops working
102    /// when the PocketIC handle that created the existing instance is dropped.
103    pub fn new_from_existing_instance(
104        server_url: Url,
105        instance_id: InstanceId,
106        max_request_time_ms: Option<u64>,
107    ) -> Self {
108        let reqwest_client = reqwest::Client::new();
109        debug!("instance_id={} Reusing existing instance.", instance_id);
110
111        Self {
112            instance_id,
113            max_request_time_ms,
114            http_gateway: None,
115            server_url,
116            reqwest_client,
117            owns_instance: false,
118        }
119    }
120
121    pub(crate) async fn from_config(
122        server_url: Url,
123        max_request_time_ms: Option<u64>,
124        instance_config: InstanceConfig,
125    ) -> Result<Self, ClientError> {
126        #[cfg(windows)]
127        let instance_config = InstanceConfig {
128            state_dir: instance_config
129                .state_dir
130                .map(|path| wsl_path(&path, "state directory").into()),
131            ..instance_config
132        };
133
134        let reqwest_client = reqwest::Client::new();
135        let url = server_url.join("instances").unwrap();
136        let mut exchange = Exchange::new(&url, max_request_time_ms);
137        let response: CreateInstanceResponse = exchange
138            .json(
139                &reqwest_client,
140                reqwest_client.post(url.clone()).json(&instance_config),
141            )
142            .await?;
143        let (instance_id, http_gateway_info) = match response {
144            CreateInstanceResponse::Created {
145                instance_id,
146                http_gateway_info,
147                ..
148            } => (instance_id, http_gateway_info),
149            CreateInstanceResponse::Error { message } => {
150                return Err(ClientError::Server {
151                    context: Box::new(exchange.context.clone()),
152                    status: exchange.context.status.unwrap_or(StatusCode::OK),
153                    message,
154                });
155            }
156        };
157        debug!("instance_id={} New instance created.", instance_id);
158
159        Ok(Self {
160            instance_id,
161            max_request_time_ms,
162            http_gateway: http_gateway_info,
163            server_url,
164            reqwest_client,
165            owns_instance: true,
166        })
167    }
168
169    /// Returns the URL of the PocketIC server on which this PocketIC instance is running.
170    pub fn get_server_url(&self) -> Url {
171        self.server_url.clone()
172    }
173
174    /// Returns the topology of the different subnets of this PocketIC instance.
175    pub async fn topology(&self) -> Result<Topology, ClientError> {
176        let endpoint = "read/topology";
177        self.get(endpoint).await
178    }
179
180    /// Upload and store a binary blob to the PocketIC server.
181    #[instrument(ret, skip(self, blob), fields(instance_id=self.instance_id, blob_len = %blob.len(), compression = ?compression))]
182    pub async fn upload_blob(
183        &self,
184        blob: Vec<u8>,
185        compression: BlobCompression,
186    ) -> Result<BlobId, ClientError> {
187        let url = self.server_url.join("blobstore/").unwrap();
188        let mut request = self.reqwest_client.post(url.clone()).body(blob);
189        if let BlobCompression::Gzip = compression {
190            request = request.header(reqwest::header::CONTENT_ENCODING, "gzip");
191        }
192        let mut exchange = Exchange::new(&url, self.max_request_time_ms);
193        let (status, bytes) = exchange.bytes(&self.reqwest_client, request).await?;
194        if !status.is_success() {
195            return Err(exchange.server_error(status, &bytes));
196        }
197        let hash = hex::decode(bytes).map_err(|error| ClientError::Protocol(error.to_string()))?;
198        Ok(BlobId(hash))
199    }
200
201    /// Set stable memory of a canister. Optional GZIP compression can be used for reduced
202    /// data traffic.
203    #[instrument(skip(self, data), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), data_len = %data.len(), compression = ?compression))]
204    pub async fn set_stable_memory(
205        &self,
206        canister_id: CanisterId,
207        data: Vec<u8>,
208        compression: BlobCompression,
209    ) -> Result<(), ClientError> {
210        let blob_id = self.upload_blob(data, compression).await?;
211        let endpoint = "update/set_stable_memory";
212        self.post::<(), _>(
213            endpoint,
214            RawSetStableMemory {
215                canister_id: canister_id.as_slice().to_vec(),
216                blob_id,
217            },
218        )
219        .await?;
220
221        Ok(())
222    }
223
224    /// Get stable memory of a canister.
225    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string()))]
226    pub async fn get_stable_memory(&self, canister_id: CanisterId) -> Result<Vec<u8>, ClientError> {
227        let endpoint = "read/get_stable_memory";
228        let RawStableMemory { blob } = self
229            .post(
230                endpoint,
231                RawCanisterId {
232                    canister_id: canister_id.as_slice().to_vec(),
233                },
234            )
235            .await?;
236        Ok(blob)
237    }
238
239    /// List all instances and their status.
240    #[instrument(ret)]
241    pub async fn list_instances(server_url: Url) -> Result<Vec<String>, ClientError> {
242        let url = server_url.join("instances").unwrap();
243        let client = reqwest::Client::new();
244        Exchange::new(&url, None)
245            .json(&client, client.get(url.clone()))
246            .await
247    }
248
249    /// Make the IC produce and progress by one block.
250    /// Note that multiple ticks might be necessary to observe
251    /// an expected effect, e.g., if the effect depends on
252    /// inter-canister calls or heartbeats.
253    #[instrument(skip(self), fields(instance_id=self.instance_id))]
254    pub async fn tick(&self) -> Result<(), ClientError> {
255        self.tick_with_configs(TickConfigs::default()).await?;
256
257        Ok(())
258    }
259
260    /// Make the IC produce and progress by one block with custom
261    /// configs for the round.
262    #[instrument(skip(self), fields(instance_id=self.instance_id))]
263    pub async fn tick_with_configs(&self, configs: TickConfigs) -> Result<(), ClientError> {
264        let endpoint = "update/tick";
265        self.post::<(), RawTickConfigs>(endpoint, configs.into())
266            .await?;
267
268        Ok(())
269    }
270
271    /// Configures the IC to make progress automatically,
272    /// i.e., periodically update the time of the IC
273    /// to the real time and execute rounds on the subnets.
274    /// Only returns after the certified time of the IC
275    /// has been updated for the first time.
276    /// Returns the URL at which `/api` requests
277    /// for this instance can be made.
278    #[instrument(skip(self), fields(instance_id=self.instance_id))]
279    pub async fn auto_progress(&self) -> Result<Url, ClientError> {
280        let endpoint = "auto_progress";
281        let auto_progress_config = AutoProgressConfig {
282            artificial_delay_ms: None,
283        };
284        self.post::<(), _>(endpoint, auto_progress_config).await?;
285        Ok(self.instance_url())
286    }
287
288    /// Returns whether automatic progress is enabled on the PocketIC instance.
289    #[instrument(skip(self), fields(instance_id=self.instance_id))]
290    pub async fn auto_progress_enabled(&self) -> Result<bool, ClientError> {
291        self.get("auto_progress").await
292    }
293
294    pub(crate) fn instance_url(&self) -> Url {
295        self.server_url
296            .join("/instances/")
297            .unwrap()
298            .join(&format!("{}/", self.instance_id))
299            .unwrap()
300    }
301
302    /// Stops automatic progress (see `auto_progress`) on the IC.
303    #[instrument(skip(self), fields(instance_id=self.instance_id))]
304    pub async fn stop_progress(&self) -> Result<(), ClientError> {
305        let endpoint = "stop_progress";
306        self.post::<(), _>(endpoint, "").await?;
307
308        Ok(())
309    }
310
311    /// Returns the URL at which `/api` requests
312    /// for this instance can be made if the HTTP
313    /// gateway has been started.
314    pub fn url(&self) -> Option<Url> {
315        self.http_gateway
316            .as_ref()
317            .map(|res| Url::parse(&format!("http://{}:{}/", LOCALHOST, res.port)).unwrap())
318    }
319
320    /// Creates an HTTP gateway for this PocketIC instance binding to `127.0.0.1`
321    /// and an optionally specified port (defaults to choosing an arbitrary unassigned port);
322    /// listening on `localhost`;
323    /// and configures the PocketIC instance to make progress automatically, i.e.,
324    /// periodically update the time of the PocketIC instance to the real time
325    /// and process messages on the PocketIC instance.
326    /// Only returns after the certified time of the PocketIC instance
327    /// has been updated for the first time.
328    /// Returns the URL at which `/api` requests
329    /// for this instance can be made.
330    #[instrument(skip(self), fields(instance_id=self.instance_id))]
331    pub async fn make_live(&mut self, listen_at: Option<u16>) -> Result<Url, ClientError> {
332        self.make_live_with_params(None, listen_at, None, None)
333            .await
334    }
335
336    /// Creates an HTTP gateway for this PocketIC instance binding
337    /// to an optionally specified IP address (defaults to `127.0.0.1`)
338    /// and port (defaults to choosing an arbitrary unassigned port);
339    /// listening on optionally specified domains (default to `localhost`);
340    /// and using an optionally specified TLS certificate (if provided, an HTTPS gateway is created)
341    /// and configures the PocketIC instance to make progress automatically, i.e.,
342    /// periodically update the time of the PocketIC instance to the real time
343    /// and process messages on the PocketIC instance.
344    /// Only returns after the certified time of the PocketIC instance
345    /// has been updated for the first time.
346    /// Returns the URL at which `/api` requests
347    /// for this instance can be made.
348    #[instrument(skip(self), fields(instance_id=self.instance_id))]
349    pub async fn make_live_with_params(
350        &mut self,
351        ip_addr: Option<IpAddr>,
352        listen_at: Option<u16>,
353        domains: Option<Vec<String>>,
354        https_config: Option<HttpsConfig>,
355    ) -> Result<Url, ClientError> {
356        if let Some(url) = self.url() {
357            return Ok(url);
358        }
359        if !self.auto_progress_enabled().await? {
360            self.auto_progress().await?;
361        }
362        self.start_http_gateway(
363            ip_addr.map(|ip_addr| ip_addr.to_string()),
364            listen_at,
365            domains,
366            https_config,
367        )
368        .await
369    }
370
371    async fn start_http_gateway(
372        &mut self,
373        ip_addr: Option<String>,
374        port: Option<u16>,
375        domains: Option<Vec<String>>,
376        https_config: Option<HttpsConfig>,
377    ) -> Result<Url, ClientError> {
378        let endpoint = self.server_url.join("http_gateway").unwrap();
379        let http_gateway_config = HttpGatewayConfig {
380            ip_addr,
381            port,
382            forward_to: HttpGatewayBackend::PocketIcInstance(self.instance_id),
383            domains: domains.clone(),
384            https_config: https_config.clone(),
385            domain_custom_provider_local_file: None,
386        };
387        let mut exchange = Exchange::new(&endpoint, self.max_request_time_ms);
388        let res: CreateHttpGatewayResponse = exchange
389            .json(
390                &self.reqwest_client,
391                self.reqwest_client
392                    .post(endpoint.clone())
393                    .json(&http_gateway_config),
394            )
395            .await?;
396        Ok(match res {
397            CreateHttpGatewayResponse::Created(info) => {
398                let port = info.port;
399                self.http_gateway = Some(info);
400                let proto = if https_config.is_some() {
401                    "https"
402                } else {
403                    "http"
404                };
405                Url::parse(&format!(
406                    "{}://{}:{}/",
407                    proto,
408                    domains
409                        .unwrap_or_default()
410                        .into_iter()
411                        .next()
412                        .unwrap_or(LOCALHOST.to_string()),
413                    port
414                ))
415                .unwrap()
416            }
417            CreateHttpGatewayResponse::Error { message } => {
418                return Err(ClientError::Server {
419                    context: Box::new(exchange.context.clone()),
420                    status: exchange.context.status.unwrap_or(StatusCode::OK),
421                    message,
422                });
423            }
424        })
425    }
426
427    async fn stop_http_gateway(&mut self) -> Result<(), ClientError> {
428        if let Some(info) = &self.http_gateway {
429            let url = self
430                .server_url
431                .join(&format!("http_gateway/{}/stop", info.instance_id))
432                .unwrap();
433            Exchange::new(&url, self.max_request_time_ms)
434                .json::<()>(&self.reqwest_client, self.reqwest_client.post(url.clone()))
435                .await?;
436            self.http_gateway = None;
437        }
438        Ok(())
439    }
440
441    /// Stops auto progress (automatic time updates and round executions)
442    /// and the HTTP gateway for this IC instance.
443    #[instrument(skip(self), fields(instance_id=self.instance_id))]
444    pub async fn stop_live(&mut self) -> Result<(), ClientError> {
445        self.stop_http_gateway().await?;
446        self.stop_progress().await?;
447
448        Ok(())
449    }
450
451    /// Get the root key of this IC instance. Returns `None` if the IC has no NNS subnet.
452    #[instrument(skip(self), fields(instance_id=self.instance_id))]
453    pub async fn root_key(&self) -> Result<Option<Vec<u8>>, ClientError> {
454        let Some(subnet_id) = self.topology().await?.get_nns() else {
455            return Ok(None);
456        };
457        let subnet_id: RawSubnetId = subnet_id.into();
458        let endpoint = "read/pub_key";
459        let res = self.post::<Vec<u8>, _>(endpoint, subnet_id).await?;
460        Ok(Some(res))
461    }
462
463    /// Get the current time of the IC.
464    #[instrument(ret, skip(self), fields(instance_id=self.instance_id))]
465    pub async fn get_time(&self) -> Result<Time, ClientError> {
466        let endpoint = "read/get_time";
467        let result: RawTime = self.get(endpoint).await?;
468        Ok(Time::from_nanos_since_unix_epoch(result.nanos_since_epoch))
469    }
470
471    /// Set the current time of the IC, on all subnets.
472    #[instrument(skip(self), fields(instance_id=self.instance_id, time = ?time))]
473    pub async fn set_time(&self, time: Time) -> Result<(), ClientError> {
474        let endpoint = "update/set_time";
475        self.post::<(), _>(
476            endpoint,
477            RawTime {
478                nanos_since_epoch: time.as_nanos_since_unix_epoch(),
479            },
480        )
481        .await?;
482
483        Ok(())
484    }
485
486    /// Set the current certified time of the IC, on all subnets.
487    #[instrument(skip(self), fields(instance_id=self.instance_id, time = ?time))]
488    pub async fn set_certified_time(&self, time: Time) -> Result<(), ClientError> {
489        let endpoint = "update/set_certified_time";
490        self.post::<(), _>(
491            endpoint,
492            RawTime {
493                nanos_since_epoch: time.as_nanos_since_unix_epoch(),
494            },
495        )
496        .await?;
497
498        Ok(())
499    }
500
501    /// Advance the time on the IC on all subnets by some nanoseconds.
502    #[instrument(skip(self), fields(instance_id=self.instance_id, duration = ?duration))]
503    pub async fn advance_time(&self, duration: Duration) -> Result<(), ClientError> {
504        let now = self.get_time().await?;
505        self.set_time(now + duration).await?;
506
507        Ok(())
508    }
509
510    /// Get the controllers of a canister.
511    /// Returns a server error if the canister does not exist.
512    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string()))]
513    pub async fn get_controllers(
514        &self,
515        canister_id: CanisterId,
516    ) -> Result<Vec<Principal>, ClientError> {
517        let endpoint = "read/get_controllers";
518        let result: Vec<RawPrincipalId> = self
519            .post(
520                endpoint,
521                RawCanisterId {
522                    canister_id: canister_id.as_slice().to_vec(),
523                },
524            )
525            .await?;
526        Ok(result.into_iter().map(|p| p.into()).collect())
527    }
528
529    /// Get the current cycles balance of a canister.
530    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string()))]
531    pub async fn cycle_balance(&self, canister_id: CanisterId) -> Result<u128, ClientError> {
532        let endpoint = "read/get_cycles";
533        let result: RawCycles = self
534            .post(
535                endpoint,
536                RawCanisterId {
537                    canister_id: canister_id.as_slice().to_vec(),
538                },
539            )
540            .await?;
541        Ok(result.cycles)
542    }
543
544    /// Add cycles to a canister. Returns the new balance.
545    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), amount = %amount))]
546    pub async fn add_cycles(
547        &self,
548        canister_id: CanisterId,
549        amount: u128,
550    ) -> Result<u128, ClientError> {
551        let endpoint = "update/add_cycles";
552        let result: RawCycles = self
553            .post(
554                endpoint,
555                RawAddCycles {
556                    canister_id: canister_id.as_slice().to_vec(),
557                    amount,
558                },
559            )
560            .await?;
561        Ok(result.cycles)
562    }
563
564    /// Submit an update call (without executing it immediately).
565    pub async fn submit_call(
566        &self,
567        canister_id: CanisterId,
568        sender: Principal,
569        method: &str,
570        payload: Vec<u8>,
571    ) -> Result<RawMessageId, ClientError> {
572        self.submit_call_with_effective_principal(
573            canister_id,
574            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
575            sender,
576            method,
577            payload,
578        )
579        .await
580    }
581
582    /// Submit an update with explicit routing, retaining its accepted ingress receipt.
583    pub async fn submit_call_with_effective_principal(
584        &self,
585        canister_id: CanisterId,
586        effective_principal: RawEffectivePrincipal,
587        sender: Principal,
588        method: &str,
589        payload: Vec<u8>,
590    ) -> Result<RawMessageId, ClientError> {
591        let endpoint = "update/submit_ingress_message";
592        let raw_canister_call = RawCanisterCall {
593            sender: sender.as_slice().to_vec(),
594            canister_id: canister_id.as_slice().to_vec(),
595            method: method.to_string(),
596            payload,
597            effective_principal,
598            sender_info: None,
599        };
600        self.post::<Result<RawMessageId, RejectResponse>, _>(endpoint, raw_canister_call)
601            .await?
602            .map_err(ClientError::from)
603    }
604
605    /// Await an update call submitted previously by `submit_call` or `submit_call_with_effective_principal`.
606    pub async fn await_call(&self, message_id: RawMessageId) -> Result<Vec<u8>, ClientError> {
607        let endpoint = "update/await_ingress_message";
608        let result: RawCanisterResult = self
609            .post(endpoint, &message_id)
610            .await
611            .map_err(|error| error.with_ingress(message_id))?;
612        Result::<Vec<u8>, RejectResponse>::from(result).map_err(ClientError::from)
613    }
614
615    /// Fetch the status of an update call submitted previously by `submit_call` or `submit_call_with_effective_principal`.
616    /// Note that the status of the update call can only change if the PocketIC instance is in live mode
617    /// or a round has been executed due to a separate PocketIC library call, e.g., `PocketIc::tick()`.
618    pub async fn ingress_status(
619        &self,
620        raw_message_id: RawMessageId,
621    ) -> Result<Option<Result<Vec<u8>, RejectResponse>>, ClientError> {
622        let status = self.ingress_status_as_caller(raw_message_id, None).await?;
623        Ok(match status {
624            IngressStatusResult::NotAvailable => None,
625            IngressStatusResult::Success(status) => Some(status),
626            IngressStatusResult::Forbidden(err) => return Err(ClientError::Protocol(err)),
627        })
628    }
629
630    /// Fetch the status of an update call submitted previously by `submit_call` or `submit_call_with_effective_principal`.
631    /// Note that the status of the update call can only change if the PocketIC instance is in live mode
632    /// or a round has been executed due to a separate PocketIC library call, e.g., `PocketIc::tick()`.
633    /// If the status of the update call is known, but the update call was submitted by a different caller, then an error is returned.
634    pub async fn ingress_status_as(
635        &self,
636        raw_message_id: RawMessageId,
637        caller: Principal,
638    ) -> Result<IngressStatusResult, ClientError> {
639        self.ingress_status_as_caller(raw_message_id, Some(caller))
640            .await
641    }
642
643    async fn ingress_status_as_caller(
644        &self,
645        raw_message_id: RawMessageId,
646        caller: Option<Principal>,
647    ) -> Result<IngressStatusResult, ClientError> {
648        let args = RawIngressStatusArgs {
649            raw_message_id: raw_message_id.clone(),
650            raw_caller: caller.map(Into::into),
651        };
652        match self
653            .post::<Option<RawCanisterResult>, _>("read/ingress_status", args)
654            .await
655        {
656            Ok(None) => Ok(IngressStatusResult::NotAvailable),
657            Ok(Some(result)) => Ok(IngressStatusResult::Success(result.into())),
658            Err(ClientError::Server {
659                status: StatusCode::FORBIDDEN,
660                message,
661                ..
662            }) => Ok(IngressStatusResult::Forbidden(message)),
663            Err(error) => Err(error.with_ingress(raw_message_id)),
664        }
665    }
666
667    /// Await an update call submitted previously by `submit_call` or `submit_call_with_effective_principal`.
668    /// Note that the status of the update call can only change if the PocketIC instance is in live mode
669    /// or a round has been executed due to a separate PocketIC library call.
670    pub async fn await_call_no_ticks(
671        &self,
672        message_id: RawMessageId,
673    ) -> Result<Vec<u8>, ClientError> {
674        let mut retry_policy = ExponentialBuilder::new()
675            .with_min_delay(Duration::from_millis(10))
676            .with_max_delay(Duration::from_secs(1))
677            .with_factor(2.0)
678            .with_jitter()
679            .with_total_delay(Some(Duration::from_secs(15 * 60)))
680            .without_max_times()
681            .build();
682        Ok(loop {
683            if let Some(ingress_status) = self.ingress_status(message_id.clone()).await? {
684                break ingress_status.map_err(ClientError::from)?;
685            }
686            let delay = retry_policy.next().ok_or_else(|| ClientError::Timeout {
687                context: Box::new(RequestContext {
688                    url: self.instance_url().to_string(),
689                    operation: None,
690                    ingress: Some(message_id.clone()),
691                    may_have_executed: true,
692                    status: None,
693                }),
694            })?;
695            tokio::time::sleep(delay).await;
696        })
697    }
698
699    /// Execute an update call on a canister.
700    #[instrument(skip(self, payload), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.to_string(), method = %method, payload_len = %payload.len()))]
701    pub async fn update_call(
702        &self,
703        canister_id: CanisterId,
704        sender: Principal,
705        method: &str,
706        payload: Vec<u8>,
707    ) -> Result<Vec<u8>, ClientError> {
708        self.update_call_with_effective_principal(
709            canister_id,
710            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
711            sender,
712            method,
713            payload,
714        )
715        .await
716    }
717
718    /// Submit once with explicit routing, then await the accepted ingress.
719    pub async fn update_call_with_effective_principal(
720        &self,
721        canister_id: CanisterId,
722        effective_principal: RawEffectivePrincipal,
723        sender: Principal,
724        method: &str,
725        payload: Vec<u8>,
726    ) -> Result<Vec<u8>, ClientError> {
727        let message_id = self
728            .submit_call_with_effective_principal(
729                canister_id,
730                effective_principal,
731                sender,
732                method,
733                payload,
734            )
735            .await?;
736        self.await_call(message_id).await
737    }
738
739    /// Execute a query call on a canister.
740    #[instrument(skip(self, payload), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.to_string(), method = %method, payload_len = %payload.len()))]
741    pub async fn query_call(
742        &self,
743        canister_id: CanisterId,
744        sender: Principal,
745        method: &str,
746        payload: Vec<u8>,
747    ) -> Result<Vec<u8>, ClientError> {
748        self.query_call_with_effective_principal(
749            canister_id,
750            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
751            sender,
752            method,
753            payload,
754        )
755        .await
756    }
757
758    /// Execute a query call on a canister explicitly specifying an effective principal to route the request:
759    /// this API is useful for making generic query calls (including management canister query calls) without using dedicated functions from this library
760    /// (e.g., making generic query calls in dfx to a PocketIC instance).
761    #[instrument(skip(self, payload), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), effective_principal = %effective_principal.to_string(), sender = %sender.to_string(), method = %method, payload_len = %payload.len()))]
762    pub async fn query_call_with_effective_principal(
763        &self,
764        canister_id: CanisterId,
765        effective_principal: RawEffectivePrincipal,
766        sender: Principal,
767        method: &str,
768        payload: Vec<u8>,
769    ) -> Result<Vec<u8>, ClientError> {
770        let endpoint = "read/query";
771        let raw_canister_call = RawCanisterCall {
772            sender: sender.as_slice().to_vec(),
773            canister_id: canister_id.as_slice().to_vec(),
774            method: method.to_string(),
775            payload,
776            effective_principal,
777            sender_info: None,
778        };
779        let result: RawCanisterResult = self.post(endpoint, raw_canister_call).await?;
780        Result::<Vec<u8>, RejectResponse>::from(result).map_err(ClientError::from)
781    }
782
783    /// Fetch canister logs via a query call to the management canister.
784    pub async fn fetch_canister_logs(
785        &self,
786        canister_id: CanisterId,
787        sender: Principal,
788    ) -> Result<Vec<CanisterLogRecord>, ClientError> {
789        Ok(with_candid::<_, (FetchCanisterLogsResult,), _>(
790            (CanisterIdRecord { canister_id },),
791            |payload| async {
792                self.query_call_with_effective_principal(
793                    Principal::management_canister(),
794                    RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
795                    sender,
796                    "fetch_canister_logs",
797                    payload,
798                )
799                .await
800            },
801        )
802        .await?
803        .0
804        .canister_log_records)
805    }
806
807    /// Request a canister's status.
808    #[instrument(skip(self), fields(instance_id=self.instance_id, sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
809    pub async fn canister_status(
810        &self,
811        canister_id: CanisterId,
812        sender: Option<Principal>,
813    ) -> Result<CanisterStatusResult, ClientError> {
814        Ok(
815            call_candid_as::<(CanisterIdRecord,), (CanisterStatusResult,)>(
816                self,
817                Principal::management_canister(),
818                RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
819                sender.unwrap_or(Principal::anonymous()),
820                "canister_status",
821                (CanisterIdRecord { canister_id },),
822            )
823            .await?
824            .0,
825        )
826    }
827
828    /// Create a canister with default settings as the anonymous principal.
829    /// The canister is created with 100T cycles.
830    #[instrument(ret, skip(self), fields(instance_id=self.instance_id))]
831    pub async fn create_canister(&self) -> Result<CanisterId, ClientError> {
832        self.create_canister_with_params(None, CreateCanisterParams::default())
833            .await
834    }
835
836    /// Create a canister with optional custom settings and a sender.
837    /// The canister is created with 100T cycles.
838    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, settings = ?settings, sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
839    pub async fn create_canister_with_settings(
840        &self,
841        sender: Option<Principal>,
842        settings: Option<CanisterSettings>,
843    ) -> Result<CanisterId, ClientError> {
844        self.create_canister_with_params(
845            sender,
846            CreateCanisterParams {
847                settings,
848                ..Default::default()
849            },
850        )
851        .await
852    }
853
854    /// Creates a canister with a specific canister ID and optional custom settings.
855    /// The canister is created with 100T cycles.
856    /// Returns an error if the canister ID is already in use.
857    /// Creates a new subnet if the canister ID is not contained in any of the subnets.
858    ///
859    /// The canister ID must be an IC mainnet canister ID that does not belong to the NNS or II subnet,
860    /// otherwise the function might panic (for NNS and II canister IDs,
861    /// the PocketIC instance should already be created with those subnets).
862    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, sender = %sender.unwrap_or(Principal::anonymous()).to_string(), settings = ?settings, canister_id = %canister_id.to_string()))]
863    pub async fn create_canister_with_id(
864        &self,
865        sender: Option<Principal>,
866        settings: Option<CanisterSettings>,
867        canister_id: CanisterId,
868    ) -> Result<CanisterId, ClientError> {
869        self.create_canister_with_params(
870            sender,
871            CreateCanisterParams {
872                settings,
873                placement: Some(CreateCanisterPlacement::CanisterId(canister_id)),
874                ..Default::default()
875            },
876        )
877        .await
878    }
879
880    /// Create a canister on a specific subnet with optional custom settings.
881    /// The canister is created with 100T cycles.
882    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, sender = %sender.unwrap_or(Principal::anonymous()).to_string(), settings = ?settings, subnet_id = %subnet_id.to_string()))]
883    pub async fn create_canister_on_subnet(
884        &self,
885        sender: Option<Principal>,
886        settings: Option<CanisterSettings>,
887        subnet_id: SubnetId,
888    ) -> Result<CanisterId, ClientError> {
889        self.create_canister_with_params(
890            sender,
891            CreateCanisterParams {
892                settings,
893                placement: Some(CreateCanisterPlacement::SubnetId(subnet_id)),
894                ..Default::default()
895            },
896        )
897        .await
898    }
899
900    /// Create a canister with optional cycles, settings, and placement.
901    /// The placement specifies either a target subnet or a specific canister ID.
902    /// Defaults to 100T cycles if `params.cycles` is `None`.
903    /// Returns an error if the specified canister ID is already in use.
904    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
905    pub async fn create_canister_with_params(
906        &self,
907        sender: Option<Principal>,
908        params: CreateCanisterParams,
909    ) -> Result<CanisterId, ClientError> {
910        let cycles = params.cycles.unwrap_or(100_000_000_000_000);
911        let (effective_principal, specified_id) = match params.placement {
912            None => (RawEffectivePrincipal::None, None),
913            Some(CreateCanisterPlacement::SubnetId(subnet_id)) => (
914                RawEffectivePrincipal::SubnetId(subnet_id.as_slice().to_vec()),
915                None,
916            ),
917            Some(CreateCanisterPlacement::CanisterId(canister_id)) => (
918                RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
919                Some(canister_id),
920            ),
921        };
922        Ok(call_candid_as::<_, (CanisterIdRecord,)>(
923            self,
924            Principal::management_canister(),
925            effective_principal,
926            sender.unwrap_or(Principal::anonymous()),
927            "provisional_create_canister_with_cycles",
928            (ProvisionalCreateCanisterWithCyclesArgs {
929                settings: params.settings,
930                amount: Some(cycles.into()),
931                specified_id,
932                sender_canister_version: None,
933            },),
934        )
935        .await?
936        .0
937        .canister_id)
938    }
939
940    /// Upload a WASM chunk to the WASM chunk store of a canister.
941    /// Returns the WASM chunk hash.
942    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
943    pub async fn upload_chunk(
944        &self,
945        canister_id: CanisterId,
946        sender: Option<Principal>,
947        chunk: Vec<u8>,
948    ) -> Result<Vec<u8>, ClientError> {
949        Ok(call_candid_as::<_, (UploadChunkResult,)>(
950            self,
951            Principal::management_canister(),
952            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
953            sender.unwrap_or(Principal::anonymous()),
954            "upload_chunk",
955            (UploadChunkArgs { canister_id, chunk },),
956        )
957        .await?
958        .0
959        .hash)
960    }
961
962    /// List WASM chunk hashes in the WASM chunk store of a canister.
963    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
964    pub async fn stored_chunks(
965        &self,
966        canister_id: CanisterId,
967        sender: Option<Principal>,
968    ) -> Result<Vec<Vec<u8>>, ClientError> {
969        Ok(call_candid_as::<_, (StoredChunksResult,)>(
970            self,
971            Principal::management_canister(),
972            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
973            sender.unwrap_or(Principal::anonymous()),
974            "stored_chunks",
975            (CanisterIdRecord { canister_id },),
976        )
977        .await?
978        .0
979        .into_iter()
980        .map(|chunk| chunk.hash)
981        .collect())
982    }
983
984    /// Clear the WASM chunk store of a canister.
985    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
986    pub async fn clear_chunk_store(
987        &self,
988        canister_id: CanisterId,
989        sender: Option<Principal>,
990    ) -> Result<(), ClientError> {
991        call_candid_as(
992            self,
993            Principal::management_canister(),
994            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
995            sender.unwrap_or(Principal::anonymous()),
996            "clear_chunk_store",
997            (CanisterIdRecord { canister_id },),
998        )
999        .await
1000    }
1001
1002    /// Install a WASM module assembled from chunks on an existing canister.
1003    #[instrument(skip(self, mode, chunk_hashes_list, wasm_module_hash, arg), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string(), store_canister_id = %store_canister_id.to_string(), arg_len = %arg.len()))]
1004    pub async fn install_chunked_canister(
1005        &self,
1006        canister_id: CanisterId,
1007        sender: Option<Principal>,
1008        mode: CanisterInstallMode,
1009        store_canister_id: CanisterId,
1010        chunk_hashes_list: Vec<Vec<u8>>,
1011        wasm_module_hash: Vec<u8>,
1012        arg: Vec<u8>,
1013    ) -> Result<(), ClientError> {
1014        call_candid_as(
1015            self,
1016            Principal::management_canister(),
1017            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1018            sender.unwrap_or(Principal::anonymous()),
1019            "install_chunked_code",
1020            (InstallChunkedCodeArgs {
1021                mode,
1022                target_canister: canister_id,
1023                store_canister: Some(store_canister_id),
1024                chunk_hashes_list: chunk_hashes_list
1025                    .into_iter()
1026                    .map(|hash| ChunkHash { hash })
1027                    .collect(),
1028                wasm_module_hash,
1029                arg,
1030                sender_canister_version: None,
1031            },),
1032        )
1033        .await
1034    }
1035
1036    async fn install_canister_helper(
1037        &self,
1038        mode: CanisterInstallMode,
1039        canister_id: CanisterId,
1040        wasm_module: Vec<u8>,
1041        arg: Vec<u8>,
1042        sender: Option<Principal>,
1043    ) -> Result<(), ClientError> {
1044        if wasm_module.len() + arg.len() < INSTALL_CHUNKED_CODE_THRESHOLD {
1045            call_candid_as::<(InstallCodeArgs,), ()>(
1046                self,
1047                Principal::management_canister(),
1048                RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1049                sender.unwrap_or(Principal::anonymous()),
1050                "install_code",
1051                (InstallCodeArgs {
1052                    mode,
1053                    canister_id,
1054                    wasm_module,
1055                    arg,
1056                    sender_canister_version: None,
1057                },),
1058            )
1059            .await?
1060        } else {
1061            self.clear_chunk_store(canister_id, sender).await?;
1062            let chunks: Vec<_> = wasm_module.chunks(INSTALL_CODE_CHUNK_SIZE).collect();
1063            let mut hashes = vec![];
1064            for chunk in chunks {
1065                let hash = self
1066                    .upload_chunk(canister_id, sender, chunk.to_vec())
1067                    .await?;
1068                hashes.push(hash);
1069            }
1070            let mut hasher = Sha256::new();
1071            hasher.update(wasm_module);
1072            let wasm_module_hash = hasher.finalize().to_vec();
1073            self.install_chunked_canister(
1074                canister_id,
1075                sender,
1076                mode,
1077                canister_id,
1078                hashes,
1079                wasm_module_hash,
1080                arg,
1081            )
1082            .await?
1083        };
1084        Ok(())
1085    }
1086
1087    /// Install a WASM module on an existing canister.
1088    #[instrument(skip(self, wasm_module, arg), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), wasm_module_len = %wasm_module.len(), arg_len = %arg.len(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1089    pub async fn install_canister(
1090        &self,
1091        canister_id: CanisterId,
1092        wasm_module: Vec<u8>,
1093        arg: Vec<u8>,
1094        sender: Option<Principal>,
1095    ) -> Result<(), ClientError> {
1096        self.install_canister_helper(
1097            CanisterInstallMode::Install,
1098            canister_id,
1099            wasm_module,
1100            arg,
1101            sender,
1102        )
1103        .await
1104    }
1105
1106    /// Upgrade a canister with a new WASM module.
1107    #[instrument(skip(self, wasm_module, arg), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), wasm_module_len = %wasm_module.len(), arg_len = %arg.len(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1108    pub async fn upgrade_canister(
1109        &self,
1110        canister_id: CanisterId,
1111        wasm_module: Vec<u8>,
1112        arg: Vec<u8>,
1113        sender: Option<Principal>,
1114    ) -> Result<(), ClientError> {
1115        self.install_canister_helper(
1116            CanisterInstallMode::Upgrade(None),
1117            canister_id,
1118            wasm_module,
1119            arg,
1120            sender,
1121        )
1122        .await
1123    }
1124
1125    /// Reinstall a canister WASM module.
1126    #[instrument(skip(self, wasm_module, arg), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), wasm_module_len = %wasm_module.len(), arg_len = %arg.len(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1127    pub async fn reinstall_canister(
1128        &self,
1129        canister_id: CanisterId,
1130        wasm_module: Vec<u8>,
1131        arg: Vec<u8>,
1132        sender: Option<Principal>,
1133    ) -> Result<(), ClientError> {
1134        self.install_canister_helper(
1135            CanisterInstallMode::Reinstall,
1136            canister_id,
1137            wasm_module,
1138            arg,
1139            sender,
1140        )
1141        .await
1142    }
1143
1144    /// Uninstall a canister.
1145    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1146    pub async fn uninstall_canister(
1147        &self,
1148        canister_id: CanisterId,
1149        sender: Option<Principal>,
1150    ) -> Result<(), ClientError> {
1151        call_candid_as::<(CanisterIdRecord,), ()>(
1152            self,
1153            Principal::management_canister(),
1154            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1155            sender.unwrap_or(Principal::anonymous()),
1156            "uninstall_code",
1157            (CanisterIdRecord { canister_id },),
1158        )
1159        .await
1160    }
1161
1162    /// Take canister snapshot.
1163    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1164    pub async fn take_canister_snapshot(
1165        &self,
1166        canister_id: CanisterId,
1167        sender: Option<Principal>,
1168        replace_snapshot: Option<Vec<u8>>,
1169    ) -> Result<Snapshot, ClientError> {
1170        Ok(call_candid_as::<_, (Snapshot,)>(
1171            self,
1172            Principal::management_canister(),
1173            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1174            sender.unwrap_or(Principal::anonymous()),
1175            "take_canister_snapshot",
1176            (TakeCanisterSnapshotArgs {
1177                canister_id,
1178                replace_snapshot,
1179                uninstall_code: None,
1180                sender_canister_version: None,
1181            },),
1182        )
1183        .await?
1184        .0)
1185    }
1186
1187    /// Load canister snapshot.
1188    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1189    pub async fn load_canister_snapshot(
1190        &self,
1191        canister_id: CanisterId,
1192        sender: Option<Principal>,
1193        snapshot_id: Vec<u8>,
1194    ) -> Result<(), ClientError> {
1195        call_candid_as(
1196            self,
1197            Principal::management_canister(),
1198            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1199            sender.unwrap_or(Principal::anonymous()),
1200            "load_canister_snapshot",
1201            (LoadCanisterSnapshotArgs {
1202                canister_id,
1203                snapshot_id,
1204                sender_canister_version: None,
1205            },),
1206        )
1207        .await
1208    }
1209
1210    /// List canister snapshots.
1211    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1212    pub async fn list_canister_snapshots(
1213        &self,
1214        canister_id: CanisterId,
1215        sender: Option<Principal>,
1216    ) -> Result<Vec<Snapshot>, ClientError> {
1217        Ok(call_candid_as::<_, (Vec<Snapshot>,)>(
1218            self,
1219            Principal::management_canister(),
1220            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1221            sender.unwrap_or(Principal::anonymous()),
1222            "list_canister_snapshots",
1223            (CanisterIdRecord { canister_id },),
1224        )
1225        .await?
1226        .0)
1227    }
1228
1229    /// Delete canister snapshot.
1230    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1231    pub async fn delete_canister_snapshot(
1232        &self,
1233        canister_id: CanisterId,
1234        sender: Option<Principal>,
1235        snapshot_id: Vec<u8>,
1236    ) -> Result<(), ClientError> {
1237        call_candid_as(
1238            self,
1239            Principal::management_canister(),
1240            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1241            sender.unwrap_or(Principal::anonymous()),
1242            "delete_canister_snapshot",
1243            (DeleteCanisterSnapshotArgs {
1244                canister_id,
1245                snapshot_id,
1246            },),
1247        )
1248        .await
1249    }
1250
1251    /// Update canister settings.
1252    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1253    pub async fn update_canister_settings(
1254        &self,
1255        canister_id: CanisterId,
1256        sender: Option<Principal>,
1257        settings: CanisterSettings,
1258    ) -> Result<(), ClientError> {
1259        call_candid_as::<_, ()>(
1260            self,
1261            Principal::management_canister(),
1262            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1263            sender.unwrap_or(Principal::anonymous()),
1264            "update_settings",
1265            (UpdateSettingsArgs {
1266                canister_id,
1267                settings,
1268                sender_canister_version: None,
1269            },),
1270        )
1271        .await
1272    }
1273
1274    /// Set canister's controllers.
1275    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1276    pub async fn set_controllers(
1277        &self,
1278        canister_id: CanisterId,
1279        sender: Option<Principal>,
1280        new_controllers: Vec<Principal>,
1281    ) -> Result<(), ClientError> {
1282        let settings = CanisterSettings {
1283            controllers: Some(new_controllers),
1284            ..CanisterSettings::default()
1285        };
1286        call_candid_as::<(UpdateSettingsArgs,), ()>(
1287            self,
1288            Principal::management_canister(),
1289            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1290            sender.unwrap_or(Principal::anonymous()),
1291            "update_settings",
1292            (UpdateSettingsArgs {
1293                canister_id,
1294                settings,
1295                sender_canister_version: None,
1296            },),
1297        )
1298        .await
1299    }
1300
1301    /// Start a canister.
1302    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1303    pub async fn start_canister(
1304        &self,
1305        canister_id: CanisterId,
1306        sender: Option<Principal>,
1307    ) -> Result<(), ClientError> {
1308        call_candid_as::<(CanisterIdRecord,), ()>(
1309            self,
1310            Principal::management_canister(),
1311            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1312            sender.unwrap_or(Principal::anonymous()),
1313            "start_canister",
1314            (CanisterIdRecord { canister_id },),
1315        )
1316        .await
1317    }
1318
1319    /// Stop a canister.
1320    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1321    pub async fn stop_canister(
1322        &self,
1323        canister_id: CanisterId,
1324        sender: Option<Principal>,
1325    ) -> Result<(), ClientError> {
1326        call_candid_as::<(CanisterIdRecord,), ()>(
1327            self,
1328            Principal::management_canister(),
1329            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1330            sender.unwrap_or(Principal::anonymous()),
1331            "stop_canister",
1332            (CanisterIdRecord { canister_id },),
1333        )
1334        .await
1335    }
1336
1337    /// Delete a canister.
1338    #[instrument(skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string(), sender = %sender.unwrap_or(Principal::anonymous()).to_string()))]
1339    pub async fn delete_canister(
1340        &self,
1341        canister_id: CanisterId,
1342        sender: Option<Principal>,
1343    ) -> Result<(), ClientError> {
1344        call_candid_as::<(CanisterIdRecord,), ()>(
1345            self,
1346            Principal::management_canister(),
1347            RawEffectivePrincipal::CanisterId(canister_id.as_slice().to_vec()),
1348            sender.unwrap_or(Principal::anonymous()),
1349            "delete_canister",
1350            (CanisterIdRecord { canister_id },),
1351        )
1352        .await
1353    }
1354
1355    /// Checks whether the provided canister exists.
1356    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string()))]
1357    pub async fn canister_exists(&self, canister_id: CanisterId) -> Result<bool, ClientError> {
1358        Ok(self.get_subnet(canister_id).await?.is_some())
1359    }
1360
1361    /// Deletes a subnet. Returns a server error if the subnet does not exist or is a named subnet.
1362    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, subnet_id = %subnet_id.to_string()))]
1363    pub async fn delete_subnet(&self, subnet_id: SubnetId) -> Result<(), ClientError> {
1364        let endpoint = "update/delete_subnet";
1365        self.post::<(), RawSubnetId>(
1366            endpoint,
1367            RawSubnetId {
1368                subnet_id: subnet_id.as_slice().to_vec(),
1369            },
1370        )
1371        .await?;
1372
1373        Ok(())
1374    }
1375
1376    /// Returns the subnet ID of the canister if the canister exists.
1377    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, canister_id = %canister_id.to_string()))]
1378    pub async fn get_subnet(
1379        &self,
1380        canister_id: CanisterId,
1381    ) -> Result<Option<SubnetId>, ClientError> {
1382        let endpoint = "read/get_subnet";
1383        let result: Option<RawSubnetId> = self
1384            .post(
1385                endpoint,
1386                RawCanisterId {
1387                    canister_id: canister_id.as_slice().to_vec(),
1388                },
1389            )
1390            .await?;
1391        Ok(result.map(|RawSubnetId { subnet_id }| SubnetId::from_slice(&subnet_id)))
1392    }
1393
1394    /// Returns subnet metrics for a given subnet.
1395    #[instrument(ret, skip(self), fields(instance_id=self.instance_id, subnet_id = %subnet_id.to_string()))]
1396    pub async fn get_subnet_metrics(
1397        &self,
1398        subnet_id: Principal,
1399    ) -> Result<Option<SubnetMetrics>, ClientError> {
1400        let path = vec![
1401            "subnet".into(),
1402            Label::from_bytes(subnet_id.as_slice()),
1403            "metrics".into(),
1404        ];
1405        let paths = vec![path.clone()];
1406        let content = ReadState {
1407            ingress_expiry: self.get_time().await?.as_nanos_since_unix_epoch() + 240_000_000_000,
1408            sender: Principal::anonymous(),
1409            paths,
1410        };
1411        let envelope = Envelope {
1412            content: std::borrow::Cow::Borrowed(&content),
1413            sender_pubkey: None,
1414            sender_sig: None,
1415            sender_delegation: None,
1416        };
1417
1418        let serialized_bytes = encode_envelope(&envelope, usize::MAX)?;
1419
1420        let endpoint = format!("api/v2/subnet/{}/read_state", subnet_id.to_text());
1421        let url = self.instance_url().join(&endpoint).unwrap();
1422        let mut exchange = Exchange::new(&url, self.max_request_time_ms);
1423        let (status, bytes) = exchange
1424            .bytes(
1425                &self.reqwest_client,
1426                self.reqwest_client
1427                    .post(url.clone())
1428                    .header(reqwest::header::CONTENT_TYPE, "application/cbor")
1429                    .body(serialized_bytes),
1430            )
1431            .await?;
1432        if !status.is_success() {
1433            return Err(exchange.server_error(status, &bytes));
1434        }
1435        let read_state_response: ReadStateResponse =
1436            decode_read_state_response(&bytes, bytes.len())?;
1437        let cert: Certificate = decode_certificate(
1438            &read_state_response.certificate,
1439            read_state_response.certificate.len(),
1440        )?;
1441
1442        let metrics = match cert.tree.lookup_path(path) {
1443            LookupResult::Found(value) => value,
1444            _ => return Ok(None),
1445        };
1446        decode_subnet_metrics(metrics, metrics.len()).map_err(ClientError::from)
1447    }
1448
1449    /// This (asynchronous) drop function must be called to drop the PocketIc instance.
1450    /// It must be called manually as Rust doesn't support asynchronous drop.
1451    pub async fn drop(mut self) -> Result<(), ClientError> {
1452        self.do_drop().await?;
1453
1454        Ok(())
1455    }
1456
1457    pub(crate) async fn do_drop(&mut self) -> Result<(), ClientError> {
1458        if self.owns_instance {
1459            let url = self.instance_url();
1460            let mut exchange = Exchange::new(&url, self.max_request_time_ms);
1461            let (status, bytes) = exchange
1462                .bytes(
1463                    &self.reqwest_client,
1464                    self.reqwest_client.delete(url.clone()),
1465                )
1466                .await?;
1467            if !status.is_success() {
1468                return Err(exchange.server_error(status, &bytes));
1469            }
1470            self.owns_instance = false;
1471            Ok(())
1472        } else {
1473            self.stop_http_gateway().await
1474        }
1475    }
1476
1477    async fn get<T: DeserializeOwned>(&self, endpoint: &str) -> Result<T, ClientError> {
1478        self.request(HttpMethod::Get, endpoint, ()).await
1479    }
1480
1481    async fn post<T: DeserializeOwned, B: Serialize>(
1482        &self,
1483        endpoint: &str,
1484        body: B,
1485    ) -> Result<T, ClientError> {
1486        self.request(HttpMethod::Post, endpoint, body).await
1487    }
1488
1489    async fn request<T: DeserializeOwned, B: Serialize>(
1490        &self,
1491        method: HttpMethod,
1492        endpoint: &str,
1493        body: B,
1494    ) -> Result<T, ClientError> {
1495        self.request_limited(method, endpoint, body, None).await
1496    }
1497
1498    async fn request_limited<T: DeserializeOwned, B: Serialize>(
1499        &self,
1500        method: HttpMethod,
1501        endpoint: &str,
1502        body: B,
1503        response_limit: Option<usize>,
1504    ) -> Result<T, ClientError> {
1505        let url = self.instance_url().join(endpoint).unwrap();
1506        let mut exchange = Exchange::new(&url, self.max_request_time_ms);
1507        exchange.response_limit = response_limit;
1508        loop {
1509            let builder = if let Some(operation) = &exchange.context.operation {
1510                self.reqwest_client.get(
1511                    self.server_url
1512                        .join(&format!(
1513                            "/read_graph/{}/{}",
1514                            operation.state_label, operation.op_id
1515                        ))
1516                        .unwrap(),
1517                )
1518            } else {
1519                match method {
1520                    HttpMethod::Get => self.reqwest_client.get(url.clone()),
1521                    HttpMethod::Post => self.reqwest_client.post(url.clone()).json(&body),
1522                }
1523            };
1524            let (status, bytes) = exchange.bytes(&self.reqwest_client, builder).await?;
1525            match status {
1526                StatusCode::OK => {
1527                    let value = exchange.decode(&bytes)?;
1528                    if let Some(operation) = &exchange.context.operation {
1529                        // The result is known; pruning is best-effort housekeeping, never a resubmission.
1530                        let request = self.reqwest_client.delete(
1531                            self.server_url
1532                                .join(&format!(
1533                                    "/prune_graph/{}/{}",
1534                                    operation.state_label, operation.op_id
1535                                ))
1536                                .unwrap(),
1537                        );
1538                        tokio::spawn(async move {
1539                            let _ = request.send().await;
1540                        });
1541                    }
1542                    return Ok(value);
1543                }
1544                StatusCode::ACCEPTED | StatusCode::CONFLICT => {
1545                    let receipt: crate::pocket_ic::common::rest::StartedOrBusyResponse =
1546                        exchange.decode(&bytes)?;
1547                    if exchange.context.operation.is_none() {
1548                        if status == StatusCode::ACCEPTED {
1549                            exchange.started(Operation {
1550                                state_label: receipt.state_label,
1551                                op_id: receipt.op_id,
1552                            });
1553                        } else {
1554                            // Busy explicitly says this submission was not accepted.
1555                            exchange.context.may_have_executed = false;
1556                        }
1557                    }
1558                    // Once accepted, always poll our original receipt, even if an
1559                    // unexpected response refers to a different computation.
1560                }
1561                StatusCode::NOT_FOUND if exchange.context.operation.is_some() => {}
1562                _ => return Err(exchange.server_error(status, &bytes)),
1563            }
1564            exchange.pause().await?;
1565        }
1566    }
1567
1568    /// Get the pending canister HTTP outcalls.
1569    /// Note that an additional `PocketIc::tick` is necessary after a canister
1570    /// executes a message making a canister HTTP outcall for the HTTP outcall
1571    /// to be retrievable here.
1572    /// Note that, unless a PocketIC instance is in auto progress mode,
1573    /// a response to the pending canister HTTP outcalls
1574    /// must be produced by the test driver and passed on to the PocketIC instace
1575    /// using `PocketIc::mock_canister_http_response`. Flexible outcall mocking is
1576    /// not supported by this client.
1577    /// In auto progress mode, the PocketIC server produces a response for every
1578    /// pending canister HTTP outcall by actually making an HTTP request
1579    /// to the specified URL.
1580    #[instrument(ret, skip(self), fields(instance_id=self.instance_id))]
1581    pub async fn get_canister_http(&self) -> Result<Vec<CanisterHttpRequest>, ClientError> {
1582        let endpoint = "read/get_canister_http";
1583        let res: Vec<RawCanisterHttpRequest> = self.get(endpoint).await?;
1584        Ok(res.into_iter().map(|r| r.into()).collect())
1585    }
1586
1587    /// Mock a response to a pending canister HTTP outcall: the same response for
1588    /// every node of the subnet, or one response per node if
1589    /// `MockCanisterHttpResponse::additional_responses` is non-empty.
1590    #[instrument(ret, skip(self), fields(instance_id=self.instance_id))]
1591    pub async fn mock_canister_http_response(
1592        &self,
1593        mock_canister_http_response: MockCanisterHttpResponse,
1594    ) -> Result<(), ClientError> {
1595        let endpoint = "update/mock_canister_http";
1596        let raw_mock_canister_http_response: RawMockCanisterHttpResponse =
1597            mock_canister_http_response.into();
1598        self.post(endpoint, raw_mock_canister_http_response).await
1599    }
1600
1601    /// Download a canister snapshot to a given snapshot directory.
1602    /// The sender must be a controller of the canister.
1603    /// The snapshot directory must be empty if it exists.
1604    #[instrument(ret, skip(self), fields(instance_id=self.instance_id))]
1605    pub async fn canister_snapshot_download(
1606        &self,
1607        canister_id: CanisterId,
1608        sender: Principal,
1609        snapshot_id: Vec<u8>,
1610        snapshot_dir: PathBuf,
1611    ) -> Result<(), ClientError> {
1612        let endpoint = "update/canister_snapshot_download";
1613        #[cfg(not(windows))]
1614        let snapshot_dir = snapshot_dir;
1615        #[cfg(windows)]
1616        let snapshot_dir = wsl_path(&snapshot_dir, "snapshot directory").into();
1617        let raw_canister_snapshot_download = RawCanisterSnapshotDownload {
1618            sender: sender.into(),
1619            canister_id: canister_id.into(),
1620            snapshot_id,
1621            snapshot_dir,
1622        };
1623        self.post(endpoint, raw_canister_snapshot_download).await
1624    }
1625
1626    /// Upload a canister snapshot from a given snapshot directory.
1627    /// The sender must be a controller of the canister.
1628    /// Returns the snapshot ID of the uploaded snapshot.
1629    #[instrument(ret, skip(self), fields(instance_id=self.instance_id))]
1630    pub async fn canister_snapshot_upload(
1631        &self,
1632        canister_id: CanisterId,
1633        sender: Principal,
1634        replace_snapshot: Option<Vec<u8>>,
1635        snapshot_dir: PathBuf,
1636    ) -> Result<Vec<u8>, ClientError> {
1637        let endpoint = "update/canister_snapshot_upload";
1638        let replace_snapshot =
1639            replace_snapshot.map(|snapshot_id| RawCanisterSnapshotId { snapshot_id });
1640        #[cfg(not(windows))]
1641        let snapshot_dir = snapshot_dir;
1642        #[cfg(windows)]
1643        let snapshot_dir = wsl_path(&snapshot_dir, "snapshot directory").into();
1644        let raw_canister_snapshot_upload = RawCanisterSnapshotUpload {
1645            sender: sender.into(),
1646            canister_id: canister_id.into(),
1647            replace_snapshot,
1648            snapshot_dir,
1649        };
1650        Ok(self
1651            .post::<RawCanisterSnapshotId, _>(endpoint, raw_canister_snapshot_upload)
1652            .await?
1653            .snapshot_id)
1654    }
1655}
1656
1657/// Call a canister candid method, authenticated. The sender can be impersonated (i.e., the
1658/// signature is not verified).
1659/// PocketIC executes update calls synchronously, so there is no need to poll for the result.
1660pub async fn call_candid_as<Input, Output>(
1661    env: &PocketIc,
1662    canister_id: CanisterId,
1663    effective_principal: RawEffectivePrincipal,
1664    sender: Principal,
1665    method: &str,
1666    input: Input,
1667) -> Result<Output, ClientError>
1668where
1669    Input: ArgumentEncoder,
1670    Output: for<'a> ArgumentDecoder<'a>,
1671{
1672    with_candid(input, |payload| async {
1673        env.update_call_with_effective_principal(
1674            canister_id,
1675            effective_principal,
1676            sender,
1677            method,
1678            payload,
1679        )
1680        .await
1681    })
1682    .await
1683}
1684
1685/// Call a canister candid method, anonymous.
1686/// PocketIC executes update calls synchronously, so there is no need to poll for the result.
1687pub async fn call_candid<Input, Output>(
1688    env: &PocketIc,
1689    canister_id: CanisterId,
1690    effective_principal: RawEffectivePrincipal,
1691    method: &str,
1692    input: Input,
1693) -> Result<Output, ClientError>
1694where
1695    Input: ArgumentEncoder,
1696    Output: for<'a> ArgumentDecoder<'a>,
1697{
1698    call_candid_as(
1699        env,
1700        canister_id,
1701        effective_principal,
1702        Principal::anonymous(),
1703        method,
1704        input,
1705    )
1706    .await
1707}
1708
1709/// Call a canister candid query method, anonymous.
1710pub async fn query_candid<Input, Output>(
1711    env: &PocketIc,
1712    canister_id: CanisterId,
1713    method: &str,
1714    input: Input,
1715) -> Result<Output, ClientError>
1716where
1717    Input: ArgumentEncoder,
1718    Output: for<'a> ArgumentDecoder<'a>,
1719{
1720    query_candid_as(env, canister_id, Principal::anonymous(), method, input).await
1721}
1722
1723/// Call a canister candid query method, authenticated. The sender can be impersonated (i.e., the
1724/// signature is not verified).
1725pub async fn query_candid_as<Input, Output>(
1726    env: &PocketIc,
1727    canister_id: CanisterId,
1728    sender: Principal,
1729    method: &str,
1730    input: Input,
1731) -> Result<Output, ClientError>
1732where
1733    Input: ArgumentEncoder,
1734    Output: for<'a> ArgumentDecoder<'a>,
1735{
1736    with_candid(input, |bytes| async {
1737        env.query_call(canister_id, sender, method, bytes).await
1738    })
1739    .await
1740}
1741
1742/// Call a canister candid update method, anonymous.
1743pub async fn update_candid<Input, Output>(
1744    env: &PocketIc,
1745    canister_id: CanisterId,
1746    method: &str,
1747    input: Input,
1748) -> Result<Output, ClientError>
1749where
1750    Input: ArgumentEncoder,
1751    Output: for<'a> ArgumentDecoder<'a>,
1752{
1753    update_candid_as(env, canister_id, Principal::anonymous(), method, input).await
1754}
1755
1756/// Call a canister candid update method, authenticated. The sender can be impersonated (i.e., the
1757/// signature is not verified).
1758pub async fn update_candid_as<Input, Output>(
1759    env: &PocketIc,
1760    canister_id: CanisterId,
1761    sender: Principal,
1762    method: &str,
1763    input: Input,
1764) -> Result<Output, ClientError>
1765where
1766    Input: ArgumentEncoder,
1767    Output: for<'a> ArgumentDecoder<'a>,
1768{
1769    with_candid(input, |bytes| async {
1770        env.update_call(canister_id, sender, method, bytes).await
1771    })
1772    .await
1773}
1774
1775/// A helper function that we use to implement both [`call_candid`] and
1776/// [`query_candid`].
1777pub async fn with_candid<Input, Output, Fut>(
1778    input: Input,
1779    f: impl FnOnce(Vec<u8>) -> Fut,
1780) -> Result<Output, ClientError>
1781where
1782    Input: ArgumentEncoder,
1783    Output: for<'a> ArgumentDecoder<'a>,
1784    Fut: Future<Output = Result<Vec<u8>, ClientError>>,
1785{
1786    let input = encode_args(input)?;
1787    let output = f(input).await?;
1788    Ok(decode_args(&output)?)
1789}
1790
1791/// Retrieves a default effective canister id for canister creation on a PocketIC instance
1792/// characterized by:
1793///  - a PocketIC instance URL of the form `http://<ip>:<port>/instances/<instance_id>`;
1794///  - a PocketIC HTTP gateway URL of the form `http://<ip>:port` for a PocketIC instance.
1795///
1796/// Returns an error if the PocketIC instance topology could not be fetched or parsed, e.g.,
1797/// because the given URL points to a replica (i.e., does not meet any of the above two properties).
1798pub async fn get_default_effective_canister_id(
1799    pocket_ic_url: String,
1800) -> Result<Principal, ClientError> {
1801    let url = Url::parse(&format!(
1802        "{}/_/topology",
1803        pocket_ic_url.trim_end_matches('/')
1804    ))
1805    .map_err(|e| ClientError::Protocol(e.to_string()))?;
1806    let client = reqwest::Client::new();
1807    let mut exchange = Exchange::new(&url, Some(crate::pocket_ic::DEFAULT_MAX_REQUEST_TIME_MS));
1808    loop {
1809        let (status, bytes) = exchange.bytes(&client, client.get(url.clone())).await?;
1810        if status == StatusCode::CONFLICT {
1811            exchange.pause().await?;
1812            continue;
1813        }
1814        if !status.is_success() {
1815            return Err(exchange.server_error(status, &bytes));
1816        }
1817        let topology: Topology = exchange.decode(&bytes)?;
1818        return Principal::try_from_slice(&topology.default_effective_canister_id.canister_id)
1819            .map_err(|e| ClientError::Protocol(e.to_string()));
1820    }
1821}
ic_testkit/pocket_ic/nonblocking.rs

ic_testkit/pocket_ic/
nonblocking.rs
