Merge pull request #21 from ibehave-ibots/mo.registration-workflows

Registration workflows

course_attendance_service/course_attendance_service/workflows/registration/api_zoom.py

@@ -0,0 +1,62 @@
+from typing import Any, Dict, List, Literal, TypedDict, Union
+import requests
+
+
+class ZoomRestApi:
+    def __init__(self, api_server: str = None):
+        self.api_server = (
+            api_server if api_server is not None else "https://api.zoom.us/v2"
+        )
+
+    @staticmethod
+    def get_json_from_zoom_request(url, access_token: str, params: Dict = None):
+        headers = {"Authorization": f"Bearer {access_token}"}
+        response = requests.get(url=url, headers=headers, params=params, timeout=10)
+        response.raise_for_status()
+        return response.json()
+
+    def get_registrants(
+        self,
+        access_token: str,
+        workshop_id: Union[str, int],
+        status="approved",
+        page_size=100,
+    ):
+        url = self.api_server + f"/meetings/{workshop_id}/registrants"
+        params = {"status": status, "page_size": page_size}
+        response_json = self.get_json_from_zoom_request(
+            url=url, access_token=access_token, params=params
+        )
+        data: ZoomGetRegistrantsResponseData = response_json
+        return data["registrants"]
+
+
+class ZoomRegistrantsData(TypedDict):
+    id: str
+    first_name: str
+    last_name: str
+    email: str
+    address: str
+    city: str
+    country: str
+    zip: Any
+    state: str
+    phone: Union[str, int]
+    industry: str
+    org: Any
+    job_title: str
+    purchasing_time_frame: Any
+    role_in_purchase_process: str
+    no_of_employee: int
+    comments: str
+    custom_questions: List[Dict]
+    status: Literal["approved", "denied", "pending"]
+    create_time: str
+    join_url: str
+
+
+class ZoomGetRegistrantsResponseData(TypedDict):
+    page_size: int
+    total_records: int
+    next_page_token: str
+    registrants: List[ZoomRegistrantsData]

course_attendance_service/course_attendance_service/workflows/registration/repo_inmemory.py

@@ -0,0 +1,10 @@
+from typing import Dict, Any, List
+from .workflows import Registrant, RegistrantsRepo
+
+
+class InMemoryRegistrantsRepo(RegistrantsRepo):
+    def __init__(self, workshops: Dict[Any, List[Registrant]]):
+        self.workshops = workshops
+
+    def get_list_of_registrants(self, workshop_id: Any) -> List[Registrant]:
+        return self.workshops[workshop_id]

course_attendance_service/course_attendance_service/workflows/registration/repo_zoom.py

@@ -0,0 +1,64 @@
+# %%
+from typing import List, Union
+
+from .api_zoom import ZoomRestApi, ZoomRegistrantsData
+from .workflows import Registrant, RegistrantsRepo
+
+
+class ZoomRegistrantsRepo(RegistrantsRepo):
+    def __init__(self, zoom_api: ZoomRestApi) -> None:
+        self.zoom_api = zoom_api
+        self.access_token = "TOP-SECRET"
+
+    @staticmethod
+    def turn_zoom_registrants_into_standard_form(
+        zoom_registrant: ZoomRegistrantsData,
+    ) -> Registrant:
+        user_id = zoom_registrant["id"]
+        name = f"{zoom_registrant['first_name']} {zoom_registrant['last_name']}"
+        affiliation = zoom_registrant["custom_questions"][0]["value"]
+        email = zoom_registrant["email"]
+        status = zoom_registrant["status"]
+        registrant = Registrant(
+            user_id=user_id,
+            name=name,
+            affiliation=affiliation,
+            email=email,
+            status=status,
+        )
+        return registrant
+
+    def get_list_of_registrants_for_specific_status(
+        self,
+        workshop_id: Union[str, int],
+        status: str,
+    ) -> List[Registrant]:
+        zoom_registrants = self.zoom_api.get_registrants(
+            access_token=self.access_token,
+            workshop_id=workshop_id,
+            status=status,
+        )
+        registrants = []
+        for zoom_registrant in zoom_registrants:
+            registrant = self.turn_zoom_registrants_into_standard_form(zoom_registrant)
+            registrants.append(registrant)
+        return registrants
+
+    def get_list_of_registrants(self, workshop_id: Union[str, int]) -> List[Registrant]:
+        all_registrants = []
+        for status in ["approved", "denied", "pending"]:
+            registrants = self.get_list_of_registrants_for_specific_status(
+                workshop_id=workshop_id, status=status
+            )
+            if len(registrants) > 0:
+                all_registrants.extend(registrants)
+        return all_registrants
+
+    def remove_registrant(self):
+        raise NotImplementedError
+
+    def add_registrant(self):
+        raise NotImplementedError
+
+    def update_registrant_status(self):
+        raise NotImplementedError

course_attendance_service/course_attendance_service/workflows/registration/test_repo_zoom.py

@@ -0,0 +1,74 @@
+from unittest.mock import Mock
+
+from .api_zoom import ZoomRestApi, ZoomRegistrantsData
+from .repo_zoom import ZoomRegistrantsRepo
+from .workflows import Registrant
+
+
+def test_repo_can_get_list_of_registrants_from_zoom_api():
+    # GIVEN: the zoom api and a workhop
+    zoom_api = Mock(ZoomRestApi)
+    zoom_api.get_registrants = lambda status, **kwargs: {
+        'approved': [
+            ZoomRegistrantsData(
+                id="some_random_chars",
+                first_name="Mo",
+                last_name="Bashiri",
+                custom_questions=[{"value": "iBehave"}],
+                email="[email protected]",
+                status="approved",
+            ),
+        ],
+        'denied': [],
+        'pending': [],
+    }[status]
+
+    workshop_id = "abc"
+
+    repo = ZoomRegistrantsRepo(zoom_api=zoom_api)
+
+    # WHEN: asked for the list of registrants
+    observed_registrants = repo.get_list_of_registrants(workshop_id=workshop_id)
+    expected_registrants = [
+        Registrant(
+            user_id="some_random_chars",
+            name="Mo Bashiri",
+            affiliation="iBehave",
+            email="[email protected]",
+            status="approved",
+        )
+    ]
+
+    # THEN: the correct list of registrants is returned by the repo
+    assert observed_registrants == expected_registrants
+
+
+def test_repo_can_get_correct_number_of_registrants_from_zoom_api():
+    # GIVEN: the zoom api and a workhop
+    zoom_api = Mock(ZoomRestApi)
+    zoom_api.get_registrants = lambda status, **kwargs: {
+        'approved': [
+            ZoomRegistrantsData(
+                id="some_random_chars",
+                first_name="Mo",
+                last_name="Bashiri",
+                custom_questions=[{"value": "iBehave"}],
+                email="[email protected]",
+                status="approved",
+            ),
+        ],
+        'denied': [],
+        'pending': [],
+    }[status]
+
+    workshop_id = "abc"
+
+    repo = ZoomRegistrantsRepo(zoom_api=zoom_api)
+
+    # WHEN: asked for the registrants
+    registrants = repo.get_list_of_registrants(workshop_id=workshop_id)
+    observed_number = len(registrants)
+    expected_number = 1
+
+    # THEN: the number of registrants is correct
+    assert observed_number == expected_number

course_attendance_service/course_attendance_service/workflows/registration/test_workflows.py

@@ -0,0 +1,185 @@
+from typing import NamedTuple
+from unittest.mock import Mock
+import pytest
+from .repo_inmemory import InMemoryRegistrantsRepo
+from .workflows import RegistrationWorkflows, Registrant
+
+
+@pytest.fixture
+def workshops():
+    return {
+        "workshop1": [
+            Mock(Registrant, status="denied"),
+            Mock(Registrant, status="approved"),
+            Mock(Registrant, status="pending"),
+        ],
+        "workshop2": [
+            Mock(Registrant, status="approved"),
+            Mock(Registrant, status="approved"),
+            Mock(Registrant, status="pending"),
+            Mock(Registrant, status="denied"),
+        ],
+    }
+
+
+def test_total_number_of_registrants(workshops):
+    # GIVEN: a workshop
+    registrants_repo = InMemoryRegistrantsRepo(workshops)
+    registration_workflows = RegistrationWorkflows(registrants_repo)
+
+    # WHEN: asked for the number registrants
+    registrants_report1 = registration_workflows.get_registrants_report(
+        workshop_id="workshop1"
+    )
+    registrants_report2 = registration_workflows.get_registrants_report(
+        workshop_id="workshop2"
+    )
+
+    # THEN: correct number of registrants is returned
+    assert registrants_report1.total_number_of_registrants == 3
+    assert registrants_report2.total_number_of_registrants == 4
+
+
+def test_number_of_approved_registrants(workshops):
+    # GIVEN: a workshop
+    registrants_repo = InMemoryRegistrantsRepo(workshops)
+    registration_workflows = RegistrationWorkflows(registrants_repo)
+
+    # WHEN: asked for the number of approved particpants
+    registratants_report1 = registration_workflows.get_registrants_report(
+        workshop_id="workshop1"
+    )
+    registratants_report2 = registration_workflows.get_registrants_report(
+        workshop_id="workshop2"
+    )
+
+    # THEN: the correct number of participants is returned
+    assert registratants_report1.number_of_approved_registrants == 1
+    assert registratants_report2.number_of_approved_registrants == 2
+
+
+def test_number_of_denied_registrants(workshops):
+    # GIVEN: a workshop
+    registrants_repo = InMemoryRegistrantsRepo(workshops)
+    registration_workflows = RegistrationWorkflows(registrants_repo)
+
+    # WHEN: asked for the number of denied particpants
+    registratants_report1 = registration_workflows.get_registrants_report(
+        workshop_id="workshop1"
+    )
+    registratants_report2 = registration_workflows.get_registrants_report(
+        workshop_id="workshop2"
+    )
+
+    # THEN: the correct number of participants is returned
+    assert registratants_report1.number_of_denied_registrants == 1
+    assert registratants_report2.number_of_denied_registrants == 1
+
+
+def test_number_of_pending_registrants(workshops):
+    # GIVEN: a workshop
+    registrants_repo = InMemoryRegistrantsRepo(workshops)
+    registration_workflows = RegistrationWorkflows(registrants_repo)
+
+    # WHEN: asked for the number of pending particpants
+    registratants_report1 = registration_workflows.get_registrants_report(
+        workshop_id="workshop1"
+    )
+    registratants_report2 = registration_workflows.get_registrants_report(
+        workshop_id="workshop2"
+    )
+
+    # THEN: the correct number of participants is returned
+    assert registratants_report1.number_of_pending_registrants == 1
+    assert registratants_report2.number_of_pending_registrants == 1
+
+
+def test_registrants_are_correct(workshops):
+    # GIVEN: a workshop
+    # WHEN: asked for a list of registrants
+    # THEN: correct registrants are returned
+
+    registrants_repo = InMemoryRegistrantsRepo(workshops)
+    registration_workflows = RegistrationWorkflows(registrants_repo)
+
+    registratants_report1 = registration_workflows.get_registrants_report(
+        workshop_id="workshop1"
+    )
+    expected_outcome1 = ["denied", "approved", "pending"]
+    observed_outcome1 = [registrant.status for registrant in registratants_report1.registrants]
+    assert observed_outcome1 == expected_outcome1
+
+
+    registratants_report2 = registration_workflows.get_registrants_report(
+        workshop_id="workshop2"
+    )
+    expected_outcome2 = ["approved", "approved", "pending", "denied"]
+    observed_outcome2 = [registrant.status for registrant in registratants_report2.registrants]
+    assert observed_outcome2 == expected_outcome2
+
+
+def test_approved_registrants_are_correct(workshops):
+    # GIVEN: a workshop
+    registrants_repo = InMemoryRegistrantsRepo(workshops)
+    registration_workflows = RegistrationWorkflows(registrants_repo)
+
+    # WHEN: asked for a list of approved registrants
+    registratants_report1 = registration_workflows.get_registrants_report(
+        workshop_id="workshop1"
+    )
+    expected_outcome1 = ["approved"]
+    observed_outcome1 = [registrant.status for registrant in registratants_report1.registrants if registrant.status == "approved"]
+    assert observed_outcome1 == expected_outcome1
+
+    registratants_report2 = registration_workflows.get_registrants_report(
+        workshop_id="workshop2"
+    )
+    expected_outcome2 = ["approved", "approved"]
+    observed_outcome2 = [registrant.status for registrant in registratants_report2.registrants if registrant.status == "approved"]
+    assert observed_outcome2 == expected_outcome2
+
+
+def test_denied_registrants_are_correct(workshops):
+    # GIVEN: a workshop
+    # WHEN: asked for a list of registrants    
+    # THEN: correct registrants are returned
+
+    registrants_repo = InMemoryRegistrantsRepo(workshops)
+    registration_workflows = RegistrationWorkflows(registrants_repo)
+
+    registratants_report1 = registration_workflows.get_registrants_report(
+        workshop_id="workshop1"
+    )
+    expected_outcome1 = ["denied"]
+    observed_outcome1 = [registrant.status for registrant in registratants_report1.registrants if registrant.status == "denied"]
+    assert observed_outcome1 == expected_outcome1
+
+
+    registratants_report2 = registration_workflows.get_registrants_report(
+        workshop_id="workshop2"
+    )
+    expected_outcome2 = ["denied"]
+    observed_outcome2 = [registrant.status for registrant in registratants_report2.registrants if registrant.status == "denied"]
+    assert observed_outcome2 == expected_outcome2
+
+
+def test_pending_registrants_are_correct(workshops):
+    # GIVEN: a workshop
+    # WHEN: asked for a list of registrants
+
+    registrants_repo = InMemoryRegistrantsRepo(workshops)
+    registration_workflows = RegistrationWorkflows(registrants_repo)
+
+    registratants_report1 = registration_workflows.get_registrants_report(
+        workshop_id="workshop1"
+    )
+    expected_outcome1 = ["pending"]
+    observed_outcome1 = [registrant.status for registrant in registratants_report1.registrants if registrant.status == "pending"]
+    assert observed_outcome1 == expected_outcome1
+
+    registratants_report2 = registration_workflows.get_registrants_report(
+        workshop_id="workshop2"
+    )
+    expected_outcome2 = ["pending"]
+    observed_outcome2 = [registrant.status for registrant in registratants_report2.registrants if registrant.status == "pending"]
+    assert observed_outcome2 == expected_outcome2