kopia lustrzana https://github.com/tsileo/little-boxes
Tweak the test suite
rodzic
4ff4b3a85c
commit
587e580755
|
@ -1,3 +1,3 @@
|
|||
VERSION = (0, 1, 0)
|
||||
VERSION = (0, 1, 0) # pragma: no cover
|
||||
|
||||
__version__ = ".".join(map(str, VERSION))
|
||||
__version__ = ".".join(map(str, VERSION)) # pragma: no cover
|
||||
|
|
|
@ -973,10 +973,6 @@ class Note(BaseActivity):
|
|||
OBJECT_REQURIED = False
|
||||
|
||||
def _init(self, **kwargs):
|
||||
print(self._data)
|
||||
# Remove the `actor` field as `attributedTo` is used for `Note` instead
|
||||
if "actor" in self._data:
|
||||
del (self._data["actor"])
|
||||
if "sensitive" not in kwargs:
|
||||
self._data["sensitive"] = False
|
||||
|
||||
|
|
|
@ -1,7 +1,5 @@
|
|||
import abc
|
||||
import typing
|
||||
from typing import Any
|
||||
from typing import Dict
|
||||
|
||||
import requests
|
||||
|
||||
|
@ -15,110 +13,110 @@ class Backend(abc.ABC):
|
|||
def user_agent(self) -> str:
|
||||
return f"Little Boxes {__version__} (+http://github.com/tsileo/little-boxes)"
|
||||
|
||||
def fetch_json(self, url: str) -> Dict[str, Any]:
|
||||
def fetch_json(self, url: str, **kwargs):
|
||||
resp = requests.get(
|
||||
url, headers={"User-Agent": self.user_agent(), "Accept": "application/json"}
|
||||
url,
|
||||
headers={"User-Agent": self.user_agent(), "Accept": "application/json"},
|
||||
**kwargs,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
|
||||
return resp.json()
|
||||
return resp
|
||||
|
||||
@abc.abstractmethod
|
||||
def base_url(self) -> str:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def fetch_iri(self, iri: str) -> "ap.ObjectType":
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def activity_url(self, obj_id: str) -> str:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_create(self, as_actor: "ap.Person", activity: "ap.Create") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_delete(self, as_actor: "ap.Person", activity: "ap.Delete") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def inbox_create(self, as_actor: "ap.Person", activity: "ap.Create") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def inbox_delete(self, as_actor: "ap.Person", activity: "ap.Delete") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_is_blocked(self, as_actor: "ap.Person", actor_id: str) -> bool:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def inbox_new(self, as_actor: "ap.Person", activity: "ap.BaseActivity") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_new(self, as_actor: "ap.Person", activity: "ap.BaseActivity") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def new_follower(self, as_actor: "ap.Person", follow: "ap.Follow") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def new_following(self, as_actor: "ap.Person", follow: "ap.Follow") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def undo_new_follower(self, as_actor: "ap.Person", follow: "ap.Follow") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def undo_new_following(self, as_actor: "ap.Person", follow: "ap.Follow") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def inbox_update(self, as_actor: "ap.Person", activity: "ap.Update") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_update(self, as_actor: "ap.Person", activity: "ap.Update") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def inbox_like(self, as_actor: "ap.Person", activity: "ap.Like") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def inbox_undo_like(self, as_actor: "ap.Person", activity: "ap.Like") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_like(self, as_actor: "ap.Person", activity: "ap.Like") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_undo_like(self, as_actor: "ap.Person", activity: "ap.Like") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def inbox_announce(self, as_actor: "ap.Person", activity: "ap.Announce") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def inbox_undo_announce(
|
||||
self, as_actor: "ap.Person", activity: "ap.Announce"
|
||||
) -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_announce(self, as_actor: "ap.Person", activity: "ap.Announce") -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
@abc.abstractmethod
|
||||
def outbox_undo_announce(
|
||||
self, as_actor: "ap.Person", activity: "ap.Announce"
|
||||
) -> None:
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
|
|
@ -21,7 +21,7 @@ def is_url_valid(url: str) -> bool:
|
|||
|
||||
# XXX in debug mode, we want to allow requests to localhost to test the federation with local instances
|
||||
debug_mode = strtobool(os.getenv("MICROBLOGPUB_DEBUG", "false"))
|
||||
if debug_mode:
|
||||
if debug_mode: # pragma: no cover
|
||||
return True
|
||||
|
||||
if parsed.hostname in ["localhost"]:
|
||||
|
|
|
@ -6,6 +6,7 @@ from urllib.parse import urlparse
|
|||
|
||||
import requests
|
||||
|
||||
from .activitypub import get_backend
|
||||
from .urlutils import check_url
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
@ -35,8 +36,8 @@ def webfinger(resource: str) -> Optional[Dict[str, Any]]:
|
|||
for i, proto in enumerate(protos):
|
||||
try:
|
||||
url = f"{proto}://{host}/.well-known/webfinger"
|
||||
# FIXME(tsileo): BACKEND.do_req so we can set a UserAgent
|
||||
resp = requests.get(url, {"resource": resource})
|
||||
# FIXME(tsileo): BACKEND.fetch_json so we can set a UserAgent
|
||||
resp = get_backend().fetch_json(url, params={"resource": resource})
|
||||
except requests.ConnectionError:
|
||||
# If we tried https first and the domain is "http only"
|
||||
if i == 0:
|
||||
|
|
|
@ -495,6 +495,44 @@ def test_little_boxes_follow_and_new_create_note_and_delete():
|
|||
)
|
||||
|
||||
|
||||
def test_little_boxes_follow_and_new_create_note_and_update():
|
||||
back, create = test_little_boxes_follow_and_new_create_note()
|
||||
|
||||
me = back.get_user("tom")
|
||||
other = back.get_user("tom2")
|
||||
|
||||
outbox = ap.Outbox(other)
|
||||
|
||||
update = ap.Update(
|
||||
actor=other.id,
|
||||
object={
|
||||
"content": "Hello2",
|
||||
"id": create.get_object().id,
|
||||
"type": "Note",
|
||||
"attributedTo": other.id,
|
||||
},
|
||||
)
|
||||
outbox.post(update)
|
||||
|
||||
back.assert_called_methods(
|
||||
other,
|
||||
(
|
||||
"an Delete activity is published",
|
||||
"outbox_new",
|
||||
lambda as_actor: _assert_eq(as_actor.id, other.id),
|
||||
lambda activity: _assert_eq(activity.id, update.id),
|
||||
),
|
||||
(
|
||||
'"outbox_update" hook is called',
|
||||
"outbox_update",
|
||||
lambda as_actor: _assert_eq(as_actor.id, other.id),
|
||||
lambda _update: _assert_eq(_update.id, update.id),
|
||||
),
|
||||
)
|
||||
|
||||
back.assert_called_methods(me)
|
||||
|
||||
|
||||
def test_little_boxes_follow_and_new_create_note_and_like():
|
||||
back, create = test_little_boxes_follow_and_new_create_note()
|
||||
|
||||
|
|
Ładowanie…
Reference in New Issue