Mercurial > silorider

comparison tests/test_silos_bluesky.py @ 46:ac5911e76ef2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Add Bluesky silo Currently supports: - Uploading images - URL facets
author Ludovic Chabant <ludovic@chabant.com>
date Sun, 20 Aug 2023 11:20:37 -0700
parents
children 486affad656e
comparison
equal deleted inserted replaced
42:67fde62e3862 46:ac5911e76ef2
1 import pytest
2 import atproto.xrpc_client.models as atprotomodels
3 from .mockutil import mock_urllib
4
5
6 def test_one_article(cli, feedutil, bskymock):
7 feed = cli.createTempFeed(feedutil.makeFeed(
8 """<h1 class="p-name">A new article</h1>
9 <div class="e-content">
10 <p>This is the text of the article.</p>
11 <p>It has 2 paragraphs.</p>
12 </div>
13 <a class="u-url" href="https://example.org/a-new-article">permalink</a>"""
14 ))
15
16 cli.appendSiloConfig('test', 'bluesky')
17 cli.setFeedConfig('feed', feed)
18 bskymock.installCredentials(cli, 'test')
19
20 ctx, _ = cli.run('process')
21 assert ctx.cache.wasPosted('test', 'https://example.org/a-new-article')
22 post = ctx.silos[0].client.posts[0]
23 assert post == ('A new article https://example.org/a-new-article',
24 None, None)
25
26
27 def test_one_micropost(cli, feedutil, bskymock):
28 feed = cli.createTempFeed(feedutil.makeFeed(
29 """<p class="p-name">This is a quick update.</p>
30 <a class="u-url" href="/01234.html">permalink</a>"""
31 ))
32
33 cli.appendSiloConfig('test', 'bluesky')
34 cli.setFeedConfig('feed', feed)
35 bskymock.installCredentials(cli, 'test')
36
37 ctx, _ = cli.run('process')
38 assert ctx.cache.wasPosted('test', '/01234.html')
39 post = ctx.silos[0].client.posts[0]
40 assert post == ("This is a quick update.", None, None)
41
42
43 def test_one_micropost_with_one_photo(cli, feedutil, bskymock, monkeypatch):
44 feed = cli.createTempFeed(feedutil.makeFeed(
45 """<p class="p-name">This is a quick photo update.</p>
46 <div>
47 <a class="u-photo" href="/fullimg.jpg"><img src="/thumbimg.jpg"/></a>
48 </div>
49 <a class="u-url" href="/01234.html">permalink</a>"""
50 ))
51
52 cli.appendSiloConfig('test', 'bluesky')
53 cli.setFeedConfig('feed', feed)
54 bskymock.installCredentials(cli, 'test')
55
56 with monkeypatch.context() as m:
57 import silorider.silos.bluesky
58 mock_urllib(m)
59 m.setattr(silorider.silos.bluesky.BlueskySilo, '_media_callback',
60 _patched_media_callback)
61 ctx, _ = cli.run('process')
62
63 assert ctx.cache.wasPosted('test', '/01234.html')
64 blob = ctx.silos[0].client.blobs[0]
65 assert blob == ('/retrieved/fullimg.jpg', None)
66 post = ctx.silos[0].client.posts[0]
67 embed = atprotomodels.AppBskyEmbedImages.Main(images=[1])
68 assert post == ("This is a quick photo update.", embed, None)
69
70
71 def test_one_micropost_with_two_photos(cli, feedutil, bskymock, monkeypatch):
72 feed = cli.createTempFeed(feedutil.makeFeed(
73 """<p class="p-name">This is a photo update with 2 photos.</p>
74 <div>
75 <a class="u-photo" href="/fullimg1.jpg"><img src="/thumbimg1.jpg"/></a>
76 <a class="u-photo" href="/fullimg2.jpg"><img src="/thumbimg2.jpg"/></a>
77 </div>
78 <a class="u-url" href="/01234.html">permalink</a>"""
79 ))
80
81 cli.appendSiloConfig('test', 'bluesky')
82 cli.setFeedConfig('feed', feed)
83 bskymock.installCredentials(cli, 'test')
84
85 with monkeypatch.context() as m:
86 import silorider.silos.bluesky
87 mock_urllib(m)
88 m.setattr(silorider.silos.bluesky.BlueskySilo, '_media_callback',
89 _patched_media_callback)
90 ctx, _ = cli.run('process')
91
92 assert ctx.cache.wasPosted('test', '/01234.html')
93 blob = ctx.silos[0].client.blobs[0]
94 assert blob == ('/retrieved/fullimg1.jpg', None)
95 blob = ctx.silos[0].client.blobs[1]
96 assert blob == ('/retrieved/fullimg2.jpg', None)
97 post = ctx.silos[0].client.posts[0]
98 embed = atprotomodels.AppBskyEmbedImages.Main(images=[1, 2])
99 assert post == ("This is a photo update with 2 photos.", embed, None)
100
101
102 def test_one_micropost_with_links(cli, feedutil, bskymock):
103 cli.appendSiloConfig('test', 'bluesky')
104 bskymock.installCredentials(cli, 'test')
105
106 feed = cli.createTempFeed(feedutil.makeFeed(
107 """<p class="p-name">This is a link: http://example.org/blah</p>
108 <a class="u-url" href="/01234.html">permalink</a>"""))
109
110 cli.setFeedConfig('feed', feed)
111 ctx, _ = cli.run('process')
112 post = ctx.silos[0].client.posts[0]
113 assert post[0] == "This is a link: http://example.org/blah"
114 assert post[2] == None
115
116 feed = cli.createTempFeed(feedutil.makeFeed(
117 """<p class="e-content">This is another link: <a href="http://example.org/blah">http://example.org/blah</a></p>
118 <a class="u-url" href="/01234.html">permalink</a>""")) # NOQA
119 cli.setFeedConfig('feed', feed)
120 ctx, _ = cli.run('process')
121 post = ctx.silos[0].client.posts[0]
122 assert post[0] == "This is another link: http://example.org/blah" # NOQA
123 facet = _make_link_facet('http://example.org/blah', 22, 45)
124 assert post[2] == [facet]
125
126 feed = cli.createTempFeed(feedutil.makeFeed(
127 """<p class="e-content">This is yet <a href="http://example.org/blah">another link</a></p>
128 <a class="u-url" href="/01234.html">permalink</a>""")) # NOQA
129 cli.setFeedConfig('feed', feed)
130 ctx, _ = cli.run('process')
131 post = ctx.silos[0].client.posts[0]
132 assert post[0] == "This is yet another link" # NOQA
133 facet = _make_link_facet('http://example.org/blah', 12, 24)
134 assert post[2] == [facet]
135
136
137 def _make_link_facet(url, start, end):
138 return atprotomodels.AppBskyRichtextFacet.Main(
139 features=[atprotomodels.AppBskyRichtextFacet.Link(uri=url)],
140 index=atprotomodels.AppBskyRichtextFacet.ByteSlice(
141 byteStart=start, byteEnd=end),
142 )
143
144
145 def _patched_media_callback(self, tmpfile, mt, url, desc):
146 return self.client.upload_blob(tmpfile, desc)
147
148
149 @pytest.fixture(scope='session')
150 def bskymock():
151 from silorider.silos.bluesky import BlueskySilo
152 BlueskySilo._CLIENT_CLASS = BlueskyMock
153 return BlueskyMockUtil()
154
155
156 class BlueskyMock:
157 def __init__(self, base_url):
158 # base_url is unused here.
159 self.posts = []
160 self.blobs = []
161
162 def login(self, email, password):
163 assert email == 'TEST_EMAIL'
164 assert password == 'TEST_PASSWORD'
165
166 def upload_blob(self, tmpfile, desc):
167 self.blobs.append((tmpfile, desc))
168 return len(self.blobs)
169
170 def send_post(self, text, embed=None, facets=None):
171 self.posts.append((text, embed, facets))
172
173
174 class BlueskyMockUtil:
175 def installCredentials(self, cli, silo_name):
176 def do_install_credentials(ctx):
177 ctx.cache.setCustomValue(
178 '%s_email' % silo_name,
179 'TEST_EMAIL')
180 ctx.cache.setCustomValue(
181 '%s_password' % silo_name,
182 'TEST_PASSWORD')
183
184 cli.preExecHook(do_install_credentials)