mirror
/
bridgy-fed
kopia lustrzana https://github.com/snarfed/bridgy-fed
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

bluesky: implement app.bsky.graph.getFollowers

pull/371/head
Ryan Barrett 2023-01-13 11:40:52 -08:00
rodzic 84163938f3
commit 898b8545ac
Nie znaleziono w bazie danych klucza dla tego podpisu
ID klucza GPG: 6BE31FDF4776E9D4
3 zmienionych plików z 111 dodań i 47 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

13
models.py
Wyświetl plik

@ -112,6 +112,11 @@ class User(StringIdModel):
base64_to_long(str(self.private_exponent))))
return rsa.exportKey(format='PEM')
def to_as1(self):
"""Returns this user as an AS1 actor dict, if possible."""
if self.actor_as2:
return as2.to_as1(json_loads(self.actor_as2))
def username(self):
"""Returns the user's preferred username from an acct: url, if available.
@ -357,3 +362,11 @@ class Follower(StringIdModel):
setattr(follower, prop, val)
follower.put()
return follower
def to_as1(self):
"""Returns this follower as an AS1 actor dict, if possible."""
if self.last_follow:
last_follow = json_loads(self.last_follow)
actor = last_follow.get('actor')
if actor:
return as2.to_as1(actor)

114
tests/test_xrpc_graph.py
Wyświetl plik

@ -1,60 +1,86 @@
"""Unit tests for graph.py."""
import copy
from unittest.mock import patch
from oauth_dropins.webutil import util
from granary import bluesky
from oauth_dropins.webutil.testutil import requests_response
from oauth_dropins.webutil.util import json_dumps, json_loads
import requests
from .test_activitypub import FOLLOW_WITH_ACTOR
from . import testutil
from models import Follower
ACTOR_DECLARATION = {
'$type': 'app.bsky.system.declRef',
'actorType': 'app.bsky.system.actorUser',
'cid': 'TODO',
}
SUBJECT = {
'$type': 'app.bsky.actor.ref#withInfo',
'did': 'did:web:foo.com',
'handle': 'foo.com',
'declaration': ACTOR_DECLARATION,
}
FOLLOWERS_BSKY = [{
'$type': 'app.bsky.graph.getFollowers#follower',
'did': 'did:web:other',
'handle': 'other-username',
'declaration': ACTOR_DECLARATION,
'indexedAt': '2022-01-02T03:04:05+00:00',
}, {
'$type': 'app.bsky.graph.getFollowers#follower',
'did': 'did:web:mastodon.social:users:swentel',
'handle': 'mastodon.social/users/swentel',
'declaration': ACTOR_DECLARATION,
'indexedAt': '2022-01-02T03:04:05+00:00',
}]
# @patch('requests.get')
# class XrpcGraphTest(testutil.TestCase):
@patch('requests.get')
class XrpcGraphTest(testutil.TestCase):
# def test_getAuthorFeed(self, mock_get):
# mock_get.return_value = requests_response("""
# <body>
# </body>
# """, url='https://foo.com/')
def test_getFollowers_not_domain(self, mock_get):
resp = self.client.get('/xrpc/app.bsky.graph.getFollowers',
query_string={'user': 'not a domain'})
self.assertEqual(400, resp.status_code)
# got = self.client.get('/xrpc/app.bsky.actor.getProfile',
# query_string={'actor': 'foo.com'},
# ).json
# self.assertEqual({
# }, got)
def test_getFollowers_empty(self, mock_get):
resp = self.client.get('/xrpc/app.bsky.graph.getFollowers',
query_string={'user': 'foo.com'})
self.assertEqual(200, resp.status_code)
self.assert_equals({
'subject': SUBJECT,
'cursor': '',
'followers': [],
}, resp.json)
# def test_getPostThread(self, mock_get):
# mock_get.return_value = requests_response("""
# <body>
# </body>
# """, url='https://foo.com/')
def test_getFollowers(self, mock_get):
Follower.get_or_create('foo.com', 'https://no/stored/follow')
Follower.get_or_create('foo.com', 'https://masto/user',
last_follow=json_dumps(FOLLOW_WITH_ACTOR))
# got = self.client.get('/xrpc/app.bsky.actor.getProfile',
# query_string={'actor': 'foo.com'},
# ).json
# self.assertEqual({
# }, got)
other = copy.deepcopy(FOLLOW_WITH_ACTOR)
other['actor'].update({
'url': 'http://other',
'preferredUsername': 'other-username',
})
Follower.get_or_create('foo.com', 'http://other', last_follow=json_dumps(other))
# def test_getRepostedBy(self, mock_get):
# mock_get.return_value = requests_response("""
# <body>
# </body>
# """, url='https://foo.com/')
other['actor']['url'] = 'http://nope'
Follower.get_or_create('nope.com', 'http://nope' , last_follow=json_dumps(other))
# got = self.client.get('/xrpc/app.bsky.actor.getProfile',
# query_string={'actor': 'foo.com'},
# ).json
# self.assertEqual({
# }, got)
resp = self.client.get('/xrpc/app.bsky.graph.getFollowers',
query_string={'user': 'foo.com'})
self.assertEqual(200, resp.status_code)
self.assert_equals({
'subject': SUBJECT,
'cursor': '',
'followers': FOLLOWERS_BSKY,
}, resp.json)
# def test_getTimeline(self, mock_get):
# mock_get.return_value = requests_response("""
# <body>
# </body>
# """, url='https://foo.com/')
# got = self.client.get('/xrpc/app.bsky.actor.getProfile',
# query_string={'actor': 'foo.com'},
# ).json
# self.assertEqual({
# }, got)
# def test_getFollows(self, mock_get):
# resp = self.client.get('/xrpc/app.bsky.graph.getFollows',
# query_string={'user': 'foo.com'})
# self.assertEqual({
# }, resp.json)

31
xrpc_graph.py
Wyświetl plik

@ -1,20 +1,45 @@
"""app.bsky.graph.* XRPC methods."""
import logging
import re
from granary import bluesky
from oauth_dropins.webutil import util
from app import xrpc_server
from models import Follower
logger = logging.getLogger(__name__)
# get these from datastore
@xrpc_server.method('app.bsky.graph.getFollowers')
def getFollowers(input):
def getFollowers(input, user=None, limit=50, before=None):
"""
lexicons/app/bsky/graph/getFollowers.json
"""
# TODO: what is user?
if not re.match(util.DOMAIN_RE, user):
raise ValueError(f'{user} is not a domain')
followers = []
for follower in Follower.query(Follower.dest == user).fetch(limit):
actor = follower.to_as1()
print('@', actor)
if actor:
followers.append({
**bluesky.actor_to_ref(actor),
'$type': 'app.bsky.graph.getFollowers#follower',
'indexedAt': util.now().isoformat(),
})
return {
'subject': bluesky.actor_to_ref({'url': f'https://{user}/'}),
'followers': followers,
'cursor': '',
}
@xrpc_server.method('app.bsky.graph.getFollows')
def getFollows(input):
def getFollows(input, user=None, limit=None, before=None):
"""
lexicons/app/bsky/graph/getFollows.json
"""