Skip to content

funfedi_connect

ApplicationPoster module-attribute 

ApplicationPoster = Callable[
    [ClientSession, str], Awaitable[None]
]

Describes how to post a message

ApplicationConfiguration

Configuration of an application to run tests with

Source code in funfedi_connect/types/__init__.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class ApplicationConfiguration:
    """Configuration of an application to run tests with"""

    domain: str
    username: str
    application_name: str
    features: list[ImplementedFeature]

    @property
    def acct_uri(self):
        return f"acct:{self.username}@{self.domain}"

    async def _determine_actor_uri(self, session: aiohttp.ClientSession):
        actor_uri, _ = await lookup_uri_with_webfinger(
            session, self.acct_uri, f"http://{self.domain}"
        )
        return actor_uri

    @abstractmethod
    async def build_object_parsing(
        self, session: aiohttp.ClientSession
    ) -> ParsingTestApplicationConfiguration:
        """Builds the object parsing configuration"""

    @abstractmethod
    async def build_event_parsing(
        self, session: aiohttp.ClientSession
    ) -> ParsingTestApplicationConfiguration:
        """Builds the event parsing configuration"""

    @abstractmethod
    async def build_poster(self, session: aiohttp.ClientSession) -> ApplicationPoster:
        """Returns a function that allows posting a message"""

build_event_parsing abstractmethod async 

build_event_parsing(
    session: ClientSession,
) -> ParsingTestApplicationConfiguration

Builds the event parsing configuration

Source code in funfedi_connect/types/__init__.py 
43
44
45
46
47
@abstractmethod
async def build_event_parsing(
    self, session: aiohttp.ClientSession
) -> ParsingTestApplicationConfiguration:
    """Builds the event parsing configuration"""

build_object_parsing abstractmethod async 

build_object_parsing(
    session: ClientSession,
) -> ParsingTestApplicationConfiguration

Builds the object parsing configuration

Source code in funfedi_connect/types/__init__.py 
37
38
39
40
41
@abstractmethod
async def build_object_parsing(
    self, session: aiohttp.ClientSession
) -> ParsingTestApplicationConfiguration:
    """Builds the object parsing configuration"""

build_poster abstractmethod async 

build_poster(session: ClientSession) -> ApplicationPoster

Returns a function that allows posting a message

Source code in funfedi_connect/types/__init__.py 
49
50
51
@abstractmethod
async def build_poster(self, session: aiohttp.ClientSession) -> ApplicationPoster:
    """Returns a function that allows posting a message"""

Attachments

Bases: StrEnum

Attachments added through allure.attach

Source code in funfedi_connect/types/__init__.py 
54
55
56
57
58
59
60
61
62
63
class Attachments(StrEnum):
    """Attachments added through allure.attach"""

    meta_data = auto()
    actor_object = auto()
    created_object = auto()
    send_activity = auto()
    timeline_item = auto()
    event_item = auto()
    webfinger_response = auto()

ImplementedFeature

Bases: StrEnum

Available implemented features

Source code in funfedi_connect/types/feature.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class ImplementedFeature(StrEnum):
    """Available implemented features"""

    webfinger = auto()
    public_timeline = auto()
    post = auto()
    event = auto()

    @staticmethod
    def from_tag(tag: str):
        """
        ```
        >>> ImplementedFeature.from_tag("public-timeline")
        <ImplementedFeature.public_timeline: 'public_timeline'>

        ```
        """
        tag = tag.replace("-", "_")

        return ImplementedFeature(tag)

    @property
    def behave_tag(self):
        """Returns as the tag used in behave, e.g.

        ```
        >>> ImplementedFeature.public_timeline.behave_tag
        'public-timeline'

        ```
        """
        return self.value.replace("_", "-")

behave_tag property 

behave_tag

Returns as the tag used in behave, e.g.

>>> ImplementedFeature.public_timeline.behave_tag
'public-timeline'

from_tag staticmethod 

from_tag(tag: str)

>>> ImplementedFeature.from_tag("public-timeline")
<ImplementedFeature.public_timeline: 'public_timeline'>
Source code in funfedi_connect/types/feature.py 
12
13
14
15
16
17
18
19
20
21
22
23
@staticmethod
def from_tag(tag: str):
    """
    ```
    >>> ImplementedFeature.from_tag("public-timeline")
    <ImplementedFeature.public_timeline: 'public_timeline'>

    ```
    """
    tag = tag.replace("-", "_")

    return ImplementedFeature(tag)

ParsingTestApplicationConfiguration dataclass

Configuration for testing activity parsing by retrieving the parsed object by its id

Parameters:

Name Type Description Default
actor_id str
required
application_name str
required
base_object_getter Callable[list, Awaitable[dict | None]]
required
poll_number int
5
wait_time float
1
Source code in funfedi_connect/types/parsing.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@dataclass
class ParsingTestApplicationConfiguration:
    """Configuration for testing activity parsing by retrieving
    the parsed object by its id"""

    actor_id: str
    application_name: str
    base_object_getter: ApplicationObjectGetter

    poll_number: int = 5
    wait_time: float = 1

    async def object_getter(
        self, session: aiohttp.ClientSession, object_id: str
    ) -> dict | None:
        """Returns the parsing result or not if the result could not be retrieved, e.g.
        due to a parsing failure. Polling is used up to poll_number with wait_time seconds
        in between."""
        for _ in range(self.poll_number):
            result = await self.base_object_getter(session, object_id)
            if result:
                return result

            await asyncio.sleep(self.wait_time)

        return None

object_getter async 

object_getter(
    session: ClientSession, object_id: str
) -> dict | None

Returns the parsing result or not if the result could not be retrieved, e.g. due to a parsing failure. Polling is used up to poll_number with wait_time seconds in between.

Source code in funfedi_connect/types/parsing.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
async def object_getter(
    self, session: aiohttp.ClientSession, object_id: str
) -> dict | None:
    """Returns the parsing result or not if the result could not be retrieved, e.g.
    due to a parsing failure. Polling is used up to poll_number with wait_time seconds
    in between."""
    for _ in range(self.poll_number):
        result = await self.base_object_getter(session, object_id)
        if result:
            return result

        await asyncio.sleep(self.wait_time)

    return None

application_for_name 

application_for_name(name: str) -> ApplicationConfiguration

For a given name returns the corresponding application configuration

Source code in funfedi_connect/applications/__init__.py 
46
47
48
49
50
51
def application_for_name(name: str) -> ApplicationConfiguration:
    """For a given name returns the corresponding application configuration"""
    if name in name_to_application:
        return name_to_application[name]

    raise Exception("Unknown application")

application_names_as_list 

application_names_as_list() -> list[str]

Returns the list of all implemented applications

Source code in funfedi_connect/applications/__init__.py 
54
55
56
def application_names_as_list() -> list[str]:
    """Returns the list of all implemented applications"""
    return sorted(list(name_to_application.keys()))