Update form repo to support adding and retrieving responses, with tests.

api/openapi_server/repositories/forms.py

@@ -1,7 +1,7 @@
 from typing import Optional
 
-from openapi_server.models.schema import form_schema, FormSchema
-from openapi_server.models.database import Form
+from openapi_server.models.schema import form_schema, response_schema, FormSchema
+from openapi_server.models.database import Form, Response
 from openapi_server.repositories.base import BaseRepo
 
 class FormsRepository(BaseRepo):
@@ -13,11 +13,36 @@ def add_form(self, form_json) -> int:
             session.commit()
             return form.form_id 
 
-    def get_form(self, form_id) -> Optional[Form]:
+    def get_form(self, form_id: int) -> Optional[Form]:
         return self.session.query(Form).get(form_id)
 
     def get_form_json(self, form_id) -> Optional[FormSchema]:
         form = self.get_form(form_id)
         if form is not None:
             return form_schema.dump(form)
-        return None
+        return None
+
+    def get_user_responses(self, user_id: int, form_id: int):
+        with self.session as session:
+            form = session.query(Form).get(form_id)
+            field_ids = [field.field_id for group in form.field_groups for field in group.fields]
+            responses = session.query(Response).filter(
+                Response.user_id == user_id,
+                Response.field_id.in_(field_ids)
+            ).all()
+            return response_schema.dump(responses)
+
+    def add_user_responses(self, user_id: int, responses) -> None:
+        '''
+        Add list of responses, from raw json. This function will 
+        parse the json and overwrite any existing responses.
+        '''
+        with self.session as session:
+            new_responses = response_schema.load(responses, session=session)
+            new_field_ids = [resp.field_id for resp in new_responses]
+            session.query(Response).filter(
+                Response.user_id == user_id,
+                Response.field_id.in_(new_field_ids)
+            ).delete(synchronize_session=False)
+            session.add_all(new_responses)
+            session.commit()

api/openapi_server/repositories/user_repo.py

@@ -33,5 +33,8 @@ def delete_user(self, user_id: int) -> bool:
     def get_user(self, email: str) -> User:
         return self.session.query(User).filter_by(email=email).first()
 
+    def get_user_id(self, email: str) -> int:
+        return self.session.query(User).filter_by(email=email).first().id
+
     def get_users_with_role(self, role: UserRole) -> List[User]:
         return self.session.query(User).filter_by(role=self._get_role(role))

api/tests/conftest.py

@@ -118,6 +118,13 @@ def empty_db_session(alembic_runner, alembic_engine) -> Generator[Session, None,
     test_engine, DataAccessLayer._engine = DataAccessLayer._engine, None 
     test_engine.dispose()
 
+@pytest.fixture()
+def empty_db_session_provider(empty_db_session):
+    class _provider:
+        def session(): return empty_db_session
+
+    return _provider
+
 @pytest.fixture()
 def client(app):
     return app.test_client()

api/tests/test_forms_repo.py

@@ -1,68 +1,127 @@
+from types import MappingProxyType
+
 from openapi_server.repositories.forms import FormsRepository
+from openapi_server.repositories.user_repo import UserRepository, UserRole
+
+TEST_FORM_READ_ONLY = MappingProxyType({
+    "title": "Employee Onboarding",
+    "description": "Collect necessary employee data.",
+    "field_groups": [
+        {
+            "title": "Personal Details",
+            "description": "Please enter your personal details.",
+            "fields": [
+                {
+                    "ref": "position",
+                    "properties": {
+                        "description": "Position in the company",
+                        "field_type": "dropdown",
+                        "choices": ['Manager', 'Developer', 'Designer'],
+                    },
+                    "validations": {
+                        "required": True,
+                        "max_length": 12
+                    }
+                }, 
+                {
+                    "ref": "service_length",
+                    "properties": {
+                        "description": "Years in the company",
+                        "field_type": "number",
+                        "choices": None,
+                    },
+                    "validations": {
+                        "required": False,
+                        "max_length": None
+                    }
+                }
+            ]
+        },
+        {
+            "title": "Second Group",
+            "description": "A second field group.",
+            "fields": [
+                {
+                    "ref": "start date",
+                    "properties": {
+                        "description": "Start date",
+                        "field_type": "date",
+                        "choices": "11-22-2005",
+                    },
+                    "validations": {
+                        "required": True,
+                        "max_length": 12
+                    }
+                }
+            ]
+        }
+    ]
+})
+
+def assert_form_equal(actual_form: dict, expected_form: dict):
+    '''
+    Do a deep equality check of a form, excluding dynamically
+    assigned values like timestamps and primary key ids.
+    '''
+    actual_copy = actual_form.copy()
+    del actual_copy['created_at']
+    for group in actual_copy['field_groups']:
+        del group['form']
+        for field in group['fields']:
+            del field['field_id']
+            del field['group']
+
+    assert actual_copy == expected_form
 
 def test_add_form_valid_json(empty_db_session_provider):
-    form_json = {
-        "title": "Employee Onboarding",
-        "description": "Collect necessary employee data.",
-        "field_groups": [
+    form_json = dict(TEST_FORM_READ_ONLY)
+
+    form_repo = FormsRepository(empty_db_session_provider.session())
+    created_form_id = form_repo.add_form(form_json)
+    retrieved_form = form_repo.get_form_json(created_form_id)
+
+    assert_form_equal(retrieved_form, form_json)
+
+def test_add_get_responses(empty_db_session_provider):
+    with empty_db_session_provider.session() as session:
+        user_repo = UserRepository(session)
+        form_repo = FormsRepository(session)
+
+        user_repo.add_user('fake@email.com', UserRole.COORDINATOR, 'firstname')
+        user_id = user_repo.get_user_id('fake@email.com')
+        created_form_id = form_repo.add_form(TEST_FORM_READ_ONLY)
+        retrieved_form = form_repo.get_form_json(created_form_id)
+
+        def _get_field_id(lcl_form, ref):
+            for group in lcl_form['field_groups']:
+                for field in group['fields']:
+                    if field['ref'] == ref:
+                        return int(field['field_id'])
+            raise ValueError(f'ref {ref} not found in test form')
+
+        expected_responses = [
             {
-                "title": "Personal Details",
-                "description": "Please enter your personal details.",
-                "fields": [
-                    {
-                        "ref": "position",
-                        "properties": {
-                            "description": "Position in the company",
-                            "field_type": "dropdown",
-                            "choices": ['Manager', 'Developer', 'Designer'],
-                        },
-                        "validations": {
-                            "required": True,
-                            "max_length": 12
-                        }
-                    }, 
-                    {
-                        "ref": "service_length",
-                        "properties": {
-                            "description": "Years in the company",
-                            "field_type": "number",
-                            "choices": None,
-                        },
-                        "validations": {
-                            "required": False,
-                            "max_length": None
-                        }
-                    }
-                ]
+                "user_id": user_id,
+                "field_id": _get_field_id(retrieved_form, 'position'),
+                "answer_text": "Designer"
             },
             {
-                "title": "Second Group",
-                "description": "A second field group.",
-                "fields": [
-                    {
-                        "ref": "start date",
-                        "properties": {
-                            "description": "Start date",
-                            "field_type": "date",
-                            "choices": "11-22-2005",
-                        },
-                        "validations": {
-                            "required": True,
-                            "max_length": 12
-                        }
-                    }
-                ]
+                "user_id": user_id,
+                "field_id": _get_field_id(retrieved_form, 'service_length'),
+                "answer_text": "5"
+            },
+            {
+                "user_id": user_id,
+                "field_id": _get_field_id(retrieved_form, 'start date'),
+                "answer_text": '2024-05-19'
             }
         ]
-    }
+        form_repo.add_user_responses(user_id, expected_responses)
 
-    form_repo = FormsRepository(empty_db_session_provider.session())
-    created_form_id = form_repo.add_form(form_json)
-    retrieved_form = form_repo.get_form_json(created_form_id)
+        retrieved_answers = form_repo.get_user_responses(user_id, created_form_id)
 
-    # the json won't match exactly because retrieved_form has ids associated with
-    # it, but we can do a spot check at least
-    assert retrieved_form['title'] == form_json['title']
-    assert len(retrieved_form['field_groups']) == len(form_json['field_groups'])
-    assert len(retrieved_form['field_groups'][1]['fields']) == len(form_json['field_groups'][1]['fields'])
-    assert retrieved_form['field_groups'][1]['fields'][0]['validations'] == form_json['field_groups'][1]['fields'][0]['validations']
+        assert len(retrieved_answers) == 3
+        for expected, actual in zip(expected_responses, retrieved_answers):
+            assert expected['answer_text'] == actual['answer_text']
+            assert expected['user_id'] == actual['user']['id']
+            assert expected['field_id'] == actual['field']['field_id']