|
46
|
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)
|
