123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604 |
- import os
- import time
- from typing import List
- from typing import Tuple
- import requests
- from html2text import html2text
- from little_boxes.collection import parse_collection
- def resp2plaintext(resp):
- """Convert the body of a requests reponse to plain text in order to make basic assertions."""
- return html2text(resp.text)
- class Instance(object):
- """Test instance wrapper."""
- def __init__(self, name, host_url, docker_url=None):
- self.host_url = host_url
- self.docker_url = docker_url or host_url
- self._create_delay = 10
- with open(
- os.path.join(
- os.path.dirname(os.path.abspath(__file__)),
- f"fixtures/{name}/config/admin_api_key.key",
- )
- ) as f:
- api_key = f.read()
- self._auth_headers = {"Authorization": f"Bearer {api_key}"}
- def _do_req(self, url):
- """Used to parse collection."""
- url = url.replace(self.docker_url, self.host_url)
- resp = requests.get(url, headers={"Accept": "application/activity+json"})
- resp.raise_for_status()
- return resp.json()
- def _parse_collection(self, payload=None, url=None):
- """Parses a collection (go through all the pages)."""
- return parse_collection(url=url, payload=payload, fetcher=self._do_req)
- def ping(self):
- """Ensures the homepage is reachable."""
- resp = requests.get(f"{self.host_url}/")
- resp.raise_for_status()
- assert resp.status_code == 200
- def debug(self):
- """Returns the debug infos (number of items in the inbox/outbox."""
- resp = requests.get(
- f"{self.host_url}/api/debug",
- headers={**self._auth_headers, "Accept": "application/json"},
- )
- resp.raise_for_status()
- return resp.json()
- def drop_db(self):
- """Drops the MongoDB DB."""
- resp = requests.delete(
- f"{self.host_url}/api/debug",
- headers={**self._auth_headers, "Accept": "application/json"},
- )
- resp.raise_for_status()
- return resp.json()
- def block(self, actor_url) -> None:
- """Blocks an actor."""
- # Instance1 follows instance2
- resp = requests.post(
- f"{self.host_url}/api/block",
- params={"actor": actor_url},
- headers=self._auth_headers,
- )
- assert resp.status_code == 201
- # We need to wait for the Follow/Accept dance
- time.sleep(self._create_delay / 2)
- return resp.json().get("activity")
- def follow(self, instance: "Instance") -> str:
- """Follows another instance."""
- # Instance1 follows instance2
- resp = requests.post(
- f"{self.host_url}/api/follow",
- json={"actor": instance.docker_url},
- headers=self._auth_headers,
- )
- assert resp.status_code == 201
- # We need to wait for the Follow/Accept dance
- time.sleep(self._create_delay)
- return resp.json().get("activity")
- def new_note(self, content, reply=None) -> str:
- """Creates a new note."""
- params = {"content": content}
- if reply:
- params["reply"] = reply
- resp = requests.post(
- f"{self.host_url}/api/new_note", json=params, headers=self._auth_headers
- )
- assert resp.status_code == 201
- time.sleep(self._create_delay)
- return resp.json().get("activity")
- def boost(self, oid: str) -> str:
- """Creates an Announce activity."""
- resp = requests.post(
- f"{self.host_url}/api/boost", json={"id": oid}, headers=self._auth_headers
- )
- assert resp.status_code == 201
- time.sleep(self._create_delay)
- return resp.json().get("activity")
- def like(self, oid: str) -> str:
- """Creates a Like activity."""
- resp = requests.post(
- f"{self.host_url}/api/like", json={"id": oid}, headers=self._auth_headers
- )
- assert resp.status_code == 201
- time.sleep(self._create_delay)
- return resp.json().get("activity")
- def delete(self, oid: str) -> str:
- """Creates a Delete activity."""
- resp = requests.post(
- f"{self.host_url}/api/note/delete",
- json={"id": oid},
- headers=self._auth_headers,
- )
- assert resp.status_code == 201
- time.sleep(self._create_delay)
- return resp.json().get("activity")
- def undo(self, oid: str) -> str:
- """Creates a Undo activity."""
- resp = requests.post(
- f"{self.host_url}/api/undo", json={"id": oid}, headers=self._auth_headers
- )
- assert resp.status_code == 201
- # We need to wait for the Follow/Accept dance
- time.sleep(self._create_delay)
- return resp.json().get("activity")
- def followers(self) -> List[str]:
- """Parses the followers collection."""
- resp = requests.get(
- f"{self.host_url}/followers",
- headers={"Accept": "application/activity+json"},
- )
- resp.raise_for_status()
- data = resp.json()
- return self._parse_collection(payload=data)
- def following(self):
- """Parses the following collection."""
- resp = requests.get(
- f"{self.host_url}/following",
- headers={"Accept": "application/activity+json"},
- )
- resp.raise_for_status()
- data = resp.json()
- return self._parse_collection(payload=data)
- def outbox(self):
- """Returns the instance outbox."""
- resp = requests.get(
- f"{self.host_url}/following",
- headers={"Accept": "application/activity+json"},
- )
- resp.raise_for_status()
- return resp.json()
- def outbox_get(self, aid):
- """Fetches a specific item from the instance outbox."""
- resp = requests.get(
- aid.replace(self.docker_url, self.host_url),
- headers={"Accept": "application/activity+json"},
- )
- resp.raise_for_status()
- return resp.json()
- def stream_jsonfeed(self):
- """Returns the "stream"'s JSON feed."""
- resp = requests.get(
- f"{self.host_url}/api/stream",
- headers={**self._auth_headers, "Accept": "application/json"},
- )
- resp.raise_for_status()
- return resp.json()
- def _instances() -> Tuple[Instance, Instance]:
- """Initializes the client for the two test instances."""
- instance1 = Instance("instance1", "http://docker:5006", "http://instance1_web:5005")
- instance1.ping()
- instance2 = Instance("instance2", "http://docker:5007", "http://instance2_web:5005")
- instance2.ping()
- # Return the DB
- instance1.drop_db()
- instance2.drop_db()
- return instance1, instance2
- def test_follow() -> None:
- """instance1 follows instance2."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 1 # An Accept activity should be there
- assert instance1_debug["outbox"] == 1 # We've sent a Follow activity
- instance2_debug = instance2.debug()
- assert instance2_debug["inbox"] == 1 # An Follow activity should be there
- assert instance2_debug["outbox"] == 1 # We've sent a Accept activity
- assert instance2.followers() == [instance1.docker_url]
- assert instance1.following() == [instance2.docker_url]
- def test_follow_unfollow():
- """instance1 follows instance2, then unfollows it."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- follow_id = instance1.follow(instance2)
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 1 # An Accept activity should be there
- assert instance1_debug["outbox"] == 1 # We've sent a Follow activity
- instance2_debug = instance2.debug()
- assert instance2_debug["inbox"] == 1 # An Follow activity should be there
- assert instance2_debug["outbox"] == 1 # We've sent a Accept activity
- assert instance2.followers() == [instance1.docker_url]
- assert instance1.following() == [instance2.docker_url]
- instance1.undo(follow_id)
- assert instance2.followers() == []
- assert instance1.following() == []
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 1 # An Accept activity should be there
- assert instance1_debug["outbox"] == 2 # We've sent a Follow and a Undo activity
- instance2_debug = instance2.debug()
- assert instance2_debug["inbox"] == 2 # An Follow and Undo activity should be there
- assert instance2_debug["outbox"] == 1 # We've sent a Accept activity
- def test_post_content():
- """Instances follow each other, and instance1 creates a note."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 0
- create_id = instance1.new_note("hello")
- instance2_debug = instance2.debug()
- assert (
- instance2_debug["inbox"] == 3
- ) # An Follow, Accept and Create activity should be there
- assert instance2_debug["outbox"] == 2 # We've sent a Accept and a Follow activity
- # Ensure the post is visible in instance2's stream
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 1
- assert inbox_stream["items"][0]["id"] == create_id
- def test_block_and_post_content():
- """Instances follow each other, instance2 blocks instance1, instance1 creates a new note."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 0
- instance2.block(instance1.docker_url)
- instance1.new_note("hello")
- instance2_debug = instance2.debug()
- assert (
- instance2_debug["inbox"] == 2
- ) # An Follow, Accept activity should be there, Create should have been dropped
- assert (
- instance2_debug["outbox"] == 3
- ) # We've sent a Accept and a Follow activity + the Block activity
- # Ensure the post is not visible in instance2's stream
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 0
- def test_post_content_and_delete():
- """Instances follow each other, instance1 creates a new note, then deletes it."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 0
- create_id = instance1.new_note("hello")
- instance2_debug = instance2.debug()
- assert (
- instance2_debug["inbox"] == 3
- ) # An Follow, Accept and Create activity should be there
- assert instance2_debug["outbox"] == 2 # We've sent a Accept and a Follow activity
- # Ensure the post is visible in instance2's stream
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 1
- assert inbox_stream["items"][0]["id"] == create_id
- instance1.delete(f"{create_id}/activity")
- instance2_debug = instance2.debug()
- assert (
- instance2_debug["inbox"] == 4
- ) # An Follow, Accept and Create and Delete activity should be there
- assert instance2_debug["outbox"] == 2 # We've sent a Accept and a Follow activity
- # Ensure the post has been delete from instance2's stream
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 0
- def test_post_content_and_like():
- """Instances follow each other, instance1 creates a new note, instance2 likes it."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- create_id = instance1.new_note("hello")
- # Ensure the post is visible in instance2's stream
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 1
- assert inbox_stream["items"][0]["id"] == create_id
- # Now, instance2 like the note
- like_id = instance2.like(f"{create_id}/activity")
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 3 # Follow, Accept and Like
- assert instance1_debug["outbox"] == 3 # Folllow, Accept, and Create
- note = instance1.outbox_get(f"{create_id}/activity")
- assert "likes" in note
- assert note["likes"]["totalItems"] == 1
- likes = instance1._parse_collection(url=note["likes"]["first"])
- assert len(likes) == 1
- assert likes[0]["id"] == like_id
- def test_post_content_and_like_unlike() -> None:
- """Instances follow each other, instance1 creates a new note, instance2 likes it, then unlikes it."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- create_id = instance1.new_note("hello")
- # Ensure the post is visible in instance2's stream
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 1
- assert inbox_stream["items"][0]["id"] == create_id
- # Now, instance2 like the note
- like_id = instance2.like(f"{create_id}/activity")
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 3 # Follow, Accept and Like
- assert instance1_debug["outbox"] == 3 # Folllow, Accept, and Create
- note = instance1.outbox_get(f"{create_id}/activity")
- assert "likes" in note
- assert note["likes"]["totalItems"] == 1
- likes = instance1._parse_collection(url=note["likes"]["first"])
- assert len(likes) == 1
- assert likes[0]["id"] == like_id
- instance2.undo(like_id)
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 4 # Follow, Accept and Like and Undo
- assert instance1_debug["outbox"] == 3 # Folllow, Accept, and Create
- note = instance1.outbox_get(f"{create_id}/activity")
- assert "likes" in note
- assert note["likes"]["totalItems"] == 0
- def test_post_content_and_boost() -> None:
- """Instances follow each other, instance1 creates a new note, instance2 "boost" it."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- create_id = instance1.new_note("hello")
- # Ensure the post is visible in instance2's stream
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 1
- assert inbox_stream["items"][0]["id"] == create_id
- # Now, instance2 like the note
- boost_id = instance2.boost(f"{create_id}/activity")
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 3 # Follow, Accept and Announce
- assert instance1_debug["outbox"] == 3 # Folllow, Accept, and Create
- note = instance1.outbox_get(f"{create_id}/activity")
- assert "shares" in note
- assert note["shares"]["totalItems"] == 1
- shares = instance1._parse_collection(url=note["shares"]["first"])
- assert len(shares) == 1
- assert shares[0]["id"] == boost_id
- def test_post_content_and_boost_unboost() -> None:
- """Instances follow each other, instance1 creates a new note, instance2 "boost" it, then "unboost" it."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- create_id = instance1.new_note("hello")
- # Ensure the post is visible in instance2's stream
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 1
- assert inbox_stream["items"][0]["id"] == create_id
- # Now, instance2 like the note
- boost_id = instance2.boost(f"{create_id}/activity")
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 3 # Follow, Accept and Announce
- assert instance1_debug["outbox"] == 3 # Folllow, Accept, and Create
- note = instance1.outbox_get(f"{create_id}/activity")
- assert "shares" in note
- assert note["shares"]["totalItems"] == 1
- shares = instance1._parse_collection(url=note["shares"]["first"])
- assert len(shares) == 1
- assert shares[0]["id"] == boost_id
- instance2.undo(boost_id)
- instance1_debug = instance1.debug()
- assert instance1_debug["inbox"] == 4 # Follow, Accept and Announce and Undo
- assert instance1_debug["outbox"] == 3 # Folllow, Accept, and Create
- note = instance1.outbox_get(f"{create_id}/activity")
- assert "shares" in note
- assert note["shares"]["totalItems"] == 0
- def test_post_content_and_post_reply() -> None:
- """Instances follow each other, instance1 creates a new note, instance2 replies to it."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 0
- instance1_create_id = instance1.new_note("hello")
- instance2_debug = instance2.debug()
- assert (
- instance2_debug["inbox"] == 3
- ) # An Follow, Accept and Create activity should be there
- assert instance2_debug["outbox"] == 2 # We've sent a Accept and a Follow activity
- # Ensure the post is visible in instance2's stream
- instance2_inbox_stream = instance2.stream_jsonfeed()
- assert len(instance2_inbox_stream["items"]) == 1
- assert instance2_inbox_stream["items"][0]["id"] == instance1_create_id
- instance2_create_id = instance2.new_note(
- f"hey @instance1@{instance1.docker_url}",
- reply=f"{instance1_create_id}/activity",
- )
- instance2_debug = instance2.debug()
- assert (
- instance2_debug["inbox"] == 3
- ) # An Follow, Accept and Create activity should be there
- assert (
- instance2_debug["outbox"] == 3
- ) # We've sent a Accept and a Follow and a Create activity
- instance1_debug = instance1.debug()
- assert (
- instance1_debug["inbox"] == 3
- ) # An Follow, Accept and Create activity should be there
- assert (
- instance1_debug["outbox"] == 3
- ) # We've sent a Accept and a Follow and a Create activity
- instance1_inbox_stream = instance1.stream_jsonfeed()
- assert len(instance1_inbox_stream["items"]) == 1
- assert instance1_inbox_stream["items"][0]["id"] == instance2_create_id
- instance1_note = instance1.outbox_get(f"{instance1_create_id}/activity")
- assert "replies" in instance1_note
- assert instance1_note["replies"]["totalItems"] == 1
- replies = instance1._parse_collection(url=instance1_note["replies"]["first"])
- assert len(replies) == 1
- assert replies[0]["id"] == f"{instance2_create_id}/activity"
- def test_post_content_and_post_reply_and_delete() -> None:
- """Instances follow each other, instance1 creates a new note, instance2 replies to it, then deletes its reply."""
- instance1, instance2 = _instances()
- # Instance1 follows instance2
- instance1.follow(instance2)
- instance2.follow(instance1)
- inbox_stream = instance2.stream_jsonfeed()
- assert len(inbox_stream["items"]) == 0
- instance1_create_id = instance1.new_note("hello")
- instance2_debug = instance2.debug()
- assert (
- instance2_debug["inbox"] == 3
- ) # An Follow, Accept and Create activity should be there
- assert instance2_debug["outbox"] == 2 # We've sent a Accept and a Follow activity
- # Ensure the post is visible in instance2's stream
- instance2_inbox_stream = instance2.stream_jsonfeed()
- assert len(instance2_inbox_stream["items"]) == 1
- assert instance2_inbox_stream["items"][0]["id"] == instance1_create_id
- instance2_create_id = instance2.new_note(
- f"hey @instance1@{instance1.docker_url}",
- reply=f"{instance1_create_id}/activity",
- )
- instance2_debug = instance2.debug()
- assert (
- instance2_debug["inbox"] == 3
- ) # An Follow, Accept and Create activity should be there
- assert (
- instance2_debug["outbox"] == 3
- ) # We've sent a Accept and a Follow and a Create activity
- instance1_debug = instance1.debug()
- assert (
- instance1_debug["inbox"] == 3
- ) # An Follow, Accept and Create activity should be there
- assert (
- instance1_debug["outbox"] == 3
- ) # We've sent a Accept and a Follow and a Create activity
- instance1_inbox_stream = instance1.stream_jsonfeed()
- assert len(instance1_inbox_stream["items"]) == 1
- assert instance1_inbox_stream["items"][0]["id"] == instance2_create_id
- instance1_note = instance1.outbox_get(f"{instance1_create_id}/activity")
- assert "replies" in instance1_note
- assert instance1_note["replies"]["totalItems"] == 1
- instance2.delete(f"{instance2_create_id}/activity")
- instance1_debug = instance1.debug()
- assert (
- instance1_debug["inbox"] == 4
- ) # An Follow, Accept and Create and Delete activity should be there
- assert (
- instance1_debug["outbox"] == 3
- ) # We've sent a Accept and a Follow and a Create activity
- instance1_note = instance1.outbox_get(f"{instance1_create_id}/activity")
- assert "replies" in instance1_note
- assert instance1_note["replies"]["totalItems"] == 0
|