1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
|
"""
`_do_link_preview_request` — routing, the media_cache image round-trip, and
the in-memory result cache.
The fetch itself (`linkpreview.fetch_preview` / `fetch_image`) is covered by
test_linkpreview.py and stubbed here, so this is purely about what the
handler does with a result: reply shape, storing the OG image under its
blake3 in the thumb store, and not re-fetching a URL it has already seen.
"""
import blake3
import pytest
from meshbay_common.protocol import MNP
from meshbay_node import linkpreview
from meshbay_node.media_cache import MediaCache
from meshbay_node.transport import webrtc_server
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
pytestmark = pytest.mark.asyncio
@pytest.fixture
async def media_cache(tmp_path):
c = MediaCache(db_path=tmp_path / "media_cache.db")
await c.open()
yield c
await c.close()
@pytest.fixture(autouse=True)
def _clear_cache():
webrtc_server._link_preview_cache.clear()
yield
webrtc_server._link_preview_cache.clear()
def _session(media_cache):
s = WebRTCPeerSession.__new__(WebRTCPeerSession)
s._ctx = {"media_cache": media_cache}
s.sent = []
s._send = s.sent.append
return s
async def test_reply_carries_the_card_and_caches_the_image(media_cache, monkeypatch):
async def fake_preview(url, **k):
return {"url": url, "title": "Hello", "description": "World",
"site_name": "Example", "image_url": "https://example.com/c.png"}
async def fake_image(url, **k):
return b"jpeg-bytes"
monkeypatch.setattr(linkpreview, "fetch_preview", fake_preview)
monkeypatch.setattr(linkpreview, "fetch_image", fake_image)
s = _session(media_cache)
await s._do_link_preview_request({"url": "https://example.com/p"})
(resp,) = s.sent
assert resp["type"] == MNP.LINK_PREVIEW_RESP
assert resp["ok"] is True
assert resp["title"] == "Hello"
assert resp["site_name"] == "Example"
want_hash = blake3.blake3(b"jpeg-bytes").hexdigest()
assert resp["image_thumb_hash"] == want_hash
# And the bytes are in the thumb store, servable via the normal file_req path.
assert await media_cache.get_thumb(want_hash) == b"jpeg-bytes"
async def test_unfurlable_failure_is_ok_false(media_cache, monkeypatch):
async def none_preview(url, **k):
return None
monkeypatch.setattr(linkpreview, "fetch_preview", none_preview)
s = _session(media_cache)
await s._do_link_preview_request({"url": "http://169.254.169.254/"})
(resp,) = s.sent
assert resp["ok"] is False
assert "image_thumb_hash" not in resp
async def test_second_request_for_the_same_url_is_served_from_cache(media_cache, monkeypatch):
calls = {"n": 0}
async def counting_preview(url, **k):
calls["n"] += 1
return {"url": url, "title": "Once", "description": None,
"site_name": None, "image_url": None}
monkeypatch.setattr(linkpreview, "fetch_preview", counting_preview)
s = _session(media_cache)
await s._do_link_preview_request({"url": "https://example.com/a"})
await s._do_link_preview_request({"url": "https://example.com/a"})
assert calls["n"] == 1
assert len(s.sent) == 2
assert s.sent[0]["title"] == s.sent[1]["title"] == "Once"
assert all(r["type"] == MNP.LINK_PREVIEW_RESP for r in s.sent)
|