Tidy and improve unit tests (#111)

* tests: Stop using global app context in fixtures

* feat: Use app ctx in fileio tests

* tests: Move citation/fulltext fixture files

* feat: Remove custom constructors from db models

* fix: Standardize db model creation calls

* feat: Make db model reprs more consistent

* tests: Add helper func to temp set current user

* feat: Allow nonexistent users for part of auth

* tests: Improve user unit tests

* tests: Improve users api unit tests

* tests: Add better delete user test

* refactor: Improve reviews is-allowed logic

* tests: Add richer reviews seed data

* tests: Improve review endpoint tests

* fix: Hide v2/v1 schema diff in post reviews

* tests: Improve reviews endpoint tests

* refactor: Standarize study api allow check

* tests: Improve study api tests

* tests: Fix broken test from seed id seq tweaks

* tests: Improve studies api tests

* fix: Minor bugs in get studies sql subqueries

* tests: Add more get studies test cases

and fix a bug in citation text content, ugh

* tests: Add get studies pagination tests

* feat: Order screenings on study relationship

* feat: Tidy rough edges in citation scrn api

* tests: Fix seed data to be "correct"

* tests: Improve citation screening api tests

colandr/apis/auth.py

@@ -64,7 +64,10 @@ def post(self, email, password):
         $ curl http://localhost:5000/api/auth/login -X POST \
             -d '{"email":"foo@gmail.com","password":"PASSWORD"}'
         """
-        user = authenticate_user(email, password)
+        try:
+            user = authenticate_user(email, password)
+        except ValueError:
+            return not_found_error("no user found matching given email and password")
         access_token = jwtext.create_access_token(identity=user, fresh=True)
         refresh_token = jwtext.create_refresh_token(identity=user)
         current_app.logger.info("%s logged in", user)
@@ -322,7 +325,7 @@ def user_identity_loader(user: t.Union[User, str]):
 
 
 @jwt.user_lookup_loader
-def user_lookup_callback(_jwt_header, jwt_data: dict) -> User:
+def user_lookup_callback(_jwt_header, jwt_data: dict) -> t.Optional[User]:
     """
     Callback function that loads a user from the database by its identity (id)
     whenever a protected API route is accessed.
@@ -336,8 +339,7 @@ def user_lookup_callback(_jwt_header, jwt_data: dict) -> User:
             sa.select(User).filter_by(email=identity)
         ).scalar_one_or_none()
     else:
-        raise TypeError()
-    assert user is not None  # type guard
+        raise TypeError(f"user identity={identity} is invalid")
     return user
 
 

colandr/apis/resources/citation_imports.py

@@ -189,8 +189,8 @@ def post(
         ).scalar_one_or_none()
         if data_source is None:
             data_source = models.DataSource(
-                source_type, source_name, source_url=source_url
-            )
+                source_type=source_type, source_name=source_name, source_url=source_url
+            )  # type: ignore
             db.session.add(data_source)
         db.session.commit()
         current_app.logger.info("inserted %s", data_source)
@@ -235,8 +235,13 @@ def post(
         db.session.execute(sa.insert(models.Study), studies_to_insert)
         # as well as a record of the import
         citations_import = models.Import(
-            review_id, user_id, data_source_id, "citation", n_citations, status=status
-        )
+            review_id=review_id,
+            user_id=user_id,
+            data_source_id=data_source_id,
+            record_type="citation",
+            num_records=n_citations,
+            status=status,
+        )  # type: ignore
         db.session.add(citations_import)
         db.session.commit()
         current_app.logger.info(

colandr/apis/resources/citation_screenings.py

@@ -67,26 +67,22 @@ def get(self, id, fields):
         study = db.session.get(models.Study, id)
         if not study:
             return not_found_error(f"<Study(id={id})> not found")
-        if (
-            current_user.is_admin is False
-            and db.session.execute(
-                current_user.review_user_assoc.select().filter_by(
-                    review_id=study.review_id
-                )
-            ).one_or_none()
-            is None
-        ):
+        if not _is_allowed(current_user, study.review_id):
             return forbidden_error(
                 f"{current_user} forbidden to get citation screenings for this review"
             )
-        current_app.logger.debug("got %s", study)
         screenings = db.session.execute(
             study.screenings.select().filter_by(stage="citation")
         ).scalars()
+        if not screenings:
+            return not_found_error(f"no screenings for <Study(id={id})> found")
         # HACK: hide the consolidated (v2) screening schema from this api
-        if fields and "citation_id" in fields:
-            fields.append("study_id")
-            fields.remove("citation_id")
+        if fields:
+            if "id" not in fields:
+                fields.append("id")
+            if "citation_id" in fields:
+                fields.append("study_id")
+                fields.remove("citation_id")
         screenings_dumped = [
             _convert_screening_v2_into_v1(record)
             for record in ScreeningV2Schema(many=True, only=fields).dump(screenings)
@@ -116,14 +112,7 @@ def delete(self, id):
         study = db.session.get(models.Study, id)
         if not study:
             return not_found_error(f"<Study(id={id})> not found")
-        if (
-            db.session.execute(
-                current_user.review_user_assoc.select().filter_by(
-                    review_id=study.review_id
-                )
-            ).one_or_none()
-            is None
-        ):
+        if not _is_allowed(current_user, study.review_id):
             return forbidden_error(
                 f"{current_user} forbidden to delete citation screening for this review"
             )
@@ -167,16 +156,8 @@ def post(self, args, id):
         # check current user authorization
         study = db.session.get(models.Study, id)
         if not study:
-            return not_found_error(f"<Citation(id={id})> not found")
-        if (
-            current_user.is_admin is False
-            and db.session.execute(
-                current_user.review_user_assoc.select().filter_by(
-                    review_id=study.review_id
-                )
-            ).one_or_none()
-            is None
-        ):
+            return not_found_error(f"<Study(id={id})> not found")
+        if not _is_allowed(current_user, study.review_id):
             return forbidden_error(
                 f"{current_user} forbidden to screen citations for this review"
             )
@@ -201,13 +182,13 @@ def post(self, args, id):
             return forbidden_error(f"{current_user} has already screened {study}")
 
         screening = models.Screening(
-            user_id,
-            study.review_id,
-            id,
-            "citation",
-            args["status"],
-            args["exclude_reasons"],
-        )
+            user_id=user_id,
+            review_id=study.review_id,
+            study_id=id,
+            stage="citation",
+            status=args["status"],
+            exclude_reasons=args["exclude_reasons"],
+        )  # type: ignore
         study.screenings.add(screening)
         db.session.commit()
         current_app.logger.info("inserted %s", screening)
@@ -225,7 +206,7 @@ def post(self, args, id):
     @use_args(
         ScreeningSchema(
             only=["user_id", "status", "exclude_reasons"],
-            partial=["status", "exclude_reasons"],
+            partial=["exclude_reasons"],
         ),
         location="json",
     )
@@ -243,7 +224,7 @@ def put(self, args, id):
         current_user = jwtext.get_current_user()
         study = db.session.get(models.Study, id)
         if not study:
-            return not_found_error(f"<Citation(id={id})> not found")
+            return not_found_error(f"<Study(id={id})> not found")
         if current_user.is_admin is True and "user_id" in args:
             screening = db.session.execute(
                 study.screenings.select().filter_by(
@@ -258,7 +239,7 @@ def put(self, args, id):
             ).scalar_one_or_none()
         if not screening:
             return not_found_error(f"{current_user} has not screened this citation")
-        if args["status"] == "excluded" and not args["exclude_reasons"]:
+        if args["status"] == "excluded" and not args.get("exclude_reasons"):
             return bad_request_error("screenings that exclude must provide a reason")
         for key, value in args.items():
             if key is missing:
@@ -342,15 +323,7 @@ def get(self, citation_id, user_id, review_id, status_counts):
             study = db.session.get(models.Study, citation_id)
             if not study:
                 return not_found_error(f"<Study(id={citation_id})> not found")
-            if (
-                current_user.is_admin is False
-                and db.session.execute(
-                    study.review.review_user_assoc.select().filter_by(
-                        user_id=current_user.id
-                    )
-                ).one_or_none()
-                is None
-            ):
+            if not _is_allowed(current_user, study.review_id):
                 return forbidden_error(
                     f"{current_user} forbidden to get screenings for {study}"
                 )
@@ -515,3 +488,17 @@ def _convert_screening_v2_into_v1(record) -> dict:
     except KeyError:
         pass
     return record
+
+
+def _is_allowed(current_user: models.User, review_id: int) -> bool:
+    is_allowed = current_user.is_admin
+    is_allowed = (
+        is_allowed
+        or db.session.execute(
+            sa.select(models.ReviewUserAssoc).filter_by(
+                user_id=current_user.id, review_id=review_id
+            )
+        ).scalar_one_or_none()
+        is not None
+    )
+    return is_allowed

colandr/apis/resources/citations.py

@@ -259,22 +259,20 @@ def post(self, args, review_id, source_type, source_name, source_url, status):
         ).scalar_one_or_none()
         if data_source is None:
             data_source = models.DataSource(
-                source_type, source_name, source_url=source_url
-            )
+                source_type=source_type, source_name=source_name, source_url=source_url
+            )  # type: ignore
             db.session.add(data_source)
         db.session.commit()
         current_app.logger.info("inserted %s", data_source)
 
         # add the study w/ citation
         citation = CitationSchema().load(args)
         study = models.Study(
-            **{
-                "user_id": current_user.id,
-                "review_id": review_id,
-                "data_source_id": data_source.id,
-                "citation": citation,
-            }
-        )
+            user_id=current_user.id,
+            review_id=review_id,
+            data_source_id=data_source.id,
+            citation=citation,
+        )  # type: ignore
         if status is not None:
             study.citation_status = status
         db.session.add(study)

colandr/apis/resources/fulltext_screenings.py

@@ -202,13 +202,13 @@ def post(self, args, id):
             return forbidden_error(f"{current_user} has already screened {study}")
 
         screening = models.Screening(
-            user_id,
-            study.review_id,
-            id,
-            "fulltext",
-            args["status"],
-            args["exclude_reasons"],
-        )
+            user_id=user_id,
+            review_id=study.review_id,
+            study_id=id,
+            stage="fulltext",
+            status=args["status"],
+            exclude_reasons=args["exclude_reasons"],
+        )  # type: ignore
         study.screenings.add(screening)
         db.session.commit()
         current_app.logger.info("inserted %s", screening)

colandr/apis/resources/reviews.py

@@ -220,7 +220,22 @@ def post(self, args):
         """create new review"""
         current_user = jwtext.get_current_user()
         name = args.pop("name")
-        review = models.Review(name=name, **args)
+        # HACK: convert from v1 to v2 review schema here
+        if "num_citation_screening_reviewers" in args:
+            args["citation_reviewer_num_pcts"] = [
+                {
+                    "num": args.pop("num_citation_screening_reviewers"),
+                    "pct": 100,
+                }
+            ]
+        if "num_fulltext_screening_reviewers" in args:
+            args["fulltext_reviewer_num_pcts"] = [
+                {
+                    "num": args.pop("num_fulltext_screening_reviewers"),
+                    "pct": 100,
+                }
+            ]
+        review = models.Review(name=name, **args)  # type: ignore
         # TODO: do we want to allow admins to set other users as owners?
         review.review_user_assoc.append(
             models.ReviewUserAssoc(review, current_user, "owner")
@@ -246,19 +261,16 @@ def _is_allowed(
     current_user: models.User, review_id: int, *, members: bool = True
 ) -> bool:
     is_allowed = current_user.is_admin
-    if members:
-        is_allowed = (
-            is_allowed
-            or db.session.execute(
-                current_user.review_user_assoc.select().filter_by(review_id=review_id)
-            ).one_or_none()
-            is not None
-        )
-    else:
-        is_allowed = is_allowed or any(
-            review.id == review_id for review in current_user.owned_reviews
-        )
-
+    user_roles = ["owner", "member"] if members else ["owner"]
+    is_allowed = (
+        is_allowed
+        or db.session.execute(
+            sa.select(models.ReviewUserAssoc)
+            .filter_by(user_id=current_user.id, review_id=review_id)
+            .where(models.ReviewUserAssoc.user_role == sa.any_(user_roles))
+        ).scalar_one_or_none()
+        is not None
+    )
     return is_allowed