Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 58 additions & 0 deletions spond/spond.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ def __init__(self, username: str, password: str) -> None:
self._auth = None
self.groups: list[JSONDict] | None = None
self.events: list[JSONDict] | None = None
self.posts: list[JSONDict] | None = None
self.messages: list[JSONDict] | None = None
self.profile: JSONDict | None = None

Expand Down Expand Up @@ -133,6 +134,63 @@ def _match_person(person: JSONDict, match_str: str) -> bool:
or ("profile" in person and person["profile"]["id"] == match_str)
)

@_SpondBase.require_authentication
async def get_posts(
self,
group_id: str | None = None,
max_posts: int = 20,
include_comments: bool = True,
) -> list[JSONDict] | None:
"""
Retrieve posts from group walls.

Posts are announcements/messages posted to group walls, as opposed to
chat messages or events.

Parameters
----------
group_id : str, optional
Filter by group. Uses `groupId` API parameter.
max_posts : int, optional
Set a limit on the number of posts returned.
For performance reasons, defaults to 20.
Uses `max` API parameter.
include_comments : bool, optional
Include comments on posts.
Defaults to True.
Uses `includeComments` API parameter.

Returns
-------
list[JSONDict] or None
A list of posts, each represented as a dictionary, or None if no
posts are available.

Raises
------
ValueError
Raised when the request to the API fails.
"""
url = f"{self.api_url}posts"
params: dict[str, str] = {
"type": "PLAIN",
"max": str(max_posts),
"includeComments": str(include_comments).lower(),
Comment on lines +174 to +178
}
if group_id:
params["groupId"] = group_id

async with self.clientsession.get(
url, headers=self.auth_headers, params=params
) as r:
if not r.ok:
error_details = await r.text()
raise ValueError(
f"Request failed with status {r.status}: {error_details}"
)
self.posts = await r.json()
return self.posts

@_SpondBase.require_authentication
async def get_messages(self, max_chats: int = 100) -> list[JSONDict] | None:
"""
Expand Down
128 changes: 128 additions & 0 deletions tests/test_spond.py
Original file line number Diff line number Diff line change
Expand Up @@ -212,3 +212,131 @@ async def test_get_export(self, mock_get, mock_token) -> None:
},
)
assert data == mock_binary


class TestPostMethods:
MOCK_POSTS: list[JSONDict] = [
{
"id": "POST1",
"type": "PLAIN",
"groupId": "GID1",
"title": "Post One",
"body": "Body of post one",
"timestamp": "2026-03-03T19:20:00.270Z",
"comments": [],
},
{
"id": "POST2",
"type": "PLAIN",
"groupId": "GID2",
"title": "Post Two",
"body": "Body of post two",
"timestamp": "2026-02-20T19:21:20.447Z",
"comments": [{"id": "C1", "text": "A comment"}],
},
]

@pytest.mark.asyncio
@patch("aiohttp.ClientSession.get")
async def test_get_posts__happy_path(self, mock_get, mock_token) -> None:
"""Test that get_posts returns posts from the API."""
s = Spond(MOCK_USERNAME, MOCK_PASSWORD)
s.token = mock_token

mock_get.return_value.__aenter__.return_value.ok = True
mock_get.return_value.__aenter__.return_value.json = AsyncMock(
return_value=self.MOCK_POSTS
)

posts = await s.get_posts()

mock_url = "https://api.spond.com/core/v1/posts"
mock_get.assert_called_once_with(
mock_url,
headers={
"content-type": "application/json",
"Authorization": f"Bearer {mock_token}",
},
params={
"type": "PLAIN",
"max": "20",
"includeComments": "true",
},
)
Comment on lines +253 to +265
assert posts == self.MOCK_POSTS
assert s.posts == self.MOCK_POSTS

@pytest.mark.asyncio
@patch("aiohttp.ClientSession.get")
async def test_get_posts__with_group_id(self, mock_get, mock_token) -> None:
"""Test that group_id is passed as a query parameter."""
s = Spond(MOCK_USERNAME, MOCK_PASSWORD)
s.token = mock_token

mock_get.return_value.__aenter__.return_value.ok = True
mock_get.return_value.__aenter__.return_value.json = AsyncMock(
return_value=[self.MOCK_POSTS[0]]
)

posts = await s.get_posts(group_id="GID1")

mock_get.assert_called_once_with(
"https://api.spond.com/core/v1/posts",
headers={
"content-type": "application/json",
"Authorization": f"Bearer {mock_token}",
},
params={
"type": "PLAIN",
"max": "20",
"includeComments": "true",
"groupId": "GID1",
},
)
assert len(posts) == 1

@pytest.mark.asyncio
@patch("aiohttp.ClientSession.get")
async def test_get_posts__custom_max(self, mock_get, mock_token) -> None:
"""Test that max_posts parameter is respected."""
s = Spond(MOCK_USERNAME, MOCK_PASSWORD)
s.token = mock_token

mock_get.return_value.__aenter__.return_value.ok = True
mock_get.return_value.__aenter__.return_value.json = AsyncMock(return_value=[])

await s.get_posts(max_posts=5)

call_params = mock_get.call_args[1]["params"]
assert call_params["max"] == "5"

@pytest.mark.asyncio
@patch("aiohttp.ClientSession.get")
async def test_get_posts__no_comments(self, mock_get, mock_token) -> None:
"""Test that include_comments=False is passed correctly."""
s = Spond(MOCK_USERNAME, MOCK_PASSWORD)
s.token = mock_token

mock_get.return_value.__aenter__.return_value.ok = True
mock_get.return_value.__aenter__.return_value.json = AsyncMock(return_value=[])

await s.get_posts(include_comments=False)

call_params = mock_get.call_args[1]["params"]
assert call_params["includeComments"] == "false"

@pytest.mark.asyncio
@patch("aiohttp.ClientSession.get")
async def test_get_posts__api_error_raises(self, mock_get, mock_token) -> None:
"""Test that a failed API response raises ValueError."""
s = Spond(MOCK_USERNAME, MOCK_PASSWORD)
s.token = mock_token

mock_get.return_value.__aenter__.return_value.ok = False
mock_get.return_value.__aenter__.return_value.status = 401
mock_get.return_value.__aenter__.return_value.text = AsyncMock(
return_value="Unauthorized"
)

with pytest.raises(ValueError, match="401"):
await s.get_posts()
Loading