atagen

macltok

public
Star0
Watch0
96fdf90aacd6dc2b8bd3d63773234245b389d28c
3
ssh://brewery@pls.lobotomise.me:33/atagen/macltok.git
tests/media.rsRust943 lines28.3 KB
1
use std::{sync::Arc, time::Duration};
2
3
use macltok::{
4
    AttachmentUpload, ClientHandle, Command, CommandErrorKind, Engine, EngineConfig, EngineError,
5
    Event, EventStream, LifecycleCommand, MediaCommand, MediaId, MediaSize, RoomCommand,
6
    TimelineDiff, TimelineItem, UploadDetails, UploadThumbnail,
7
};
8
use matrix_sdk::{
9
    ruma::{
10
        api::client::sync::sync_events::v5,
11
        event_id,
12
        events::room::{
13
            MediaSource,
14
            message::{
15
                AudioMessageEventContent, FileInfo, FileMessageEventContent,
16
                ImageMessageEventContent, MessageType, RoomMessageEventContent,
17
            },
18
        },
19
        media::Method,
20
        mxc_uri, room_id,
21
    },
22
    test_utils::mocks::MatrixMockServer,
23
};
24
use matrix_sdk_test::{ALICE, JoinedRoomBuilder, event_factory::EventFactory};
25
use serde_json::{Value, json};
26
use wiremock::{
27
    Mock, ResponseTemplate,
28
    matchers::{method, path_regex},
29
};
30
31
struct MediaHarness {
32
    server: MatrixMockServer,
33
    client: matrix_sdk::Client,
34
    handle: ClientHandle,
35
    events: EventStream,
36
    running: tokio::task::JoinHandle<Result<(), EngineError>>,
37
    media_id: MediaId,
38
}
39
40
#[tokio::test]
41
async fn full_size_uses_full_source() {
42
    let mut harness = start(file_with_thumbnail()).await;
43
    mount_downloads(&harness.server).await;
44
45
    let bytes = fetch_ready(&mut harness, MediaSize::Full).await;
46
    assert_eq!(&*bytes, b"binaryjpegfullimagedata");
47
    assert!(
48
        media_paths(&harness.server)
49
            .await
50
            .iter()
51
            .any(|path| path.ends_with("/localhost/full"))
52
    );
53
54
    stop(harness).await;
55
}
56
57
#[tokio::test]
58
async fn simultaneous_fetches_share_one_download() {
59
    let mut harness = start(plain_image()).await;
60
    Mock::given(method("GET"))
61
        .and(path_regex("/download/localhost/image$"))
62
        .respond_with(
63
            ResponseTemplate::new(200)
64
                .set_body_bytes(b"shared image bytes".to_vec())
65
                .set_delay(Duration::from_millis(300)),
66
        )
67
        .mount(harness.server.server())
68
        .await;
69
70
    let mut pending = Vec::new();
71
    for _ in 0..2 {
72
        pending.push(
73
            harness
74
                .handle
75
                .send(Command::Media(MediaCommand::Fetch {
76
                    id: harness.media_id.clone(),
77
                    size: MediaSize::Full,
78
                }))
79
                .await
80
                .unwrap(),
81
        );
82
    }
83
    let mut ready = false;
84
    tokio::time::timeout(Duration::from_secs(5), async {
85
        loop {
86
            match harness.events.recv().await.expect("event stream closed") {
87
                Event::MediaReady {
88
                    media_id, bytes, ..
89
                } if media_id == harness.media_id => {
90
                    assert_eq!(&*bytes, b"shared image bytes");
91
                    ready = true;
92
                }
93
                Event::CommandSucceeded { id, .. } => pending.retain(|waiting| *waiting != id),
94
                Event::CommandFailed(failure) if pending.contains(&failure.id) => {
95
                    panic!("fetch failed: {failure:?}");
96
                }
97
                _ => {}
98
            }
99
            if ready && pending.is_empty() {
100
                break;
101
            }
102
        }
103
    })
104
    .await
105
    .expect("both fetches did not complete");
106
    let downloads = harness
107
        .server
108
        .received_requests()
109
        .await
110
        .unwrap()
111
        .iter()
112
        .filter(|request| request.url.path().ends_with("/download/localhost/image"))
113
        .count();
114
    assert_eq!(downloads, 1, "duplicate in-flight media download");
115
    stop(harness).await;
116
}
117
118
#[tokio::test]
119
async fn downloaded_media_over_limit_is_rejected() {
120
    let mut harness = start_with_config(
121
        plain_image(),
122
        EngineConfig::default()
123
            .with_initial_history_limit(0)
124
            .with_max_media_bytes(4),
125
    )
126
    .await;
127
    Mock::given(method("GET"))
128
        .and(path_regex("/download/localhost/image$"))
129
        .respond_with(ResponseTemplate::new(200).set_body_bytes(b"too large".to_vec()))
130
        .mount(harness.server.server())
131
        .await;
132
    let id = harness
133
        .handle
134
        .send(Command::Media(MediaCommand::Fetch {
135
            id: harness.media_id.clone(),
136
            size: MediaSize::Full,
137
        }))
138
        .await
139
        .unwrap();
140
    receive_until(&mut harness.events, |event| {
141
        matches!(event, Event::CommandFailed(failure) if failure.id == id && failure.kind == CommandErrorKind::MediaTooLarge)
142
    })
143
    .await;
144
    stop(harness).await;
145
}
146
147
#[tokio::test]
148
async fn thumbnail_size_prefers_explicit_thumbnail() {
149
    let mut harness = start(file_with_thumbnail()).await;
150
    mount_downloads(&harness.server).await;
151
152
    let bytes = fetch_ready(&mut harness, MediaSize::Thumbnail).await;
153
    assert_eq!(&*bytes, b"binaryjpegfullimagedata");
154
    let paths = media_paths(&harness.server).await;
155
    assert!(
156
        paths
157
            .iter()
158
            .any(|path| path.ends_with("/localhost/explicit-thumbnail"))
159
    );
160
    assert!(!paths.iter().any(|path| path.ends_with("/localhost/full")));
161
162
    stop(harness).await;
163
}
164
165
#[tokio::test]
166
async fn plain_image_thumbnail_uses_thumbnail_endpoint() {
167
    let mut harness = start(plain_image()).await;
168
    harness
169
        .server
170
        .mock_media_thumbnail(Method::Scale, 800, 800, false)
171
        .ok()
172
        .mount()
173
        .await;
174
    harness
175
        .server
176
        .mock_authed_media_thumbnail(Method::Scale, 800, 800, false)
177
        .ok()
178
        .mount()
179
        .await;
180
181
    let bytes = fetch_ready(&mut harness, MediaSize::Thumbnail).await;
182
    assert_eq!(&*bytes, b"binaryjpegthumbnaildata");
183
    assert!(
184
        media_paths(&harness.server)
185
            .await
186
            .iter()
187
            .any(|path| path.contains("/thumbnail/localhost/image"))
188
    );
189
190
    stop(harness).await;
191
}
192
193
#[tokio::test]
194
async fn unavailable_thumbnail_reports_returned_command_id() {
195
    let mut harness = start(plain_audio()).await;
196
    let command_id = harness
197
        .handle
198
        .send(Command::Media(MediaCommand::Fetch {
199
            id: harness.media_id.clone(),
200
            size: MediaSize::Thumbnail,
201
        }))
202
        .await
203
        .unwrap();
204
205
    let event = receive_until(&mut harness.events, |event| {
206
        matches!(
207
            event,
208
            Event::CommandFailed(failure)
209
                if failure.id == command_id && failure.kind == CommandErrorKind::MediaUnavailable
210
        )
211
    })
212
    .await;
213
    assert!(matches!(
214
        event,
215
        Event::CommandFailed(failure)
216
            if failure.id == command_id && failure.kind == CommandErrorKind::MediaUnavailable
217
    ));
218
219
    stop(harness).await;
220
}
221
222
#[tokio::test]
223
async fn media_upload_is_queued_and_sent() {
224
    let mut harness = start(plain_image()).await;
225
    harness
226
        .server
227
        .mock_authenticated_media_config()
228
        .ok_default()
229
        .mount()
230
        .await;
231
    harness
232
        .server
233
        .mock_upload()
234
        .expect_mime_type("image/jpeg")
235
        .ok(mxc_uri!("mxc://localhost/uploaded"))
236
        .expect(1)
237
        .mount()
238
        .await;
239
    harness
240
        .server
241
        .mock_room_send()
242
        .ok(event_id!("$uploaded:localhost"))
243
        .expect(1)
244
        .mount()
245
        .await;
246
247
    let id = harness
248
        .handle
249
        .upload_media(
250
            room_id!("!media:localhost").to_owned().into(),
251
            "photo.jpg".to_owned(),
252
            "image/jpeg".to_owned(),
253
            b"photo data".to_vec(),
254
            Some("holiday".to_owned()),
255
        )
256
        .await
257
        .unwrap();
258
    receive_until(
259
        &mut harness.events,
260
        |event| matches!(event, Event::CommandSucceeded { id: completed, .. } if *completed == id),
261
    )
262
    .await;
263
264
    tokio::time::timeout(Duration::from_secs(5), async {
265
        loop {
266
            let requests = harness.server.received_requests().await.unwrap();
267
            if requests.iter().any(|request| {
268
                request.url.path().contains("/send/m.room.message/")
269
                    && String::from_utf8_lossy(&request.body).contains("mxc://localhost/uploaded")
270
            }) {
271
                break;
272
            }
273
            tokio::task::yield_now().await;
274
        }
275
    })
276
    .await
277
    .expect("queued media was not sent");
278
279
    stop(harness).await;
280
}
281
282
#[tokio::test]
283
async fn queued_offline_attachment_exposes_fetchable_local_media() {
284
    let mut harness = start(plain_image()).await;
285
    harness.client.send_queue().set_enabled(false).await;
286
    let original = b"offline image bytes".to_vec();
287
    let mut upload = AttachmentUpload::new(
288
        room_id!("!media:localhost").to_owned(),
289
        "pending.png",
290
        "image/png",
291
        original.clone(),
292
    );
293
    upload.details = Some(UploadDetails::Image {
294
        width: Some(32),
295
        height: Some(16),
296
        blurhash: None,
297
    });
298
    harness.handle.upload_attachment(upload).await.unwrap();
299
    let event = receive_until(&mut harness.events, |event| {
300
        event_messages(event).any(|message| {
301
            message.own
302
                && message.transaction_id.is_some()
303
                && message
304
                    .media
305
                    .as_ref()
306
                    .is_some_and(|media| media.filename == "pending.png")
307
        })
308
    })
309
    .await;
310
    let media = event_messages(&event)
311
        .find(|message| message.own && message.media.is_some())
312
        .and_then(|message| message.media.as_ref())
313
        .unwrap();
314
    assert_eq!((media.width, media.height), (Some(32), Some(16)));
315
    assert_eq!(media.size_bytes, Some(original.len() as u64));
316
    let media_id = media.id.clone();
317
318
    let fetch_id = harness
319
        .handle
320
        .send(Command::Media(MediaCommand::Fetch {
321
            id: media_id.clone(),
322
            size: MediaSize::Full,
323
        }))
324
        .await
325
        .unwrap();
326
    let ready = receive_until(
327
        &mut harness.events,
328
        |event| matches!(event, Event::MediaReady { media_id: id, .. } if id == &media_id),
329
    )
330
    .await;
331
    assert!(matches!(ready, Event::MediaReady { bytes, .. } if *bytes == original));
332
    receive_until(
333
        &mut harness.events,
334
        |event| matches!(event, Event::CommandSucceeded { id, .. } if *id == fetch_id),
335
    )
336
    .await;
337
    stop(harness).await;
338
}
339
340
#[tokio::test]
341
async fn image_upload_uses_caller_supplied_metadata_and_thumbnail() {
342
    let mut harness = start(plain_image()).await;
343
    harness
344
        .server
345
        .mock_authenticated_media_config()
346
        .ok_default()
347
        .mount()
348
        .await;
349
    harness
350
        .server
351
        .mock_upload()
352
        .expect_mime_type("image/png")
353
        .respond_with(|request: &wiremock::Request| {
354
            let uri = if request.body == b"thumbnail bytes" {
355
                "mxc://localhost/image-thumb"
356
            } else {
357
                "mxc://localhost/image-upload"
358
            };
359
            ResponseTemplate::new(200).set_body_json(json!({ "content_uri": uri }))
360
        })
361
        .expect(2)
362
        .mount()
363
        .await;
364
    harness
365
        .server
366
        .mock_room_send()
367
        .ok(event_id!("$image-upload:localhost"))
368
        .expect(1)
369
        .mount()
370
        .await;
371
372
    let original = b"image bytes".to_vec();
373
    let thumbnail_bytes = b"thumbnail bytes".to_vec();
374
    let mut upload = AttachmentUpload::new(
375
        room_id!("!media:localhost").to_owned(),
376
        "large.png",
377
        "image/png",
378
        original.clone(),
379
    );
380
    upload.details = Some(UploadDetails::Image {
381
        width: Some(900),
382
        height: Some(450),
383
        blurhash: None,
384
    });
385
    upload.thumbnail = Some(UploadThumbnail {
386
        bytes: thumbnail_bytes.clone(),
387
        mimetype: "image/png".to_owned(),
388
        width: 800,
389
        height: 400,
390
    });
391
    let id = harness.handle.upload_attachment(upload).await.unwrap();
392
    receive_until(
393
        &mut harness.events,
394
        |event| matches!(event, Event::CommandSucceeded { id: completed, .. } if *completed == id),
395
    )
396
    .await;
397
398
    let content = uploaded_content(&harness.server).await;
399
    assert_eq!(content["msgtype"], "m.image");
400
    assert_eq!(content["info"]["w"], 900);
401
    assert_eq!(content["info"]["h"], 450);
402
    assert_eq!(content["info"]["size"], original.len());
403
    assert_eq!(content["info"]["thumbnail_info"]["w"], 800);
404
    assert_eq!(content["info"]["thumbnail_info"]["h"], 400);
405
    assert_eq!(content["info"]["thumbnail_info"]["mimetype"], "image/png");
406
    assert_eq!(
407
        content["info"]["thumbnail_info"]["size"],
408
        thumbnail_bytes.len()
409
    );
410
    assert_eq!(
411
        content["info"]["thumbnail_url"],
412
        "mxc://localhost/image-thumb"
413
    );
414
    assert_eq!(content["url"], "mxc://localhost/image-upload");
415
416
    let requests = harness.server.received_requests().await.unwrap();
417
    let uploads = requests
418
        .iter()
419
        .filter(|request| request.url.path().ends_with("/_matrix/media/v3/upload"))
420
        .collect::<Vec<_>>();
421
    assert_eq!(uploads.len(), 2);
422
    assert!(uploads.iter().any(|request| request.body == original));
423
    assert!(
424
        uploads
425
            .iter()
426
            .any(|request| request.body == thumbnail_bytes)
427
    );
428
429
    stop(harness).await;
430
}
431
432
#[tokio::test]
433
async fn video_upload_preserves_metadata_and_supplied_thumbnail() {
434
    let mut harness = start(plain_image()).await;
435
    harness
436
        .server
437
        .mock_authenticated_media_config()
438
        .ok_default()
439
        .mount()
440
        .await;
441
    harness
442
        .server
443
        .mock_upload()
444
        .expect_mime_type("image/png")
445
        .ok(mxc_uri!("mxc://localhost/video-thumb"))
446
        .expect(1)
447
        .mount()
448
        .await;
449
    harness
450
        .server
451
        .mock_upload()
452
        .expect_mime_type("video/mp4")
453
        .ok(mxc_uri!("mxc://localhost/video-upload"))
454
        .expect(1)
455
        .mount()
456
        .await;
457
    harness
458
        .server
459
        .mock_room_send()
460
        .ok(event_id!("$video-upload:localhost"))
461
        .expect(1)
462
        .mount()
463
        .await;
464
465
    let mut upload = AttachmentUpload::new(
466
        room_id!("!media:localhost").to_owned(),
467
        "clip.mp4",
468
        "video/mp4",
469
        b"video bytes".to_vec(),
470
    );
471
    upload.details = Some(UploadDetails::Video {
472
        width: Some(1920),
473
        height: Some(1080),
474
        duration_ms: Some(3500),
475
        blurhash: Some("blurhash".to_owned()),
476
    });
477
    upload.thumbnail = Some(UploadThumbnail {
478
        bytes: b"poster bytes".to_vec(),
479
        mimetype: "image/png".to_owned(),
480
        width: 320,
481
        height: 180,
482
    });
483
    let id = harness.handle.upload_attachment(upload).await.unwrap();
484
    receive_until(
485
        &mut harness.events,
486
        |event| matches!(event, Event::CommandSucceeded { id: completed, .. } if *completed == id),
487
    )
488
    .await;
489
490
    let content = uploaded_content(&harness.server).await;
491
    assert_eq!(content["msgtype"], "m.video");
492
    assert_eq!(content["info"]["w"], 1920);
493
    assert_eq!(content["info"]["h"], 1080);
494
    assert_eq!(content["info"]["duration"], 3500);
495
    assert_eq!(content["info"]["size"], 11);
496
    assert_eq!(content["info"]["thumbnail_info"]["w"], 320);
497
    assert_eq!(content["info"]["thumbnail_info"]["h"], 180);
498
    assert_eq!(content["info"]["thumbnail_info"]["size"], 12);
499
    assert_eq!(
500
        content["info"]["thumbnail_url"],
501
        "mxc://localhost/video-thumb"
502
    );
503
    assert_eq!(content["url"], "mxc://localhost/video-upload");
504
505
    let requests = harness.server.received_requests().await.unwrap();
506
    assert!(requests.iter().any(|request| {
507
        request.url.path().ends_with("/_matrix/media/v3/upload") && request.body == b"poster bytes"
508
    }));
509
510
    stop(harness).await;
511
}
512
513
#[tokio::test]
514
async fn voice_upload_emits_voice_details_and_correlated_progress() {
515
    let mut harness = start(plain_image()).await;
516
    harness
517
        .server
518
        .mock_authenticated_media_config()
519
        .ok_default()
520
        .mount()
521
        .await;
522
    harness
523
        .server
524
        .mock_upload()
525
        .expect_mime_type("audio/ogg")
526
        .respond_with(
527
            ResponseTemplate::new(200)
528
                .set_body_json(json!({ "content_uri": "mxc://localhost/voice-upload" }))
529
                .set_delay(Duration::from_millis(500)),
530
        )
531
        .expect(1)
532
        .mount()
533
        .await;
534
    harness
535
        .server
536
        .mock_room_send()
537
        .ok(event_id!("$voice-upload:localhost"))
538
        .expect(1)
539
        .mount()
540
        .await;
541
542
    let data = b"voice bytes".to_vec();
543
    let mut upload = AttachmentUpload::new(
544
        room_id!("!media:localhost").to_owned(),
545
        "voice.ogg",
546
        "audio/ogg",
547
        data.clone(),
548
    );
549
    upload.details = Some(UploadDetails::Voice {
550
        duration_ms: 4200,
551
        waveform: vec![0, 512, 1024],
552
    });
553
    let id = harness.handle.upload_attachment(upload).await.unwrap();
554
    receive_until(
555
        &mut harness.events,
556
        |event| matches!(event, Event::CommandSucceeded { id: completed, .. } if *completed == id),
557
    )
558
    .await;
559
560
    let pending = harness
561
        .handle
562
        .pending_send_ids(room_id!("!media:localhost"))
563
        .await
564
        .unwrap();
565
    assert_eq!(pending.len(), 1);
566
    let transaction_id = pending[0].clone();
567
    let progress = receive_until(&mut harness.events, |event| {
568
        matches!(event, Event::MediaUploadProgress { transaction_id: seen, complete: true, .. } if seen == &transaction_id)
569
    })
570
    .await;
571
    assert!(matches!(progress, Event::MediaUploadProgress {
572
        room_id,
573
        transaction_id: seen,
574
        uploaded_bytes,
575
        total_bytes,
576
        complete: true,
577
    } if room_id == room_id!("!media:localhost") && seen == transaction_id && uploaded_bytes == data.len() as u64 && total_bytes == data.len() as u64));
578
579
    let content = uploaded_content(&harness.server).await;
580
    assert_eq!(content["msgtype"], "m.audio");
581
    assert_eq!(content["info"]["duration"], 4200);
582
    assert_eq!(content["info"]["size"], data.len());
583
    assert_eq!(content["org.matrix.msc1767.audio"]["duration"], 4200);
584
    assert_eq!(
585
        content["org.matrix.msc1767.audio"]["waveform"],
586
        json!([0, 512, 1024])
587
    );
588
    assert_eq!(content["org.matrix.msc3245.voice"], json!({}));
589
590
    stop(harness).await;
591
}
592
593
#[tokio::test]
594
async fn mismatched_upload_metadata_and_invalid_waveform_fail_without_uploading() {
595
    let mut harness = start(plain_image()).await;
596
    let mut mismatched = AttachmentUpload::new(
597
        room_id!("!media:localhost").to_owned(),
598
        "clip.mp4",
599
        "video/mp4",
600
        b"video bytes".to_vec(),
601
    );
602
    mismatched.details = Some(UploadDetails::Image {
603
        width: Some(32),
604
        height: Some(32),
605
        blurhash: None,
606
    });
607
    let id = harness.handle.upload_attachment(mismatched).await.unwrap();
608
    let failure = receive_until(
609
        &mut harness.events,
610
        |event| matches!(event, Event::CommandFailed(failure) if failure.id == id),
611
    )
612
    .await;
613
    assert!(
614
        matches!(failure, Event::CommandFailed(failure) if failure.kind == CommandErrorKind::Other)
615
    );
616
617
    let mut voice = AttachmentUpload::new(
618
        room_id!("!media:localhost").to_owned(),
619
        "voice.ogg",
620
        "audio/ogg",
621
        b"voice bytes".to_vec(),
622
    );
623
    voice.details = Some(UploadDetails::Voice {
624
        duration_ms: 1000,
625
        waveform: vec![1025],
626
    });
627
    let id = harness.handle.upload_attachment(voice).await.unwrap();
628
    let failure = receive_until(
629
        &mut harness.events,
630
        |event| matches!(event, Event::CommandFailed(failure) if failure.id == id),
631
    )
632
    .await;
633
    assert!(
634
        matches!(failure, Event::CommandFailed(failure) if failure.kind == CommandErrorKind::Other)
635
    );
636
637
    assert!(
638
        !harness
639
            .server
640
            .received_requests()
641
            .await
642
            .unwrap()
643
            .iter()
644
            .any(|request| { request.url.path().ends_with("/_matrix/media/v3/upload") })
645
    );
646
    stop(harness).await;
647
}
648
649
#[tokio::test]
650
async fn upload_over_media_limit_fails_without_uploading() {
651
    let mut harness = start_with_config(
652
        plain_image(),
653
        EngineConfig::default()
654
            .with_initial_history_limit(0)
655
            .with_max_media_bytes(4),
656
    )
657
    .await;
658
    let upload = AttachmentUpload::new(
659
        room_id!("!media:localhost").to_owned(),
660
        "large.bin",
661
        "application/octet-stream",
662
        b"too large".to_vec(),
663
    );
664
    let id = harness.handle.upload_attachment(upload).await.unwrap();
665
    let failure = receive_until(
666
        &mut harness.events,
667
        |event| matches!(event, Event::CommandFailed(failure) if failure.id == id),
668
    )
669
    .await;
670
    assert!(matches!(
671
        failure,
672
        Event::CommandFailed(failure) if failure.kind == CommandErrorKind::MediaTooLarge
673
    ));
674
    assert!(
675
        !harness
676
            .server
677
            .received_requests()
678
            .await
679
            .unwrap()
680
            .iter()
681
            .any(|request| { request.url.path().ends_with("/_matrix/media/v3/upload") })
682
    );
683
    stop(harness).await;
684
}
685
686
#[tokio::test]
687
async fn audio_thumbnail_fails_without_uploading() {
688
    let mut harness = start(plain_image()).await;
689
    let mut upload = AttachmentUpload::new(
690
        room_id!("!media:localhost").to_owned(),
691
        "voice.ogg",
692
        "audio/ogg",
693
        b"voice bytes".to_vec(),
694
    );
695
    upload.thumbnail = Some(UploadThumbnail {
696
        bytes: b"thumbnail".to_vec(),
697
        mimetype: "image/png".to_owned(),
698
        width: 80,
699
        height: 80,
700
    });
701
    let id = harness.handle.upload_attachment(upload).await.unwrap();
702
    let failure = receive_until(
703
        &mut harness.events,
704
        |event| matches!(event, Event::CommandFailed(failure) if failure.id == id),
705
    )
706
    .await;
707
    assert!(matches!(
708
        failure,
709
        Event::CommandFailed(failure) if failure.kind == CommandErrorKind::MediaUnavailable
710
    ));
711
    assert!(
712
        !harness
713
            .server
714
            .received_requests()
715
            .await
716
            .unwrap()
717
            .iter()
718
            .any(|request| { request.url.path().ends_with("/_matrix/media/v3/upload") })
719
    );
720
    stop(harness).await;
721
}
722
723
#[tokio::test]
724
async fn invalid_upload_mime_reports_a_command_failure() {
725
    let mut harness = start(plain_image()).await;
726
    let id = harness
727
        .handle
728
        .upload_media(
729
            room_id!("!media:localhost").to_owned().into(),
730
            "photo.jpg".to_owned(),
731
            "not a MIME type".to_owned(),
732
            b"photo data".to_vec(),
733
            None,
734
        )
735
        .await
736
        .unwrap();
737
    let event = receive_until(
738
        &mut harness.events,
739
        |event| matches!(event, Event::CommandFailed(failure) if failure.id == id),
740
    )
741
    .await;
742
    assert!(
743
        matches!(event, Event::CommandFailed(failure) if failure.kind == CommandErrorKind::Other)
744
    );
745
746
    stop(harness).await;
747
}
748
749
async fn start(content: RoomMessageEventContent) -> MediaHarness {
750
    start_with_config(
751
        content,
752
        EngineConfig::default().with_initial_history_limit(0),
753
    )
754
    .await
755
}
756
757
async fn start_with_config(content: RoomMessageEventContent, config: EngineConfig) -> MediaHarness {
758
    let server = MatrixMockServer::new().await;
759
    server
760
        .mock_sliding_sync()
761
        .ok(v5::Response::new("1".to_owned()))
762
        .mount()
763
        .await;
764
    let client = server.client_builder().build().await;
765
    let room_id = room_id!("!media:localhost");
766
    server.mock_room_state_encryption().plain().mount().await;
767
    server.sync_joined_room(&client, room_id).await;
768
    let sync_client = client.clone();
769
    let (engine, handle, mut events) = Engine::from_client(client, config).await.unwrap();
770
    let running = tokio::spawn(engine.run());
771
772
    handle
773
        .send(Command::Room(RoomCommand::Open(room_id.to_owned())))
774
        .await
775
        .unwrap();
776
    receive_until(
777
        &mut events,
778
        |event| matches!(event, Event::TimelineReset { timeline_id: reset, .. } if reset.room_id() == room_id),
779
    )
780
    .await;
781
782
    server
783
        .sync_room(
784
            &sync_client,
785
            JoinedRoomBuilder::new(room_id)
786
                .add_timeline_event(EventFactory::new().event(content).sender(&ALICE)),
787
        )
788
        .await;
789
    let event = receive_until(&mut events, |event| {
790
        event_messages(event).any(|message| message.media.is_some())
791
    })
792
    .await;
793
    let media_id = event_messages(&event)
794
        .find_map(|message| message.media.as_ref().map(|media| media.id.clone()))
795
        .unwrap();
796
797
    MediaHarness {
798
        server,
799
        client: sync_client,
800
        handle,
801
        events,
802
        running,
803
        media_id,
804
    }
805
}
806
807
async fn mount_downloads(server: &MatrixMockServer) {
808
    server.mock_media_download().ok_image().mount().await;
809
    server.mock_authed_media_download().ok_image().mount().await;
810
}
811
812
async fn fetch_ready(harness: &mut MediaHarness, size: MediaSize) -> Arc<[u8]> {
813
    harness
814
        .handle
815
        .send(Command::Media(MediaCommand::Fetch {
816
            id: harness.media_id.clone(),
817
            size,
818
        }))
819
        .await
820
        .unwrap();
821
    match receive_until(&mut harness.events, |event| {
822
        matches!(
823
            event,
824
            Event::MediaReady { media_id, size: received, .. }
825
                if media_id == &harness.media_id && received == &size
826
        )
827
    })
828
    .await
829
    {
830
        Event::MediaReady { bytes, .. } => bytes,
831
        _ => unreachable!(),
832
    }
833
}
834
835
async fn stop(mut harness: MediaHarness) {
836
    harness
837
        .handle
838
        .send(Command::Lifecycle(LifecycleCommand::Shutdown))
839
        .await
840
        .unwrap();
841
    receive_until(&mut harness.events, |event| matches!(event, Event::Stopped)).await;
842
    harness.running.await.unwrap().unwrap();
843
}
844
845
async fn media_paths(server: &MatrixMockServer) -> Vec<String> {
846
    server
847
        .received_requests()
848
        .await
849
        .unwrap()
850
        .into_iter()
851
        .map(|request| request.url.path().to_owned())
852
        .filter(|path| path.contains("/media/"))
853
        .collect()
854
}
855
856
async fn uploaded_content(server: &MatrixMockServer) -> Value {
857
    tokio::time::timeout(Duration::from_secs(5), async {
858
        loop {
859
            if let Some(request) = server
860
                .received_requests()
861
                .await
862
                .unwrap()
863
                .iter()
864
                .find(|request| request.url.path().contains("/send/m.room.message/"))
865
            {
866
                return serde_json::from_slice(&request.body).unwrap();
867
            }
868
            tokio::task::yield_now().await;
869
        }
870
    })
871
    .await
872
    .expect("queued media was not sent")
873
}
874
875
fn file_with_thumbnail() -> RoomMessageEventContent {
876
    let mut info = FileInfo::new();
877
    info.thumbnail_source = Some(MediaSource::Plain(
878
        mxc_uri!("mxc://localhost/explicit-thumbnail").to_owned(),
879
    ));
880
    RoomMessageEventContent::new(MessageType::File(
881
        FileMessageEventContent::plain(
882
            "file.bin".to_owned(),
883
            mxc_uri!("mxc://localhost/full").to_owned(),
884
        )
885
        .info(Box::new(info)),
886
    ))
887
}
888
889
fn plain_image() -> RoomMessageEventContent {
890
    RoomMessageEventContent::new(MessageType::Image(ImageMessageEventContent::plain(
891
        "image.jpg".to_owned(),
892
        mxc_uri!("mxc://localhost/image").to_owned(),
893
    )))
894
}
895
896
fn plain_audio() -> RoomMessageEventContent {
897
    RoomMessageEventContent::new(MessageType::Audio(AudioMessageEventContent::plain(
898
        "audio.ogg".to_owned(),
899
        mxc_uri!("mxc://localhost/audio").to_owned(),
900
    )))
901
}
902
903
fn event_messages(event: &Event) -> std::vec::IntoIter<&macltok::TimelineMessage> {
904
    let items = match event {
905
        Event::TimelineReset { items, .. } => items.iter().collect::<Vec<_>>(),
906
        Event::TimelineChanged { diffs, .. } => {
907
            diffs.iter().flat_map(diff_items).collect::<Vec<_>>()
908
        }
909
        _ => Vec::new(),
910
    };
911
    items
912
        .into_iter()
913
        .filter_map(|item| match item {
914
            TimelineItem::Message(message) => Some(message.as_ref()),
915
            _ => None,
916
        })
917
        .collect::<Vec<_>>()
918
        .into_iter()
919
}
920
921
fn diff_items(diff: &TimelineDiff) -> Vec<&TimelineItem> {
922
    match diff {
923
        TimelineDiff::Append { values } | TimelineDiff::Reset { values } => values.iter().collect(),
924
        TimelineDiff::PushFront { value }
925
        | TimelineDiff::PushBack { value }
926
        | TimelineDiff::Insert { value, .. }
927
        | TimelineDiff::Set { value, .. } => vec![value],
928
        _ => Vec::new(),
929
    }
930
}
931
932
async fn receive_until(events: &mut EventStream, predicate: impl Fn(&Event) -> bool) -> Event {
933
    tokio::time::timeout(Duration::from_secs(5), async {
934
        while let Some(event) = events.recv().await {
935
            if predicate(&event) {
936
                return event;
937
            }
938
        }
939
        panic!("event stream closed before the expected event");
940
    })
941
    .await
942
    .expect("expected event did not arrive")
943
}