harn_vm/connectors/
testkit.rs

1use std::collections::BTreeMap;
2use std::fs;
3use std::io;
4use std::path::{Path, PathBuf};
5use std::sync::atomic::{AtomicBool, Ordering};
6use std::sync::Arc;
7use std::time::Duration as StdDuration;
8
9use serde_json::{json, Value as JsonValue};
10use time::OffsetDateTime;
11use tokio::sync::mpsc;
12use uuid::Uuid;
13
14pub use crate::http::framing::{
15    http_content_length_from_header_lines, HttpContentLengthLimitError, TEST_HTTP_MAX_BODY_BYTES,
16};
17pub use crate::http::{HttpMockCallSnapshot, HttpMockResponse};
18pub use crate::triggers::test_util::clock::{
19    active_mock_clock, install_override as install_clock_override, instant_now, now_ms, now_utc,
20    ClockInstant, ClockOverrideGuard, MockClock,
21};
22
23use crate::connectors::{
24    ConnectorCtx, MetricsRegistry, RateLimiterFactory, RawInbound, TriggerBinding,
25};
26use crate::event_log::{AnyEventLog, MemoryEventLog};
27pub use crate::secrets::MemorySecretProvider;
28use crate::secrets::SecretId;
29use crate::triggers::{InboxIndex, ProviderId, TenantId};
30
31impl MemorySecretProvider {
32    pub fn with_scoped_secret(
33        self,
34        namespace: impl Into<String>,
35        tenant_id: impl AsRef<str>,
36        binding_id: impl AsRef<str>,
37        name: impl AsRef<str>,
38        value: impl AsRef<[u8]>,
39    ) -> Self {
40        let id = scoped_secret_id(namespace, tenant_id, binding_id, name);
41        self.with_secret(id, value)
42    }
43
44    pub fn insert_scoped(
45        &mut self,
46        namespace: impl Into<String>,
47        tenant_id: impl AsRef<str>,
48        binding_id: impl AsRef<str>,
49        name: impl AsRef<str>,
50        value: impl AsRef<[u8]>,
51    ) -> SecretId {
52        let id = scoped_secret_id(namespace, tenant_id, binding_id, name);
53        self.insert(id.clone(), value);
54        id
55    }
56}
57
58pub fn scoped_secret_id(
59    namespace: impl Into<String>,
60    tenant_id: impl AsRef<str>,
61    binding_id: impl AsRef<str>,
62    name: impl AsRef<str>,
63) -> SecretId {
64    SecretId::new(
65        namespace,
66        format!(
67            "tenants/{}/bindings/{}/{}",
68            tenant_id.as_ref(),
69            binding_id.as_ref(),
70            name.as_ref()
71        ),
72    )
73}
74
75#[derive(Clone)]
76pub struct ConnectorTestkit {
77    pub clock: Arc<MockClock>,
78    pub event_log: Arc<AnyEventLog>,
79    pub inbox: Arc<InboxIndex>,
80    pub metrics: Arc<MetricsRegistry>,
81    pub rate_limiter: Arc<RateLimiterFactory>,
82    pub secrets: Arc<MemorySecretProvider>,
83}
84
85impl ConnectorTestkit {
86    pub async fn new(start: OffsetDateTime) -> Self {
87        Self::with_secrets(start, MemorySecretProvider::empty()).await
88    }
89
90    pub async fn with_secrets(start: OffsetDateTime, secrets: MemorySecretProvider) -> Self {
91        let clock = MockClock::new(start);
92        let event_log = Arc::new(AnyEventLog::Memory(MemoryEventLog::new(64)));
93        let metrics = Arc::new(MetricsRegistry::default());
94        let inbox = Arc::new(
95            InboxIndex::new(event_log.clone(), metrics.clone())
96                .await
97                .expect("connector testkit inbox should initialize"),
98        );
99        Self {
100            clock,
101            event_log,
102            inbox,
103            metrics,
104            rate_limiter: Arc::new(RateLimiterFactory::default()),
105            secrets: Arc::new(secrets),
106        }
107    }
108
109    pub fn ctx(&self) -> ConnectorCtx {
110        ConnectorCtx {
111            event_log: self.event_log.clone(),
112            secrets: self.secrets.clone(),
113            inbox: self.inbox.clone(),
114            metrics: self.metrics.clone(),
115            rate_limiter: self.rate_limiter.clone(),
116        }
117    }
118
119    pub fn install_clock(&self) -> ClockOverrideGuard {
120        install_clock_override(self.clock.clone())
121    }
122}
123
124#[derive(Debug)]
125pub struct TempPackageWorkspace {
126    root: PathBuf,
127}
128
129impl TempPackageWorkspace {
130    pub fn new(prefix: impl AsRef<str>) -> io::Result<Self> {
131        let root = std::env::temp_dir().join(format!(
132            "{}-{}",
133            prefix.as_ref().trim_matches('-'),
134            Uuid::new_v4()
135        ));
136        fs::create_dir_all(&root)?;
137        Ok(Self { root })
138    }
139
140    pub fn path(&self) -> &Path {
141        &self.root
142    }
143
144    pub fn write_file(
145        &self,
146        relative: impl AsRef<Path>,
147        contents: impl AsRef<[u8]>,
148    ) -> io::Result<PathBuf> {
149        let path = self.root.join(relative.as_ref());
150        if let Some(parent) = path.parent() {
151            fs::create_dir_all(parent)?;
152        }
153        fs::write(&path, contents)?;
154        Ok(path)
155    }
156
157    pub fn write_harn_package(&self, name: &str) -> io::Result<PathBuf> {
158        self.write_file(
159            "Harn.toml",
160            format!("[package]\nname = \"{name}\"\nversion = \"0.0.0-test\"\n"),
161        )
162    }
163
164    pub fn write_cargo_package(&self, name: &str) -> io::Result<PathBuf> {
165        self.write_file(
166            "Cargo.toml",
167            format!("[package]\nname = \"{name}\"\nversion = \"0.0.0\"\nedition = \"2021\"\n"),
168        )
169    }
170
171    pub fn write_npm_package(&self, name: &str) -> io::Result<PathBuf> {
172        self.write_file(
173            "package.json",
174            format!("{{\"name\":\"{name}\",\"version\":\"0.0.0-test\"}}\n"),
175        )
176    }
177}
178
179impl Drop for TempPackageWorkspace {
180    fn drop(&mut self) {
181        let _ = fs::remove_dir_all(&self.root);
182    }
183}
184
185pub struct HttpMockGuard;
186
187impl HttpMockGuard {
188    pub fn new() -> Self {
189        crate::http::reset_http_state();
190        Self
191    }
192
193    pub fn push(
194        &self,
195        method: impl Into<String>,
196        url_pattern: impl Into<String>,
197        responses: Vec<HttpMockResponse>,
198    ) {
199        crate::http::push_http_mock(method, url_pattern, responses);
200    }
201
202    pub fn calls(&self) -> Vec<HttpMockCallSnapshot> {
203        crate::http::http_mock_calls_snapshot()
204    }
205}
206
207impl Default for HttpMockGuard {
208    fn default() -> Self {
209        Self::new()
210    }
211}
212
213impl Drop for HttpMockGuard {
214    fn drop(&mut self) {
215        crate::http::reset_http_state();
216    }
217}
218
219#[derive(Clone, Debug, PartialEq, Eq)]
220pub enum MockStreamEvent {
221    Json(JsonValue),
222    Bytes(Vec<u8>),
223    Cancelled,
224}
225
226#[derive(Clone, Debug)]
227pub struct MockStreamHandle {
228    tx: mpsc::UnboundedSender<MockStreamEvent>,
229    cancelled: Arc<AtomicBool>,
230}
231
232#[derive(Debug)]
233pub struct MockStreamReader {
234    rx: mpsc::UnboundedReceiver<MockStreamEvent>,
235    cancelled: Arc<AtomicBool>,
236}
237
238pub fn mock_stream() -> (MockStreamHandle, MockStreamReader) {
239    let (tx, rx) = mpsc::unbounded_channel();
240    let cancelled = Arc::new(AtomicBool::new(false));
241    (
242        MockStreamHandle {
243            tx,
244            cancelled: cancelled.clone(),
245        },
246        MockStreamReader { rx, cancelled },
247    )
248}
249
250impl MockStreamHandle {
251    pub fn send_json(
252        &self,
253        value: JsonValue,
254    ) -> Result<(), mpsc::error::SendError<MockStreamEvent>> {
255        self.tx.send(MockStreamEvent::Json(value))
256    }
257
258    pub fn send_bytes(
259        &self,
260        value: impl Into<Vec<u8>>,
261    ) -> Result<(), mpsc::error::SendError<MockStreamEvent>> {
262        self.tx.send(MockStreamEvent::Bytes(value.into()))
263    }
264
265    pub fn cancel(&self) {
266        self.cancelled.store(true, Ordering::SeqCst);
267        let _ = self.tx.send(MockStreamEvent::Cancelled);
268    }
269
270    pub fn is_cancelled(&self) -> bool {
271        self.cancelled.load(Ordering::SeqCst)
272    }
273}
274
275impl MockStreamReader {
276    pub async fn next(&mut self) -> Option<MockStreamEvent> {
277        self.rx.recv().await
278    }
279
280    pub fn is_cancelled(&self) -> bool {
281        self.cancelled.load(Ordering::SeqCst)
282    }
283}
284
285// `RawInbound.metadata` is `serde_json::Value`, which isn't `Eq`.
286#[allow(clippy::derive_partial_eq_without_eq)]
287#[derive(Clone, Debug, PartialEq)]
288pub struct WebhookFixture {
289    pub raw: RawInbound,
290    pub body: Vec<u8>,
291}
292
293impl WebhookFixture {
294    pub fn with_binding(mut self, binding: &TriggerBinding) -> Self {
295        self.raw.metadata = json!({
296            "binding_id": binding.binding_id,
297            "binding_version": 1,
298        });
299        self
300    }
301
302    pub fn with_tenant(mut self, tenant_id: impl Into<String>) -> Self {
303        self.raw.tenant_id = Some(TenantId(tenant_id.into()));
304        self
305    }
306}
307
308pub fn github_ping_fixture(secret: &str, received_at: OffsetDateTime) -> WebhookFixture {
309    let body = br#"{"zen":"Keep it logically awesome.","hook_id":42}"#.to_vec();
310    let mut raw = RawInbound::new(
311        "webhook",
312        BTreeMap::from([
313            ("content-type".to_string(), "application/json".to_string()),
314            ("x-github-event".to_string(), "ping".to_string()),
315            ("x-github-delivery".to_string(), "delivery-1".to_string()),
316            (
317                "x-hub-signature-256".to_string(),
318                format!("sha256={}", hmac_sha256_hex(secret.as_bytes(), &body)),
319            ),
320        ]),
321        body.clone(),
322    );
323    raw.received_at = received_at;
324    WebhookFixture { raw, body }
325}
326
327pub fn slack_message_fixture(
328    secret: &str,
329    timestamp: i64,
330    received_at: OffsetDateTime,
331) -> WebhookFixture {
332    let body = br#"{"type":"event_callback","event_id":"Ev1","team_id":"T1","event":{"type":"message","channel_type":"channel","channel":"C1","user":"U1","text":"hello","event_ts":"1710000000.000100"}}"#.to_vec();
333    let signed = format!("v0:{timestamp}:{}", String::from_utf8_lossy(&body));
334    let mut raw = RawInbound::new(
335        "webhook",
336        BTreeMap::from([
337            ("content-type".to_string(), "application/json".to_string()),
338            (
339                "x-slack-request-timestamp".to_string(),
340                timestamp.to_string(),
341            ),
342            (
343                "x-slack-signature".to_string(),
344                format!(
345                    "v0={}",
346                    hmac_sha256_hex(secret.as_bytes(), signed.as_bytes())
347                ),
348            ),
349        ]),
350        body.clone(),
351    );
352    raw.received_at = received_at;
353    WebhookFixture { raw, body }
354}
355
356pub fn linear_issue_update_fixture(secret: &str, received_at: OffsetDateTime) -> WebhookFixture {
357    let body = br#"{"type":"Issue","action":"update","createdAt":"2026-04-19T00:00:00Z","data":{"id":"issue-1","identifier":"ENG-1","title":"connector"},"updatedFrom":{"title":"old"}}"#.to_vec();
358    let mut raw = RawInbound::new(
359        "webhook",
360        BTreeMap::from([
361            ("content-type".to_string(), "application/json".to_string()),
362            (
363                "linear-signature".to_string(),
364                hmac_sha256_hex(secret.as_bytes(), &body),
365            ),
366        ]),
367        body.clone(),
368    );
369    raw.received_at = received_at;
370    WebhookFixture { raw, body }
371}
372
373pub fn notion_page_content_updated_fixture(
374    secret: &str,
375    received_at: OffsetDateTime,
376) -> WebhookFixture {
377    let body = br#"{"id":"evt_1","type":"page.content_updated","workspace_id":"ws_1","subscription_id":"sub_1","integration_id":"int_1","entity":{"id":"page_1","type":"page"},"api_version":"2022-06-28"}"#.to_vec();
378    let mut raw = RawInbound::new(
379        "webhook",
380        BTreeMap::from([
381            ("content-type".to_string(), "application/json".to_string()),
382            (
383                "x-notion-signature".to_string(),
384                format!("sha256={}", hmac_sha256_hex(secret.as_bytes(), &body)),
385            ),
386            ("request-id".to_string(), "req_1".to_string()),
387        ]),
388        body.clone(),
389    );
390    raw.received_at = received_at;
391    WebhookFixture { raw, body }
392}
393
394pub fn webhook_binding(
395    provider: impl Into<String>,
396    binding_id: impl Into<String>,
397    signing_secret: Option<SecretId>,
398) -> TriggerBinding {
399    let provider = provider.into();
400    let mut binding =
401        TriggerBinding::new(ProviderId::from(provider.clone()), "webhook", binding_id);
402    let mut secrets = serde_json::Map::new();
403    if let Some(secret) = signing_secret {
404        secrets.insert(
405            "signing_secret".to_string(),
406            JsonValue::String(secret.to_string()),
407        );
408    }
409    binding.config = json!({
410        "path": format!("/hooks/{provider}"),
411        "match": {"events": ["*"]},
412        "secrets": secrets,
413    });
414    binding
415}
416
417fn hmac_sha256_hex(secret: &[u8], data: &[u8]) -> String {
418    hex::encode(crate::connectors::hmac::hmac_sha256(secret, data))
419}
420
421pub async fn advance_until<F>(
422    clock: &MockClock,
423    timeout: StdDuration,
424    tick: StdDuration,
425    mut predicate: F,
426) -> bool
427where
428    F: FnMut() -> bool,
429{
430    let mut elapsed = StdDuration::ZERO;
431    while elapsed <= timeout {
432        if predicate() {
433            return true;
434        }
435        clock.advance_std(tick).await;
436        elapsed += tick;
437    }
438    predicate()
439}
440
441#[cfg(test)]
442mod tests {
443    use super::*;
444    use crate::secrets::{SecretBytes, SecretProvider, SecretVersion};
445
446    fn parse_ts(value: &str) -> OffsetDateTime {
447        OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).unwrap()
448    }
449
450    #[tokio::test]
451    async fn memory_secret_provider_scopes_and_versions_secrets() {
452        let mut provider = MemorySecretProvider::new("test");
453        let scoped = provider.insert_scoped("github", "tenant-a", "binding-a", "token", "v1");
454        provider
455            .put(&scoped, SecretBytes::from("v2"))
456            .await
457            .expect("put latest");
458
459        let latest = provider.get(&scoped).await.expect("latest");
460        assert_eq!(latest.with_exposed(|bytes| bytes.to_vec()), b"v2".to_vec());
461        let first = provider
462            .get(&scoped.clone().with_version(SecretVersion::Exact(1)))
463            .await
464            .expect("v1");
465        assert_eq!(first.with_exposed(|bytes| bytes.to_vec()), b"v1".to_vec());
466        assert!(provider
467            .get(&scoped_secret_id(
468                "github",
469                "tenant-b",
470                "binding-a",
471                "token"
472            ))
473            .await
474            .is_err());
475    }
476
477    #[tokio::test]
478    async fn connector_testkit_controls_clock_and_deadlines() {
479        let kit = ConnectorTestkit::new(parse_ts("2026-04-19T00:00:00Z")).await;
480        let _guard = kit.install_clock();
481        let mut fired = false;
482        assert!(
483            !advance_until(
484                &kit.clock,
485                StdDuration::from_millis(20),
486                StdDuration::from_millis(10),
487                || fired,
488            )
489            .await
490        );
491        fired = true;
492        assert!(
493            advance_until(
494                &kit.clock,
495                StdDuration::from_millis(20),
496                StdDuration::from_millis(10),
497                || fired,
498            )
499            .await
500        );
501        assert_eq!(instant_now().as_millis(), 30);
502    }
503
504    #[tokio::test]
505    async fn mock_stream_cancels_reader_without_wall_clock_sleep() {
506        let (handle, mut reader) = mock_stream();
507        handle.send_json(json!({"event": "one"})).expect("send");
508        assert_eq!(
509            reader.next().await,
510            Some(MockStreamEvent::Json(json!({"event": "one"})))
511        );
512        handle.cancel();
513        assert_eq!(reader.next().await, Some(MockStreamEvent::Cancelled));
514        assert!(reader.is_cancelled());
515    }
516
517    #[test]
518    fn temp_workspace_writes_package_markers() {
519        let workspace = TempPackageWorkspace::new("harn-testkit").expect("workspace");
520        workspace.write_harn_package("demo").expect("harn package");
521        workspace
522            .write_cargo_package("demo")
523            .expect("cargo package");
524        workspace.write_npm_package("demo").expect("npm package");
525        assert!(workspace.path().join("Harn.toml").exists());
526        assert!(workspace.path().join("Cargo.toml").exists());
527        assert!(workspace.path().join("package.json").exists());
528    }
529
530    #[test]
531    fn webhook_fixtures_include_provider_signatures() {
532        let received_at = parse_ts("2026-04-19T00:00:00Z");
533        let github = github_ping_fixture("topsecret", received_at);
534        assert!(github.raw.headers["x-hub-signature-256"].starts_with("sha256="));
535        let slack = slack_message_fixture("topsecret", received_at.unix_timestamp(), received_at);
536        assert!(slack.raw.headers["x-slack-signature"].starts_with("v0="));
537        let linear = linear_issue_update_fixture("topsecret", received_at);
538        assert_eq!(linear.raw.headers["linear-signature"].len(), 64);
539        let notion = notion_page_content_updated_fixture("topsecret", received_at);
540        assert!(notion.raw.headers["x-notion-signature"].starts_with("sha256="));
541    }
542}
harn_vm/connectors/testkit.rs

harn_vm/connectors/
testkit.rs
