From d330f0b2498d83686b39a161e32b003682a8d14c Mon Sep 17 00:00:00 2001 From: Johannes Ernst Date: Thu, 17 Oct 2024 12:42:11 -0700 Subject: [PATCH] Reorder methods on FediverseNode for better comprehensibility Add some FediverseNode methods for next set of tests Rename account_id_*validate to acct_uri_*validate for consistency Add list-of validation functions --- .../nodedrivers/fallback/fediverse.py | 130 ++++++--- src/feditest/nodedrivers/mastodon/__init__.py | 273 +++++++++++------- src/feditest/protocols/fediverse/__init__.py | 137 ++++++--- src/feditest/utils.py | 20 +- tests.smoke/tests/node_with_mastodon_api.py | 2 +- .../nodes_with_mastodon_api_communicate.py | 2 +- 6 files changed, 377 insertions(+), 187 deletions(-) diff --git a/src/feditest/nodedrivers/fallback/fediverse.py b/src/feditest/nodedrivers/fallback/fediverse.py index c9561fb..f9fce3b 100644 --- a/src/feditest/nodedrivers/fallback/fediverse.py +++ b/src/feditest/nodedrivers/fallback/fediverse.py @@ -27,7 +27,17 @@ FediverseNonExistingAccount ) from feditest.testplan import TestPlanConstellationNode, TestPlanNodeAccountField, TestPlanNodeNonExistingAccountField -from feditest.utils import appname_validate, boolean_parse_validate, hostname_validate, https_uri_validate, prompt_user, prompt_user_parse_validate +from feditest.utils import ( + acct_uri_list_validate, + acct_uri_validate, + appname_validate, + boolean_parse_validate, + hostname_validate, + https_uri_list_validate, + https_uri_validate, + prompt_user, + prompt_user_parse_validate +) class FallbackFediverseNode(FediverseNode): @@ -64,6 +74,38 @@ def obtain_non_existing_actor_acct_uri(self, rolename: str | None = None ) -> st return non_account.actor_acct_uri + # Python 3.12 @override + def make_follow(self, actor_acct_uri: str, to_follow_actor_acct_uri: str) -> None: + prompt_user( + f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" follow actor "{ to_follow_actor_acct_uri }"' + + ' and hit return when done.') + + + def make_unfollow(self, actor_acct_uri: str, following_actor_acct_uri: str) -> None: + prompt_user( + f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" unfollow actor "{ following_actor_acct_uri }"' + + ' and hit return when done.') + + + # Python 3.12 @override + def actor_is_following_actor(self, actor_acct_uri: str, leader_actor_acct_uri: str) -> bool: + answer = prompt_user_parse_validate( + f'On FediverseNode "{ self.hostname }", is actor "{ actor_acct_uri }" following actor "{ leader_actor_acct_uri }"?' + + ' Enter "true" or "false".', + parse_validate=boolean_parse_validate) + return answer + + + # Python 3.12 @override + def actor_is_followed_by_actor(self, actor_acct_uri: str, follower_actor_acct_uri: str) -> bool: + answer = prompt_user_parse_validate( + f'On FediverseNode "{ self.hostname }", is actor "{ actor_acct_uri }" being followed by actor "{ follower_actor_acct_uri }"?' + + ' Enter "true" or "false".', + parse_validate=boolean_parse_validate) + return answer + + # All other follow-related methods: We leave the NotImplementedByNodeError raised by the superclass until we have a better idea :-) + # Python 3.12 @override def make_create_note(self, actor_acct_uri: str, content: str, deliver_to: list[str] | None = None) -> str: if deliver_to : @@ -80,72 +122,92 @@ def make_create_note(self, actor_acct_uri: str, content: str, deliver_to: list[s parse_validate=https_uri_validate) + # Python 3.12 @override + def update_note(self, actor_acct_uri: str, note_uri: str, new_content: str) -> None: + prompt_user( + f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" update the note at "{ note_uri }"' + + ' with new content:"""\n{ new_content }\n"""' + + ' and hit return when done.') + # Python 3.12 @override - def make_announce_object(self, actor_acct_uri, to_be_announced_object_uri: str) -> str: - return prompt_user_parse_validate( - f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" boost "{ to_be_announced_object_uri }"' - + ' and enter the Announce object\'s local URI:', - parse_validate=https_uri_validate) + def delete_object(self, actor_acct_uri: str, object_uri: str) -> None: + prompt_user( + f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" delete the object at "{ object_uri }"' + + ' and hit return when done.') # Python 3.12 @override def make_reply_note(self, actor_acct_uri, to_be_replied_to_object_uri: str, reply_content: str) -> str: return prompt_user_parse_validate( f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" reply to object with "{ to_be_replied_to_object_uri }"' - + ' and enter the Announce object\'s URI when created.' + + ' and enter the reply note\'s URI when created.' + f' Reply content:"""\n{ reply_content }\n"""', parse_validate=https_uri_validate) # Python 3.12 @override - def make_follow(self, actor_acct_uri: str, to_follow_actor_acct_uri: str) -> None: + def like_object(self, actor_acct_uri: str, object_uri: str) -> None: prompt_user( - f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" follow actor "{ to_follow_actor_acct_uri }"' + f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" like the object at "{ object_uri }"' + + ' and hit return when done.') + + + # Python 3.12 @override + def announce_object(self, actor_acct_uri: str, object_uri: str) -> None: + prompt_user( + f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" announce/reblog/boost the object at "{ object_uri }"' + ' and hit return when done.') - # We leave the NotImplementedByNodeError raised by the superclass for all other follow-related actions - # until we have a better idea :-) # Python 3.12 @override - def actor_has_received_note(self, actor_acct_uri: str, object_uri: str) -> str | None: + def actor_has_received_object(self, actor_acct_uri: str, object_uri: str) -> str | None: answer = prompt_user( - f'On FediverseNode "{ self.hostname }", has actor "{ actor_acct_uri }" received the note "{ object_uri }"?' - + ' Enter the content of the note, or leave empty if it didn\'t happen.') - if answer: - return answer - return None + f'On FediverseNode "{ self.hostname }", has actor "{ actor_acct_uri }" received the object "{ object_uri }"?' + + ' Enter the content of the object, or leave empty if it didn\'t happen.') + return answer if answer else None # Python 3.12 @override - def actor_is_following_actor(self, actor_acct_uri: str, leader_actor_acct_uri: str) -> bool: + def note_content(self, actor_acct_uri: str, note_uri: str) -> str | None: + answer = prompt_user( + f'On FediverseNode "{ self.hostname }", have actor "{ actor_acct_uri }" access note "{ note_uri }" and enter its content.') + return answer if answer else None + + + # Python 3.12 @override + def object_author(self, actor_acct_uri: str, object_uri: str) -> str | None: answer = prompt_user_parse_validate( - f'On FediverseNode "{ self.hostname }", is actor "{ actor_acct_uri }" following actor "{ leader_actor_acct_uri }"?' - + ' Enter "true" or "false".', - parse_validate=boolean_parse_validate) + f'On FediverseNode "{ self.hostname }", have actor "{ actor_acct_uri }" access object "{ object_uri }" and enter the acct URI of the object\'s author.', + parse_validate=acct_uri_validate) return answer # Python 3.12 @override - def note_has_direct_reply(self, actor_acct_uri: str, note_uri: str, reply_uri: str) -> str | None: - answer = prompt_user( - f'On FediverseNode "{ self.hostname }", does actor "{ actor_acct_uri }" see that note "{ note_uri }" has reply "{ reply_uri }"?' - + ' Enter the reply content.') - return answer if answer else None + def direct_replies_to_object(self, actor_acct_uri: str, object_uri: str) -> list[str]: + answer = prompt_user_parse_validate( + f'On FediverseNode "{ self.hostname }", have actor "{ actor_acct_uri }" access object "{ object_uri }"' + + ' and enter the https URIs of all objects that directly reply to it (space-separated list).', + parse_validate=https_uri_list_validate) + return answer.split() # Python 3.12 @override - def access_note(self, actor_acct_uri: str, note_uri: str) -> str | None: - answer = prompt_user( - f'On FediverseNode "{ self.hostname }", have actor "{ actor_acct_uri }" access note "{ note_uri }" and enter its content.') - return answer if answer else None + def object_likers(self, actor_acct_uri: str, object_uri: str) -> list[str]: + answer = prompt_user_parse_validate( + f'On FediverseNode "{ self.hostname }", have actor "{ actor_acct_uri }" access object "{ object_uri }"' + + ' and enter the acct URIs of all accounts that like it (space-separated list).', + parse_validate=acct_uri_list_validate) + return answer.split() # Python 3.12 @override - def update_note(self, actor_acct_uri: str, note_uri: str, new_content: str) -> None: - prompt_user( - f'On FediverseNode "{ self.hostname }", make actor "{ actor_acct_uri }" update the object at "{ note_uri }"' - + ' with new content:"""\n{ new_content }\n"""') + def object_announcers(self, actor_acct_uri: str, object_uri: str) -> list[str]: + answer = prompt_user_parse_validate( + f'On FediverseNode "{ self.hostname }", have actor "{ actor_acct_uri }" access object "{ object_uri }"' + + ' and enter the acct URIs of all accounts that have announced/reblogged/boosted it (space-separated list).', + parse_validate=acct_uri_list_validate) + return answer.split() # From WebFingerServer diff --git a/src/feditest/nodedrivers/mastodon/__init__.py b/src/feditest/nodedrivers/mastodon/__init__.py index f875f53..3c91cee 100644 --- a/src/feditest/nodedrivers/mastodon/__init__.py +++ b/src/feditest/nodedrivers/mastodon/__init__.py @@ -92,12 +92,14 @@ def _password_validate(candidate: str) -> str | None: ) -def mastodon_api_invoke_get( +def mastodon_api_invoke_get_or_delete( + method: str, api_base_url: str, session: requests.Session, path: str, headers: dict[str,str] | None ) -> dict[str,Any]: + method = method.lower() url = api_base_url + path real_headers = { 'user-agent' : 'FediTest', @@ -109,13 +111,18 @@ def mastodon_api_invoke_get( if is_trace_active(): curl = f'curl { url }' + if method != 'GET': + curl += f' -X { method }' for key, value in real_headers.items(): curl += f' -H "{ key }: { value }"' trace(f'Mastodon API call as curl: { curl }') response_json = None try : - response = session.get(url, headers=real_headers) + if 'get' == method: + response = session.get(url, headers=real_headers) + else: + response = session.delete(url, headers=real_headers) if response.status_code >= 400: # taken from requests' raise_for_status() raise HTTPError(f'HTTP status { response.status_code }: { response.content.decode("utf-8") }', response=response) response_json = response.json() @@ -136,6 +143,7 @@ def mastodon_api_invoke_post_or_put( args: dict[str,str] | None = None, headers: dict[str,str] | None = None ) -> dict[str,Any]: + method = method.lower() url = api_base_url + path real_headers = { 'user-agent' : 'FediTest', @@ -156,7 +164,7 @@ def mastodon_api_invoke_post_or_put( response_json = None try : - if 'post' == method.lower(): + if 'post' == method: response = session.post(url, data=args, headers=real_headers) else: response = session.put(url, data=args, headers=real_headers) @@ -212,7 +220,7 @@ def __init__(self, app: MastodonOAuthApp, account: 'AccountOnNodeWithMastodonAPI def http_get(self, path: str) -> Any: - return mastodon_api_invoke_get(self._app.api_base_url, self._app.session, path, self._auth_header) + return mastodon_api_invoke_get_or_delete('GET', self._app.api_base_url, self._app.session, path, self._auth_header) def http_post(self, path: str, args: dict[str,str] | None = None) -> Any: @@ -223,6 +231,40 @@ def http_put(self, path: str, args: dict[str,str] | None = None) -> Any: return mastodon_api_invoke_post_or_put('PUT', self._app.api_base_url, self._app.session, path, args=args, headers=self._auth_header) + def http_delete(self, path: str) -> Any: + return mastodon_api_invoke_get_or_delete('DELETE', self._app.api_base_url, self._app.session, path, self._auth_header) + + + def make_follow(self, to_follow_actor_acct_uri: str) -> dict[str, str]: + if to_follow_account := self._find_account_dict_by_other_actor_acct_uri(to_follow_actor_acct_uri): + local_to_follow_account_id = to_follow_account['id'] + response = self.http_post(f'/api/v1/accounts/{ local_to_follow_account_id }/follow') + return response + raise ValueError(f'Cannot find account for Actor on { self }: "{ to_follow_actor_acct_uri }"') + + + def make_unfollow(self, following_actor_acct_uri: str) -> dict[str,str]: + if following_account := self._find_account_dict_by_other_actor_acct_uri(following_actor_acct_uri): + following_account_id = following_account['id'] + response = self.http_post(f'/api/v1/accounts/{ following_account_id }/unfollow') + return response + raise ValueError(f'Account not found with Actor URI: { following_actor_acct_uri }') + + + def actor_is_following_actor(self, leader_actor_acct_uri: str) -> bool: + this_account_id = self._account.internal_userid + response = self.http_get(f'/api/v1/accounts/{ this_account_id }/following') + found = find_first_in_array(response, lambda r: r['acct'] == leader_actor_acct_uri[5:]) # remove acct: + return found is not None + + + def actor_is_followed_by_actor(self, follower_actor_acct_uri: str) -> bool: + this_account_id = self._account.internal_userid + response = self.http_get(f'/api/v1/accounts/{ this_account_id }/followers') + found = find_first_in_array(response, lambda r: r['acct'] == follower_actor_acct_uri[5:]) # remove acct: + return found is not None + + def make_create_note(self, content: str, deliver_to: list[str] | None = None) -> dict[str, str]: if deliver_to: # The only way we can address specific accounts in Mastodon for to in deliver_to: @@ -239,12 +281,23 @@ def make_create_note(self, content: str, deliver_to: list[str] | None = None) -> return response - def make_announce_object(self, to_be_announced_object_uri: str) -> dict[str, str]: - if local_note := self._find_note_dict_by_uri(to_be_announced_object_uri): - local_note_id = local_note['id'] - response = self.http_post(f'/api/v1/statuses/{ local_note_id }/reblog') + def update_note(self, note_uri: str, new_content: str) -> dict[str, Any]: + if note := self._find_note_dict_by_uri(note_uri): + note_id = note['id'] + args = { + 'status' : new_content + } + response = self.http_put(f'/api/v1/statuses/{ note_id }', args) return response - raise ValueError(f'Cannot find Note on { self } : "{ to_be_announced_object_uri }"') + raise ValueError(f'Cannot find Note on { self }: "{ note_uri }"') + + + def delete_object(self, note_uri: str) -> None: + if note := self._find_note_dict_by_uri(note_uri): + note_id = note['id'] + response = self.http_delete(f'/api/v1/statuses/{ note_id }') + return response + raise ValueError(f'Cannot find Note on { self }: "{ note_uri }"') def make_reply_note(self, to_be_replied_to_object_uri: str, reply_content: str) -> dict[str, str]: @@ -260,23 +313,23 @@ def make_reply_note(self, to_be_replied_to_object_uri: str, reply_content: str) raise ValueError(f'Cannot find Note on { self }: "{ to_be_replied_to_object_uri }"') - def make_follow(self, to_follow_actor_acct_uri: str) -> dict[str, str]: - if to_follow_account := self._find_account_dict_by_other_actor_acct_uri(to_follow_actor_acct_uri): - local_to_follow_account_id = to_follow_account['id'] - response = self.http_post(f'/api/v1/accounts/{ local_to_follow_account_id }/follow') + def like_object(self, object_uri: str) -> None: + if note := self._find_note_dict_by_uri(object_uri): + note_id = note['id'] + response = self.http_post(f'/api/v1/statuses/{ note_id }/favourite') return response - raise ValueError(f'Cannot find account for Actor on { self }: "{ to_follow_actor_acct_uri }"') + raise ValueError(f'Cannot find Note on { self }: "{ object_uri }"') - def make_follow_undo(self, following_actor_acct_uri: str) -> dict[str,str]: - if following_account := self._find_account_dict_by_other_actor_acct_uri(following_actor_acct_uri): - following_account_id = following_account['id'] - response = self.http_post(f'/api/v1/accounts/{ following_account_id }/unfollow') + def announce_object(self, object_uri: str) -> None: + if note := self._find_note_dict_by_uri(object_uri): + note_id = note['id'] + response = self.http_post(f'/api/v1/statuses/{ note_id }/reblog') return response - raise ValueError(f'Account not found with Actor URI: { following_actor_acct_uri }') + raise ValueError(f'Cannot find Note on { self }: "{ object_uri }"') - def actor_has_received_note(self, object_uri: str) -> dict[str, Any]: + def actor_has_received_object(self, object_uri: str) -> dict[str, Any]: # Depending on how the Note is addressed and follow status, Mastodon puts it into the Home timeline or only # into notifications. elements = self.http_get('/api/v1/timelines/home') @@ -289,44 +342,34 @@ def actor_has_received_note(self, object_uri: str) -> dict[str, Any]: return response - def actor_is_following_actor(self, leader_actor_acct_uri: str) -> bool: - this_account_id = self._account.internal_userid - response = self.http_get(f'/api/v1/accounts/{ this_account_id }/following') - found = find_first_in_array(response, lambda r: r['acct'] == leader_actor_acct_uri[5:]) # remove acct: - return found is not None - - - def actor_is_followed_by_actor(self, follower_actor_acct_uri: str) -> bool: - this_account_id = self._account.internal_userid - response = self.http_get(f'/api/v1/accounts/{ this_account_id }/followers') - found = find_first_in_array(response, lambda r: r['acct'] == follower_actor_acct_uri[5:]) # remove acct: - return found is not None - - - def note_has_direct_reply(self, note_uri: str, reply_uri: str) -> dict[str, Any] | None: + def note_dict(self, note_uri: str) -> dict[str, Any]: if note := self._find_note_dict_by_uri(note_uri): - note_id = note['id'] - response = self.http_get(f'/api/v1/statuses/{ note_id }/context') - found = find_first_in_array(response['descendants'], lambda r: r['uri'] == reply_uri) - return found + return note raise ValueError(f'Cannot find Note on { self }: "{ note_uri }"') - def access_note(self, note_uri: str) -> dict[str, Any]: - if note := self._find_note_dict_by_uri(note_uri): - return note - raise ValueError(f'Cannot find Note on { self }: "{ note_uri }"') + def object_context(self, object_uri: str) -> dict[str,Any]: + if obj := self._find_note_dict_by_uri(object_uri): + obj_id = obj['id'] + response = self.http_get(f'/api/v1/statuses/{ obj_id }/context') + return response + raise ValueError(f'Cannot find object on { self }: "{ object_uri }"') - def update_note(self, note_uri: str, new_content: str) -> dict[str, Any]: - if note := self._find_note_dict_by_uri(note_uri): - note_id = note['id'] - args = { - 'status' : new_content - } - response = self.http_put(f'/api/v1/statuses/{ note_id }', args) + def object_likers(self, object_uri: str) -> list[dict[str, Any]]: + if obj := self._find_note_dict_by_uri(object_uri): + obj_id = obj['id'] + response = self.http_get(f'/api/v1/statuses/{ obj_id }/favourited_by') return response - raise ValueError(f'Cannot find Note on { self }: "{ note_uri }"') + raise ValueError(f'Cannot find object on { self }: "{ object_uri }"') + + + def object_announcers(self, object_uri: str) -> list[dict[str, Any]]: + if obj := self._find_note_dict_by_uri(object_uri): + obj_id = obj['id'] + response = self.http_get(f'/api/v1/statuses/{ obj_id }/reblogged_by') + return response + raise ValueError(f'Cannot find object on { self }: "{ object_uri }"') def account_dict(self) -> dict[str, Any]: @@ -560,115 +603,141 @@ def obtain_actor_acct_uri(self, rolename: str | None = None) -> str: # Python 3.12 @override - def make_create_note(self, actor_acct_uri: str, content: str, deliver_to: list[str] | None = None) -> str: - trace('make_create_note:') + def make_follow(self, actor_acct_uri: str, to_follow_actor_acct_uri: str) -> None: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - response = mastodon_client.make_create_note(content, deliver_to) + mastodon_client.make_follow(to_follow_actor_acct_uri) self._run_poor_mans_cron() - return response['uri'] # Python 3.12 @override - def make_announce_object(self, actor_acct_uri: str, to_be_announced_object_uri: str) -> str: - trace('make_announce_object:') + def set_auto_accept_follow(self, actor_acct_uri: str, auto_accept_follow: bool = True) -> None: + if auto_accept_follow: + return # Default for Mastodon + + raise NotImplementedByNodeError(self, NodeWithMastodonAPI.set_auto_accept_follow) # Can't find an API call for this + + + # Python 3.12 @override + def make_follow_accept(self, actor_acct_uri: str, follower_actor_acct_uri: str) -> None: + super().make_follow_accept(actor_acct_uri, follower_actor_acct_uri) # FIXME + + + # Python 3.12 @override + def make_follow_reject(self, actor_acct_uri: str, follower_actor_acct_uri: str) -> None: + super().make_follow_reject(actor_acct_uri, follower_actor_acct_uri) # FIXME + + + # Python 3.12 @override + def make_unfollow(self, actor_acct_uri: str, following_actor_acct_uri: str) -> None: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - response = mastodon_client.make_announce_object(to_be_announced_object_uri) + mastodon_client.make_unfollow(following_actor_acct_uri) self._run_poor_mans_cron() - return response['uri'] + + + # Python 3.12 @override + def actor_is_following_actor(self, actor_acct_uri: str, leader_actor_acct_uri: str) -> bool: + mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) + return mastodon_client.actor_is_following_actor(leader_actor_acct_uri) # Python 3.12 @override - def make_reply_note(self, actor_acct_uri: str, to_be_replied_to_object_uri: str, reply_content: str) -> str: - trace('make_reply_note:') + def actor_is_followed_by_actor(self, actor_acct_uri: str, follower_actor_acct_uri: str) -> bool: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - response = mastodon_client.make_reply_note(to_be_replied_to_object_uri, reply_content) + return mastodon_client.actor_is_followed_by_actor(follower_actor_acct_uri) + + + # Python 3.12 @override + def make_create_note(self, actor_acct_uri: str, content: str, deliver_to: list[str] | None = None) -> str: + mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) + response = mastodon_client.make_create_note(content, deliver_to) self._run_poor_mans_cron() return response['uri'] # Python 3.12 @override - def make_follow(self, actor_acct_uri: str, to_follow_actor_acct_uri: str) -> None: - trace('make_follow:') + def update_note(self, actor_acct_uri: str, note_uri: str, new_content: str) -> None: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - mastodon_client.make_follow(to_follow_actor_acct_uri) + mastodon_client.update_note(note_uri, new_content) self._run_poor_mans_cron() # Python 3.12 @override - def set_auto_accept_follow(self, actor_acct_uri: str, auto_accept_follow: bool = True) -> None: - if auto_accept_follow: - return # Default for Mastodon - - raise NotImplementedByNodeError(self, NodeWithMastodonAPI.set_auto_accept_follow) # Can't find an API call for this + def delete_object(self, actor_acct_uri: str, object_uri: str) -> None: + mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) + mastodon_client.delete_object(object_uri) + self._run_poor_mans_cron() # Python 3.12 @override - def make_follow_accept(self, actor_acct_uri: str, follower_actor_acct_uri: str) -> None: - super().make_follow_accept(actor_acct_uri, follower_actor_acct_uri) # FIXME + def make_reply_note(self, actor_acct_uri: str, to_be_replied_to_object_uri: str, reply_content: str) -> str: + mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) + response = mastodon_client.make_reply_note(to_be_replied_to_object_uri, reply_content) + self._run_poor_mans_cron() + return response['uri'] # Python 3.12 @override - def make_follow_reject(self, actor_acct_uri: str, follower_actor_acct_uri: str) -> None: - super().make_follow_reject(actor_acct_uri, follower_actor_acct_uri) # FIXME + def like_object(self, actor_acct_uri: str, object_uri: str) -> None: + mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) + mastodon_client.like_object(object_uri) + self._run_poor_mans_cron() # Python 3.12 @override - def make_follow_undo(self, actor_acct_uri: str, following_actor_acct_uri: str) -> None: - trace('make_follow_undo:') + def announce_object(self, actor_acct_uri: str, object_uri: str) -> None: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - mastodon_client.make_follow_undo(following_actor_acct_uri) + mastodon_client.announce_object(object_uri) self._run_poor_mans_cron() # Python 3.12 @override - def actor_has_received_note(self, actor_acct_uri: str, object_uri: str) -> str | None: - trace('actor_has_received_note:') + def actor_has_received_object(self, actor_acct_uri: str, object_uri: str) -> str | None: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - response = mastodon_client.actor_has_received_note(object_uri) + response = mastodon_client.actor_has_received_object(object_uri) if response: return response['content'] return None - # Python 3.12 @override - def actor_is_following_actor(self, actor_acct_uri: str, leader_actor_acct_uri: str) -> bool: - trace(f'actor_is_following_actor: actor_acct_uri = { actor_acct_uri }, leader_actor_acct_uri = { leader_actor_acct_uri }') + # Python 3.12 @override + def note_content(self, actor_acct_uri: str, note_uri: str) -> str | None: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - return mastodon_client.actor_is_following_actor(leader_actor_acct_uri) + response = mastodon_client.note_dict(note_uri) + if response: + return cast(str, response['content']) + return None # Python 3.12 @override - def actor_is_followed_by_actor(self, actor_acct_uri: str, follower_actor_acct_uri: str) -> bool: - trace(f'actor_is_followed_by_actor: actor_acct_uri = { actor_acct_uri }, follower_actor_acct_uri = { follower_actor_acct_uri }') + def object_author(self, actor_acct_uri: str, object_uri: str) -> str | None: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - return mastodon_client.actor_is_followed_by_actor(follower_actor_acct_uri) + response = mastodon_client.note_dict(object_uri) + return cast(str, response['author']['acct']) # Python 3.12 @override - def note_has_direct_reply(self, actor_acct_uri: str, note_uri: str, reply_uri: str) -> str | None: - trace(f'note_has_direct_reply: actor_acct_uri = { actor_acct_uri }, note_uri = { note_uri }, reply_uri = { reply_uri }') + def direct_replies_to_object(self, actor_acct_uri: str, object_uri: str) -> list[str]: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - response = mastodon_client.note_has_direct_reply(note_uri, reply_uri) - if response: - return cast(str, response['content']) - return None + response = mastodon_client.object_context(object_uri) + ret = [ d['uri'] for d in response['descendants']] + return ret # Python 3.12 @override - def access_note(self, actor_acct_uri: str, note_uri: str) -> str | None: - trace(f'access_note: actor_acct_uri = { actor_acct_uri }, note_uri = { note_uri }') + def object_likers(self, actor_acct_uri: str, object_uri: str) -> list[str]: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - response = mastodon_client.access_note(note_uri) - if response: - return cast(str, response['content']) - return None + response = mastodon_client.object_likers(object_uri) + ret = [ 'acct:' + x['acct'] for x in response ] + return ret # Python 3.12 @override - def update_note(self, actor_acct_uri: str, note_uri: str, new_content: str) -> None: - trace(f'update_note: actor_acct_uri = { actor_acct_uri }, note_uri = { note_uri }, new_content = { new_content }') + def object_announcers(self, actor_acct_uri: str, object_uri: str) -> list[str]: mastodon_client = self._get_mastodon_client_by_actor_acct_uri(actor_acct_uri) - mastodon_client.update_note(note_uri, new_content) + response = mastodon_client.object_announcers(object_uri) + ret = [ 'acct:' + x['acct'] for x in response ] + return ret + # From ActivityPubNode diff --git a/src/feditest/protocols/fediverse/__init__.py b/src/feditest/protocols/fediverse/__init__.py index 0cbf7e6..3d82da9 100644 --- a/src/feditest/protocols/fediverse/__init__.py +++ b/src/feditest/protocols/fediverse/__init__.py @@ -120,34 +120,7 @@ def obtain_actor_acct_uri(self, rolename: str | None = None) -> str: """ raise NotImplementedByNodeError(self, FediverseNode.obtain_actor_acct_uri) - - def make_create_note(self, actor_acct_uri: str, content: str, deliver_to: list[str] | None = None) -> str: - """" - Perform whatever actions are necessary so the actor with actor_acct_uri will have created - a Note object on this Node with the specified content. - deliver_to: make sure the Node is delivered to these Actors (i.e. in arrives in their inbox) - return: URI to the Note object - """ - raise NotImplementedByNodeError(self, FediverseNode.make_create_note) - - - def make_announce_object(self, actor_acct_uri, to_be_announced_object_uri: str) -> str: - """ - Perform whatever actions are necessary so the actor with actor_acct_uri will have Announced - on this Node, the object with announced_object_uri. - return: URI to the Announce object - """ - raise NotImplementedByNodeError(self, FediverseNode.make_announce_object) - - - def make_reply_note(self, actor_acct_uri, to_be_replied_to_object_uri: str, reply_content: str) -> str: - """ - Perform whatever actions are necessary so the actor with actor_acct_uri will have created - a Note object that replies to the object at to_be_replied_to_object_uri with the specified content. - return: URI to the Reply object - """ - raise NotImplementedByNodeError(self, FediverseNode.make_reply_note) - +# Operations related to relations between actors def make_follow(self, actor_acct_uri: str, to_follow_actor_acct_uri: str) -> None: """ @@ -196,49 +169,121 @@ def make_follow_reject(self, actor_acct_uri: str, would_be_follower_actor_acct_u raise NotImplementedByNodeError(self, FediverseNode.make_follow_reject) - def make_follow_undo(self, actor_acct_uri: str, following_actor_acct_uri: str) -> None: + def make_unfollow(self, actor_acct_uri: str, following_actor_acct_uri: str) -> None: """ - Perform whatever actions are necessary so the actor with actor_acct_uri will have created + Perform whatever actions are necessary so the actor with actor_acct_uri will have unfollowed the Actor with following_actor_acct_uri. The actor with actor_acct_uri must be on this Node. """ - raise NotImplementedByNodeError(self, FediverseNode.make_follow_undo) + raise NotImplementedByNodeError(self, FediverseNode.make_unfollow) - def actor_has_received_note(self, actor_acct_uri: str, object_uri: str) -> str | None: + def actor_is_following_actor(self, actor_acct_uri: str, leader_actor_acct_uri: str) -> bool: """ - If the note at object_uri has arrived with the Actor at actor_acct_uri, return the content - of the note. + Return True if the Actor at actor_acct_uri is following the Actor at leader_actor_acct_uri, + in the opinion of this Node. """ - raise NotImplementedByNodeError(self, FediverseNode.actor_has_received_note) + raise NotImplementedByNodeError(self, FediverseNode.actor_is_following_actor) - def actor_is_following_actor(self, actor_acct_uri: str, leader_actor_acct_uri: str) -> bool: + def actor_is_followed_by_actor(self, actor_acct_uri: str, follower_actor_acct_uri: str) -> bool: """ - Return True if the Actor at actor_acct_uri is following the Actor at leader_actor_acct_uri, + Return True if the Actor at actor_acct_uri is followed by Actor at follower_actor_acct_uri, in the opinion of this Node. """ - raise NotImplementedByNodeError(self, FediverseNode.actor_is_following_actor) + raise NotImplementedByNodeError(self, FediverseNode.actor_is_followed_by_actor) +# Operations related to content creation, modification, deletion - def note_has_direct_reply(self, actor_acct_uri: str, note_uri: str, reply_uri: str) -> str | None: + def make_create_note(self, actor_acct_uri: str, content: str, deliver_to: list[str] | None = None) -> str: + """" + Perform whatever actions are necessary so the actor with actor_acct_uri will have created + a Note object on this Node with the specified content. + deliver_to: make sure the Node is delivered to these Actors (i.e. in arrives in their inbox) + return: URI to the Note object + """ + raise NotImplementedByNodeError(self, FediverseNode.make_create_note) + + + def update_note(self, actor_acct_uri: str, note_uri: str, new_content: str) -> None: """ Return the reply's content if the Actor at actor_acct_uri can see that the Note at note_uri has a reply note with reply_uri on this Node. """ - raise NotImplementedByNodeError(self, FediverseNode.note_has_direct_reply) + raise NotImplementedByNodeError(self, FediverseNode.update_note) - def access_note(self, actor_acct_uri: str, note_uri: str) -> str | None: + def delete_object(self, actor_acct_uri: str, object_uri: str) -> None: + """ + Delete a note (boost, announce). + """ + raise NotImplementedByNodeError(self, FediverseNode.delete_object) + +# Operations related to engaging with existing content + + def make_reply_note(self, actor_acct_uri: str, to_be_replied_to_object_uri: str, reply_content: str) -> str: + """ + Perform whatever actions are necessary so the actor with actor_acct_uri will have created + a Note object that replies to the object at to_be_replied_to_object_uri with the specified content. + return: URI to the Reply object + """ + raise NotImplementedByNodeError(self, FediverseNode.make_reply_note) + + + def like_object(self, actor_acct_uri: str, object_uri: str) -> None: + """ + Like an object (like a note). + """ + raise NotImplementedByNodeError(self, FediverseNode.like_object) + + + def announce_object(self, actor_acct_uri: str, object_uri: str) -> None: + """ + Announce an object (boost, reblog). + """ + raise NotImplementedByNodeError(self, FediverseNode.announce_object) + + + def actor_has_received_object(self, actor_acct_uri: str, object_uri: str) -> str | None: + """ + If the object at object_uri has arrived with the Actor at actor_acct_uri, return the content + of the object. + """ + raise NotImplementedByNodeError(self, FediverseNode.actor_has_received_object) + +# Operations related to examining existing objects + + def note_content(self, actor_acct_uri: str, note_uri: str) -> str | None: """ Return the content of the Note at not_uri if the Actor at actor_acct_uri can access it. """ - raise NotImplementedByNodeError(self, FediverseNode.access_note) + raise NotImplementedByNodeError(self, FediverseNode.note_content) - def update_note(self, actor_acct_uri: str, note_uri: str, new_content: str) -> None: + def object_author(self, actor_acct_uri: str, object_uri: str) -> str | None: """ - Return the reply's content if the Actor at actor_acct_uri can see that the Note at note_uri has a reply - note with reply_uri on this Node. + Return the actor acct URI of actor that is the author of the object at object_uri. """ - raise NotImplementedByNodeError(self, FediverseNode.update_note) + raise NotImplementedByNodeError(self, FediverseNode.object_author) + + + def direct_replies_to_object(self, actor_acct_uri: str, object_uri: str) -> list[str]: + """ + Return the URIs of the objects that directly reply to the object at object_uri. + """ + raise NotImplementedByNodeError(self, FediverseNode.direct_replies_to_object) + + + def object_likers(self, actor_acct_uri: str, object_uri: str) -> list[str]: + """ + Return the set of actor acct URI of the actors that liked this object. + """ + raise NotImplementedByNodeError(self, FediverseNode.object_likers) + + + def object_announcers(self, actor_acct_uri: str, object_uri: str) -> list[str]: + """ + Return the set of actor acct URI of the actors that announced/boosted/reblogged this object. + """ + raise NotImplementedByNodeError(self, FediverseNode.object_announcers) + diff --git a/src/feditest/utils.py b/src/feditest/utils.py index 7747abb..3725d63 100644 --- a/src/feditest/utils.py +++ b/src/feditest/utils.py @@ -270,7 +270,7 @@ def boolean_parse_validate(candidate: Any | None) -> bool | None: return None -def account_id_parse_validate(candidate: str) -> ParsedUri | None: +def acct_uri_parse_validate(candidate: str) -> ParsedUri | None: """ Validate that the provided string is of the form 'acct:foo@bar.com'. return ParsedUri if valid, None otherwise @@ -281,13 +281,20 @@ def account_id_parse_validate(candidate: str) -> ParsedUri | None: return None -def account_id_validate(candidate: str) -> str | None: - parsed = account_id_parse_validate(candidate) +def acct_uri_validate(candidate: str) -> str | None: + parsed = acct_uri_parse_validate(candidate) if parsed: return parsed.uri return None +def acct_uri_list_validate(candidate: str) -> str | None: + for uri in candidate.split(): + if not acct_uri_validate(uri): + return None + return candidate + + def https_uri_parse_validate(candidate: str) -> ParsedUri | None: """ Validate that the provided string is a valid HTTPS URI. @@ -306,6 +313,13 @@ def https_uri_validate(candidate: str) -> str | None: return None +def https_uri_list_validate(candidate: str) -> str | None: + for uri in candidate.split(): + if not https_uri_validate(uri): + return None + return candidate + + def http_https_uri_parse_validate(candidate: str) -> ParsedUri | None: """ Validate that the provided string is a valid HTTP or HTTPS URI. diff --git a/tests.smoke/tests/node_with_mastodon_api.py b/tests.smoke/tests/node_with_mastodon_api.py index 5889918..04426dd 100644 --- a/tests.smoke/tests/node_with_mastodon_api.py +++ b/tests.smoke/tests/node_with_mastodon_api.py @@ -50,7 +50,7 @@ def create_note(self): @step def wait_for_note_in_inbox(self): - poll_until(lambda: self.server.actor_has_received_note(self.actor_acct_uri, self.note_uri)) + poll_until(lambda: self.server.actor_has_received_object(self.actor_acct_uri, self.note_uri)) # @step diff --git a/tests.smoke/tests/nodes_with_mastodon_api_communicate.py b/tests.smoke/tests/nodes_with_mastodon_api_communicate.py index 1904711..a5c1f10 100644 --- a/tests.smoke/tests/nodes_with_mastodon_api_communicate.py +++ b/tests.smoke/tests/nodes_with_mastodon_api_communicate.py @@ -65,7 +65,7 @@ def leader_creates_note(self): @step def wait_until_note_received(self): - poll_until(lambda: self.follower_node.actor_has_received_note(self.follower_actor_acct_uri, self.leader_note_uri)) + poll_until(lambda: self.follower_node.actor_has_received_object(self.follower_actor_acct_uri, self.leader_note_uri)) # @step