feat: Implement invite_participants in client

docker-compose.yml

@@ -61,6 +61,12 @@ services:
     ports:
       - "1025:1025"
       - "8025:8025"
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8025/api/v2/messages?limit=1"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 5s
 
 volumes:
   apache:

src/citric/client.py

@@ -5,13 +5,15 @@
 import base64
 import datetime
 import io
+import re
 import typing as t
 from dataclasses import dataclass
 from pathlib import Path
 
 import requests
 
 from citric import enums
+from citric.exceptions import LimeSurveyStatusError
 from citric.session import Session
 
 if t.TYPE_CHECKING:
@@ -32,6 +34,8 @@
     else:
         from typing_extensions import Self, Unpack
 
+EMAILS_SENT_STATUS_PATTERN = re.compile(r"(-?\d+) left to send")
+
 
 @dataclass
 class QuestionReference:
@@ -1606,3 +1610,36 @@ def upload_file(
 
         with Path(path).open("rb") as file:
             return self.upload_file_object(survey_id, field, filename, file)
+
+    def invite_participants(
+        self,
+        survey_id: int,
+        *,
+        token_ids: list[int] | None = None,
+        strategy: int = enums.EmailSendStrategy.PENDING,
+    ) -> int:
+        """Invite participants to a survey.
+
+        Args:
+            survey_id: ID of the survey to invite participants to.
+            token_ids: IDs of the participants to invite.
+            strategy: Strategy to use for sending emails. See
+                :class:`~citric.enums.EmailSendStrategy`.
+
+        Returns:
+            Number of emails left to send.
+
+        Raises:
+            LimeSurveyStatusError: If the number of emails left to send could not be
+                determined.
+
+        .. versionadded:: NEXT_VERSION
+        """
+        email_flag = enums.EmailSendStrategy.to_flag(strategy)
+        try:
+            self.session.invite_participants(survey_id, token_ids, email_flag)
+        except LimeSurveyStatusError as error:
+            status_match = re.match(EMAILS_SENT_STATUS_PATTERN, error.args[0])
+            if status_match:
+                return int(status_match[1])
+            raise

src/citric/enums.py

@@ -101,3 +101,15 @@ def integer_value(self) -> int:
             self.CONFIRM_TERMINATE: 2,
         }
         return mapping[self]
+
+
+class EmailSendStrategy(enum.IntEnum):
+    """Email send flag."""
+
+    PENDING = 1
+    RESEND = 2
+
+    @classmethod
+    def to_flag(cls: type[EmailSendStrategy], value: int) -> bool:
+        """Return the flag for this email send enum."""
+        return value == cls.PENDING

tests/test_integration.py

@@ -57,6 +57,37 @@ def survey_id(client: citric.Client) -> t.Generator[int, None, None]:
     client.delete_survey(survey_id)
 
 
+@pytest.fixture
+def participants(faker: Faker) -> list[dict[str, t.Any]]:
+    """Create participants for a survey."""
+    return [
+        {
+            "email": faker.email(),
+            "firstname": faker.first_name(),
+            "lastname": faker.last_name(),
+            "token": "1",
+            "attribute_1": "Dog person",
+            "attribute_2": "Night owl",
+        },
+        {
+            "email": faker.email(),
+            "firstname": faker.first_name(),
+            "lastname": faker.last_name(),
+            "token": "2",
+            "attribute_1": "Cat person",
+            "attribute_2": "Early bird",
+        },
+        {
+            "email": faker.email(),
+            "firstname": faker.first_name(),
+            "lastname": faker.last_name(),
+            "token": "2",
+            "attribute_1": "Cat person",
+            "attribute_2": "Night owl",
+        },
+    ]
+
+
 @pytest.mark.integration_test
 def test_fieldmap(client: citric.Client, survey_id: int):
     """Test fieldmap."""
@@ -275,69 +306,47 @@ def test_activate_tokens(client: citric.Client, survey_id: int):
 
 
 @pytest.mark.integration_test
-def test_participants(faker: Faker, client: citric.Client, survey_id: int):
+def test_participants(
+    faker: Faker,
+    client: citric.Client,
+    survey_id: int,
+    participants: list[dict[str, str]],
+):
     """Test participants methods."""
     client.activate_survey(survey_id)
     client.activate_tokens(survey_id, [1, 2])
 
-    data = [
-        {
-            "email": faker.email(),
-            "firstname": faker.first_name(),
-            "lastname": faker.last_name(),
-            "token": "1",
-            "attribute_1": "Dog person",
-            "attribute_2": "Night owl",
-        },
-        {
-            "email": faker.email(),
-            "firstname": faker.first_name(),
-            "lastname": faker.last_name(),
-            "token": "2",
-            "attribute_1": "Cat person",
-            "attribute_2": "Early bird",
-        },
-        {
-            "email": faker.email(),
-            "firstname": faker.first_name(),
-            "lastname": faker.last_name(),
-            "token": "2",
-            "attribute_1": "Cat person",
-            "attribute_2": "Night owl",
-        },
-    ]
-
     # Add participants
     added = client.add_participants(
         survey_id,
-        participant_data=data,
+        participant_data=participants,
         create_tokens=False,
     )
-    for p, d in zip(added, data):
+    for p, d in zip(added, participants):
         assert p["email"] == d["email"]
         assert p["firstname"] == d["firstname"]
         assert p["lastname"] == d["lastname"]
         assert p["attribute_1"] == d["attribute_1"]
         assert p["attribute_2"] == d["attribute_2"]
 
-    participants = client.list_participants(
+    participants_list = client.list_participants(
         survey_id,
         attributes=["attribute_1", "attribute_2"],
     )
 
     # Confirm that the participants are deduplicated based on token
-    assert len(participants) == 2
+    assert len(participants_list) == 2
 
     # Check added participant properties
-    for p, d in zip(participants, data[:2]):
+    for p, d in zip(participants_list, participants[:2]):
         assert p["participant_info"]["email"] == d["email"]
         assert p["participant_info"]["firstname"] == d["firstname"]
         assert p["participant_info"]["lastname"] == d["lastname"]
         assert p["attribute_1"] == d["attribute_1"]
         assert p["attribute_2"] == d["attribute_2"]
 
     # Get participant properties
-    for p, d in zip(added, data[:2]):
+    for p, d in zip(added, participants[:2]):
         properties = client.get_participant_properties(survey_id, p["tid"])
         assert properties["email"] == d["email"]
         assert properties["firstname"] == d["firstname"]
@@ -365,6 +374,37 @@ def test_participants(faker: Faker, client: citric.Client, survey_id: int):
     assert len(client.list_participants(survey_id)) == 1
 
 
+@pytest.mark.integration_test
+def test_invite_participants(
+    client: citric.Client,
+    survey_id: int,
+    participants: list[dict[str, str]],
+):
+    """Test inviting participants to a survey."""
+    client.activate_survey(survey_id)
+    client.activate_tokens(survey_id, [1, 2])
+
+    # Add participants
+    client.add_participants(
+        survey_id,
+        participant_data=participants,
+        create_tokens=False,
+    )
+
+    pending = enums.EmailSendStrategy.PENDING
+    resend = enums.EmailSendStrategy.RESEND
+
+    with pytest.raises(LimeSurveyStatusError, match="Error: No candidate tokens"):
+        client.invite_participants(survey_id, strategy=resend)
+
+    assert client.invite_participants(survey_id, strategy=pending) == 0
+
+    with pytest.raises(LimeSurveyStatusError, match="Error: No candidate tokens"):
+        client.invite_participants(survey_id, strategy=pending)
+
+    assert client.invite_participants(survey_id, strategy=resend) == -2
+
+
 @pytest.mark.integration_test
 def test_responses(client: citric.Client, survey_id: int):
     """Test adding and exporting responses."""