Merge pull request #214 from akvo/feature/213-increase-test-coverage

Feature/213 increase test coverage

backend/db/crud_answer.py

@@ -37,7 +37,7 @@ def add_answer(
     session.add(answer)
     session.commit()
     session.flush()
-    session.refresh(answer)
+    # session.refresh(answer)
     return answer
 
 
@@ -56,10 +56,6 @@ def update_answer(
     return answer
 
 
-def get_answer(session: Session) -> List[AnswerDict]:
-    return session.query(Answer).all()
-
-
 def get_answer_by_question(
         session: Session,
         question: int,

backend/requirements.txt

@@ -32,3 +32,4 @@ numpy==1.22.1
 pandas==1.4.0
 coveralls==3.3.1
 inquirer==2.9.2
+gevent==22.8.0

backend/routes/hint.py

@@ -34,7 +34,7 @@ def get(req: Request, question_id: int,
         Question.id == question_id).first()
     answers = session.query(Answer).filter(
         Answer.question == question_id).all()
-    if question.type == QuestionType.number:
+    if question.type == QuestionType.number and answers:
         values = [answer.value for answer in answers]
         mean = f"Average value is {round(np.mean(values), 2)}"
         q1 = f"First quantile value is {round(np.quantile(values, .25), 2)}"

backend/routes/project.py

@@ -12,6 +12,7 @@
 @project_route.get("/project/{id:path}",
                    response_model=List[ProjectBase],
                    summary="get project by id",
+                   name="project:get_by_id",
                    tags=["Custom"])
 def get_by_id(req: Request, id: int, session: Session = Depends(get_session)):
     project = crud_answer.get_answer_by_question(session=session, question=id)

backend/routes/user.py

@@ -74,6 +74,7 @@ def add(req: Request,
 @user_route.get("/user",
                 response_model=UserResponse,
                 summary="get all users",
+                name="user:get",
                 tags=["User"])
 def get(req: Request,
         active: int = 0,
@@ -113,6 +114,7 @@ def get(req: Request,
 @user_route.get("/user/{id:path}",
                 response_model=UserAccessBase,
                 summary="get user detail",
+                name="user:get_by_id",
                 tags=["User"])
 def get_by_id(req: Request,
               id: int,
@@ -128,6 +130,7 @@ def get_by_id(req: Request,
 @user_route.put("/user/{id:path}",
                 response_model=UserAccessBase,
                 summary="Update user",
+                name="user:update",
                 tags=["User"])
 def update_by_id(req: Request,
                  id: int,

backend/test.sh

@@ -4,7 +4,7 @@ set -euo pipefail
 
 echo "Running tests"
 COVERAGE_PROCESS_START=./.coveragerc \
-  coverage run --parallel-mode --concurrency=multiprocessing --rcfile=./.coveragerc \
+  coverage run --parallel-mode --concurrency=thread,gevent --rcfile=./.coveragerc \
   /usr/local/bin/pytest -vvv -rP
 
 echo "Coverage"

backend/tests/test_01_auth.py

@@ -14,21 +14,22 @@
 
 
 class Acc:
-    def __init__(self, verified):
+    def __init__(
+            self, verified: bool = False, email: str = None, name: str = None):
         self.exp_date = (datetime.now() + timedelta(days=30)).timestamp()
         self.data = {
-            "email": "support@akvo.org",
-            "name": "Akvo Support",
+            "email": email if email else "support@akvo.org",
+            "name": name if name else "Akvo Support",
             "exp": self.exp_date,
-            "email_verified": verified
+            "email_verified": verified,
         }
         self.token = jwt.encode(self.data, "secret", algorithm="HS256")
         self.decoded = jwt.decode(self.token, "secret", algorithms=["HS256"])
 
 
 class TestAuthorizationSetup:
     def test_token_verification(self):
-        account = Acc(True)
+        account = Acc(verified=True)
         assert account.token != ""
         assert account.decoded == account.data
         assert True if verify_token(account.decoded) else False
@@ -37,36 +38,190 @@ def test_token_verification(self):
             verify_token(account.decoded)
 
     @pytest.mark.asyncio
-    async def test_user_get_registered(self, app: FastAPI, session: Session,
-                                       client: AsyncClient) -> None:
-        org = crud_organisation.add_organisation(session=session,
-                                                 name="Akvo",
-                                                 type="iNGO")
+    async def test_user_get_registered(
+        self, app: FastAPI, session: Session, client: AsyncClient
+    ) -> None:
+        org = crud_organisation.add_organisation(
+            session=session, name="Akvo", type="iNGO"
+        )
         org = org.serialize
         assert org["id"] == 1
         assert org["name"] == "Akvo"
         assert org["type"] == OrganisationType.iNGO
-        account = Acc(True)
+        account = Acc(verified=True)
         res = await client.post(
             app.url_path_for("user:register"),
             params={
                 "first_name": "Akvo",
                 "last_name": "Support",
-                "organisation": org["id"]
+                "organisation": org["id"],
             },
-            headers={"Authorization": f"Bearer {account.token}"})
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
         assert res.status_code == 200
         res = res.json()
-        user = crud_user.update_user_by_id(session=session,
-                                           id=1,
-                                           role="admin",
-                                           active=True)
+        user = crud_user.update_user_by_id(
+            session=session, id=1, role="admin", active=True
+        )
         assert res["email"] == user.email
         assert res["active"] is False
         res = await client.get(
             app.url_path_for("user:me"),
-            headers={"Authorization": f"Bearer {account.token}"})
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
         assert res.status_code == 200
         res = res.json()
         assert res["active"] is user.active
         assert res["role"] == "admin"
+
+    @pytest.mark.asyncio
+    async def test_get_user(
+        self, app: FastAPI, session: Session, client: AsyncClient
+    ) -> None:
+        account = Acc(verified=True)
+        res = await client.get(
+            app.url_path_for("user:get"),
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
+        assert res.status_code == 404
+
+        res = await client.get(
+            app.url_path_for("user:get"),
+            params={"page": 2},
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
+        assert res.status_code == 404
+        # get active user
+        res = await client.get(
+            app.url_path_for("user:get"),
+            params={"active": 1},
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
+        assert res.status_code == 200
+        res = res.json()
+        assert res["current"] == 1
+        assert res["total"] == 1
+        assert res["total_page"] == 1
+        assert len(res["data"]) == 1
+        assert res["data"] == [
+            {
+                "id": 1,
+                "email": "support@akvo.org",
+                "name": "Akvo Support",
+                "role": "admin",
+                "active": True,
+                "email_verified": True,
+                "picture": None,
+                "organisation": 1,
+            }
+        ]
+        # register as new user
+        new_account = Acc(
+            verified=True, email="john_doe@mail.com", name="John Doe")
+        res = await client.post(
+            app.url_path_for("user:register"),
+            params={
+                "first_name": "John",
+                "last_name": "Doe",
+                "organisation": 1,
+            },
+            headers={"Authorization": f"Bearer {new_account.token}"},
+        )
+        assert res.status_code == 200
+        # get non active user
+        res = await client.get(
+            app.url_path_for("user:get"),
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
+        assert res.status_code == 200
+        res = res.json()
+        assert res["current"] == 1
+        assert res["total"] == 1
+        assert res["total_page"] == 1
+        assert len(res["data"]) == 1
+        assert res["data"] == [
+            {
+                "id": 2,
+                "email": "john_doe@mail.com",
+                "name": "John Doe",
+                "role": "user",
+                "active": False,
+                "email_verified": None,
+                "picture": None,
+                "organisation": 1,
+            }
+        ]
+        # get user by id
+        res = await client.get(
+            app.url_path_for("user:get_by_id", id=3),
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
+        assert res.status_code == 404
+
+        res = await client.get(
+            app.url_path_for("user:get_by_id", id=1),
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
+        assert res.status_code == 200
+        res = res.json()
+        assert res == {
+            "id": 1,
+            "email": "support@akvo.org",
+            "name": "Akvo Support",
+            "role": "admin",
+            "active": True,
+            "access": [],
+            "organisation": 1
+        }
+
+    @pytest.mark.asyncio
+    async def test_update_user(
+        self, app: FastAPI, session: Session, client: AsyncClient
+    ) -> None:
+        account = Acc(verified=True)
+        res = await client.get(
+            app.url_path_for("user:get_by_id", id=2),
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
+        assert res.status_code == 200
+        user = res.json()
+        # update John Doe
+        res = await client.put(
+            app.url_path_for("user:update", id=user["id"]),
+            params={
+                "active": True,
+                "role": "admin",
+                "first_name": "John",
+                "last_name": "Doe",
+                "organisation": 1,
+            },
+            headers={"Authorization": f"Bearer {account.token}"},
+        )
+        assert res.status_code == 200
+        res = res.json()
+        assert res == {
+            "id": 2,
+            "email": "john_doe@mail.com",
+            "name": "John Doe",
+            "role": "admin",
+            "active": True,
+            "access": [],
+            "organisation": 1
+        }
+
+        @pytest.mark.asyncio
+        async def test_delete_user(
+            self, app: FastAPI, session: Session, client: AsyncClient
+        ) -> None:
+            account = Acc(verified=True)
+            res = await client.get(
+                app.url_path_for("user:get_by_id", id=2),
+                headers={"Authorization": f"Bearer {account.token}"},
+            )
+            assert res.status_code == 200
+            user = res.json()
+            res = await client.delete(
+                app.url_path_for("user:delete", id=user["id"]),
+                headers={"Authorization": f"Bearer {account.token}"},
+            )
+            assert res.status_code == 204

backend/tests/test_03_form.py

@@ -205,7 +205,8 @@ async def test_datapoint_name_question(self, app: FastAPI,
                 "form": 1,
                 "question_group": "Test Question Group",
                 "meta": True,
-                "type": "text"
+                "type": "text",
+                "required": False,
             },
             json={"dependency": [{
                 "id": 1,
@@ -221,7 +222,7 @@ async def test_datapoint_name_question(self, app: FastAPI,
         assert res["order"] == 4
         assert res["name"] == "Test Datapoint Text Question"
         assert res["meta"] is True
-        assert res["required"] is True
+        assert res["required"] is False
         assert res["rule"] is None
         assert res["type"] == "text"
         assert res["dependency"] == [{