Merge pull request #701 from SUNET/ylle_remove_dead_code

Remove_dead_code

ruff.toml

@@ -3,6 +3,6 @@ line-length = 120
 target-version = "py310"
 
 [lint]
-select = ["E", "F", "W", "I", "ASYNC", "UP", "FLY", "PERF", "FURB"]
+select = ["E", "F", "W", "I", "ASYNC", "UP", "FLY", "PERF", "FURB", "ERA"]
 
 ignore = ["E501"]

src/eduid/queue/db/change_event.py

@@ -55,7 +55,7 @@ class ChangeEvent:
     full_document: dict[str, Any] | None = None
     to: NS | None = None
     update_description: UpdateDescription | None = None
-    # Available in MongoDB >=4
+    # Available in MongoDB >=4  # noqa: ERA001
     # clusterTime
     # txnNumber
     # lsid

src/eduid/satosa/entity_category/auth_server.py

@@ -23,6 +23,7 @@
     SUNETAUTHSERVERv1: True,
 }
 
+# ruff: noqa: ERA001
 # These restrictions are parsed (and validated) into a list of saml2.assertion.EntityCategoryRule instances.
 RESTRICTIONS: list[dict[str, Any]] = [
     # Example of conversion of some of the rules in RELEASE to this new format:

src/eduid/scimapi/middleware.py

@@ -55,33 +55,9 @@ class ScimMiddleware(BaseMiddleware):
     async def dispatch(self, req: Request, call_next) -> Response:
         req = self.make_context_request(request=req, context_class=ScimApiContext)
         self.context.logger.debug(f"process_request: {req.method} {req.url.path}")
-        # TODO: fix me? is this needed?
-        # if req.method == 'POST':
-        #     if req.path == '/login':
-        #         if req.content_type != 'application/json':
-        #             raise UnsupportedMediaTypeMalformed(
-        #                 detail=f'{req.content_type} is an unsupported media type for {req.path}'
-        #             )
-        #     elif req.path == '/notifications':
-        #         if req.content_type == 'text/plain; charset=UTF-8':
-        #             # We know the body is json, set the correct content type
-        #             req.content_type = 'application/json'
-        #     elif req.content_type != 'application/scim+json':
-        #         raise UnsupportedMediaTypeMalformed(detail=f'{req.content_type} is an unsupported media type')
         resp = await call_next(req)
 
         self.context.logger.debug(f"process_response: {req.method} {req.url.path}")
-        # Default to 'application/json' if responding with an error message
-        # if req_succeeded and resp.body:
-        #     # candidates should be sorted by increasing desirability
-        #     # preferred = request.client_prefers(('application/json', 'application/scim+json'))
-        #
-        #     preferred = None
-        #     self.context.logger.debug(f'Client prefers content-type {preferred}')
-        #     if preferred is None:
-        #         preferred = 'application/scim+json'
-        #         self.context.logger.debug(f'Default content-type {preferred} used')
-        #     resp.headers.content_type = preferred
         return resp
 
 

src/eduid/scimapi/routers/invites.py

@@ -72,83 +72,13 @@ async def on_put(
         raise BadRequest(detail="Invite completed and cannot be updated")
 
     invite_changed = False
-    profiles_changed = False
 
     if update_request.nutid_invite_v1.completed is not None:
         signup_invite = replace(signup_invite, completed_ts=update_request.nutid_invite_v1.completed)
         db_invite = replace(db_invite, completed=update_request.nutid_invite_v1.completed)
         invite_changed = True
 
-    # TODO: decide what can be updated
-    # # Update the invite
-    # invite_changed = False
-    # if SCIMSchema.NUTID_INVITE_V1 in update_request.schemas:
-    #     name_in = ScimApiName(**update_request.nutid_invite_v1.name.dict(exclude_none=True))
-    #     emails_in = set(ScimApiEmail(**email.dict()) for email in update_request.nutid_invite_v1.emails)
-    #     phone_numbers_in = set(
-    #         ScimApiPhoneNumber(**number.dict()) for number in update_request.nutid_invite_v1.phone_numbers
-    #     )
-    #     # external_id
-    #     if update_request.external_id != db_invite.external_id:
-    #         db_invite = replace(db_invite, external_id=update_request.external_id)
-    #         invite_changed = True
-    #     # preferred_language
-    #     if update_request.nutid_invite_v1.preferred_language != db_invite.preferred_language:
-    #         db_invite = replace(db_invite, preferred_language=update_request.nutid_invite_v1.preferred_language)
-    #         invite_changed = True
-    #     # name
-    #     if name_in != db_invite.name:
-    #         db_invite = replace(db_invite, name=name_in)
-    #         invite_changed = True
-    #     # emails
-    #     if emails_in != set(db_invite.emails):
-    #         db_invite = replace(db_invite, emails=list(emails_in))
-    #         invite_changed = True
-    #     # phone_numbers
-    #     if phone_numbers_in != set(db_invite.phone_numbers):
-    #         db_invite = replace(db_invite, phone_numbers=list(phone_numbers_in))
-    #         invite_changed = True
-    #     # nin
-    #     if update_request.nutid_invite_v1.national_identity_number != db_invite.national_identity_number:
-    #         db_invite = replace(
-    #             db_invite, national_identity_number=update_request.nutid_invite_v1.national_identity_number
-    #         )
-    #         invite_changed = True
-    #     # finish_url
-    #     if update_request.nutid_invite_v1.finish_url != db_invite.finish_url:
-    #         db_invite = replace(db_invite, finish_url=update_request.nutid_invite_v1.finish_url)
-    #         invite_changed = True
-    #     # completed_ts
-    #     if update_request.nutid_invite_v1.completed != db_invite.completed_ts:
-    #         db_invite = replace(db_invite, completed_ts=update_request.nutid_invite_v1.completed)
-    #         invite_changed = True
-    #
-    # profiles_changed = False
-    # if SCIMSchema.NUTID_USER_V1 in update_request.schemas and update_request.nutid_user_v1 is not None:
-    #
-    #     # Look for changes in profiles
-    #     for this in update_request.nutid_user_v1.profiles.keys():
-    #         if this not in db_invite.profiles:
-    #             req.app.context.logger.info(
-    #                 f"Adding profile {this}/{update_request.nutid_user_v1.profiles[this]} to invite"
-    #             )
-    #             profiles_changed = True
-    #         elif update_request.nutid_user_v1.profiles[this].to_dict() != db_invite.profiles[this].to_dict():
-    #             req.app.context.logger.info(f"Profile {this}/{update_request.nutid_user_v1.profiles[this]} updated")
-    #             profiles_changed = True
-    #         else:
-    #             req.app.context.logger.info(f"Profile {this}/{update_request.nutid_user_v1.profiles[this]} not changed")  # noqa: E501
-    #     for this in db_invite.profiles.keys():
-    #         if this not in update_request.nutid_user_v1.profiles:
-    #             req.app.context.logger.info(f"Profile {this}/{db_invite.profiles[this]} removed")
-    #             profiles_changed = True
-    #
-    #     if profiles_changed:
-    #         for profile_name, profile in update_request.nutid_user_v1.profiles.items():
-    #             db_profile = ScimApiProfile(attributes=profile.attributes, data=profile.data)
-    #             db_invite.profiles[profile_name] = db_profile
-
-    if invite_changed or profiles_changed:
+    if invite_changed:
         save_invite(
             req=req,
             db_invite=db_invite,

src/eduid/scimapi/routers/utils/groups.py

@@ -53,10 +53,6 @@ def db_group_to_response(req: ContextRequest, resp: Response, db_group: ScimApiG
 
     resp.headers["Location"] = location
     resp.headers["ETag"] = make_etag(db_group.version)
-    # TODO: Needed?
-    # if SCIMSchema.NUTID_GROUP_V1 not in group.schemas and SCIMSchema.NUTID_GROUP_V1.value in dumped_group:
-    #    # Serialization will always put the NUTID_GROUP_V1 in the dumped_group, even if there was no data
-    #    del dumped_group[SCIMSchema.NUTID_GROUP_V1.value]
     req.app.context.logger.debug(f"Extra debug: Response:\n{group.model_dump_json(exclude_none=True, indent=2)}")
     return group
 

src/eduid/scimapi/routers/utils/status.py

@@ -21,8 +21,7 @@ def check_mongo(req: ContextRequest, default_data_owner: str):
 def check_neo4j(req: ContextRequest, default_data_owner: str):
     group_db = req.app.context.get_groupdb(default_data_owner)
     try:
-        # TODO: Implement is_healthy
-        # db.is_healthy()
+        # TODO: Implement is_healthy, check if there is a better way for neo4j
         q = """
             MATCH (n)
             RETURN count(*) as exists LIMIT 1

src/eduid/scimapi/testing.py

@@ -51,7 +51,6 @@ def _get_config(self) -> dict[str, Any]:
             "data_owners": {"eduid.se": {"db_name": "eduid_se"}},
             "logging_config": {
                 "loggers": {
-                    #'eduid_groupdb': {'handlers': ['console'], 'level': 'DEBUG'},
                     "neo4j": {"handlers": ["console"], "level": "WARNING"},
                     "root": {"handlers": ["console"], "level": "DEBUG"},
                 },

src/eduid/scimapi/tests/test_notifications.py

@@ -7,9 +7,6 @@
 from eduid.scimapi.testing import ScimApiTestCase
 from eduid.userdb.scimapi import EventLevel
 
-# from moto import mock_sns
-
-
 __author__ = "lundberg"
 
 

src/eduid/scimapi/tests/test_sciminvite.py

@@ -341,14 +341,6 @@ def _perform_search(
         resources = response.json().get("Resources")
         return resources
 
-    # def test_get_invites(self):
-    #    for i in range(9):
-    #        self.add_user(identifier=str(uuid4()), external_id=f'test-id-{i}', profiles={'test': self.test_profile})
-    #    parsed_response = self.client.get(url=f'/Users', headers=self.headers)
-    #    self.assertEqual([SCIMSchema.API_MESSAGES_20_LIST_RESPONSE.value], parsed_response.json().get('schemas'))
-    #    resources = parsed_response.json().get('Resources')
-    #    self.assertEqual(self.userdb.db_count(), len(resources))
-
     def test_create_invite(self):
         req = {
             "schemas": [

src/eduid/scimapi/tests/test_scimuser.py

@@ -787,7 +787,6 @@ def _perform_search(
 class TestAsyncUserResource(IsolatedAsyncioTestCase, ScimApiTestCase):
     def setUp(self) -> None:
         super().setUp()
-        # ScimApiTestCase.setUp(self)
         self.async_client = AsyncClient(app=self.api, base_url="http://testserver")
         # create users
         self.user_count = 10

src/eduid/userdb/db/async_db.py

@@ -277,7 +277,7 @@ async def setup_indexes(self, indexes: Mapping[str, Any]) -> None:
         """
         To update an index add a new item in indexes and remove the previous version.
         """
-        # indexes={'index-name': {'key': [('key', 1)], 'param1': True, 'param2': False}, }
+        # indexes={'index-name': {'key': [('key', 1)], 'param1': True, 'param2': False}, }  # noqa: ERA001
         # http://docs.mongodb.org/manual/reference/method/db.collection.ensureIndex/
         default_indexes = ["_id_"]  # _id_ index can not be deleted from a mongo collection
         current_indexes = await self._coll.index_information()

src/eduid/userdb/db/sync_db.py

@@ -302,7 +302,7 @@ def setup_indexes(self, indexes: Mapping[str, Any]) -> None:
         """
         To update an index add a new item in indexes and remove the previous version.
         """
-        # indexes={'index-name': {'key': [('key', 1)], 'param1': True, 'param2': False}, }
+        # indexes={'index-name': {'key': [('key', 1)], 'param1': True, 'param2': False}, }  # noqa: ERA001
         # http://docs.mongodb.org/manual/reference/method/db.collection.ensureIndex/
         default_indexes = ["_id_"]  # _id_ index can not be deleted from a mongo collection
         current_indexes = self._coll.index_information()

src/eduid/userdb/orcid.py

@@ -16,7 +16,7 @@ class OidcIdToken(Element):
     iss: str
     # Subject identifier
     sub: str
-    # Audience(s)
+    # Audience(s)  # noqa: ERA001
     aud: list[str]
     # Expiration time
     exp: int

src/eduid/userdb/security/__init__.py

@@ -1,12 +1,7 @@
-from eduid.userdb.security.db import PasswordResetStateDB, SecurityUserDB
-from eduid.userdb.security.state import PasswordResetEmailAndPhoneState, PasswordResetEmailState, PasswordResetState
+from eduid.userdb.security.db import SecurityUserDB
 from eduid.userdb.security.user import SecurityUser
 
 __all__ = [
-    "PasswordResetState",
-    "PasswordResetEmailState",
-    "PasswordResetEmailAndPhoneState",
-    "PasswordResetStateDB",
     "SecurityUser",
     "SecurityUserDB",
 ]

src/eduid/userdb/security/db.py

@@ -1,12 +1,6 @@
-import copy
 import logging
-from collections.abc import Mapping
-from typing import Any
 
-from eduid.common.decorators import deprecated
-from eduid.userdb.db import BaseDB, SaveResult, TUserDbDocument
-from eduid.userdb.exceptions import MultipleDocumentsReturned
-from eduid.userdb.security.state import PasswordResetEmailAndPhoneState, PasswordResetEmailState, PasswordResetState
+from eduid.userdb.db import TUserDbDocument
 from eduid.userdb.security.user import SecurityUser
 from eduid.userdb.userdb import UserDB
 
@@ -22,89 +16,3 @@ def __init__(self, db_uri: str, db_name: str = "eduid_security", collection: str
     @classmethod
     def user_from_dict(cls, data: TUserDbDocument) -> SecurityUser:
         return SecurityUser.from_dict(data)
-
-
-# @deprecated("Remove once the password reset views are served from their own webapp")
-class PasswordResetStateDB(BaseDB):
-    @deprecated("Remove once the password reset views are served from their own webapp")
-    def __init__(self, db_uri, db_name="eduid_security", collection="password_reset_data"):
-        super().__init__(db_uri, db_name, collection=collection)
-
-    def get_state_by_email_code(self, email_code: str) -> PasswordResetState | None:
-        """
-        Locate a state in the db given the state's email code.
-
-        :param email_code: Code sent to the user
-
-        :return: state, if found
-
-        :raise self.MultipleDocumentsReturned: More than one document matches the search criteria
-        """
-        spec = {"email_code.code": email_code}
-        states = list(self._get_documents_by_filter(spec))
-
-        if len(states) == 0:
-            return None
-
-        if len(states) > 1:
-            raise MultipleDocumentsReturned(f"Multiple matching users for filter {filter!r}")
-
-        return self.init_state(states[0])
-
-    def get_state_by_eppn(self, eppn: str) -> PasswordResetState | None:
-        """
-        Locate a state in the db given the users eppn.
-
-        :param eppn: Users unique eppn
-
-        :return: state, if found
-
-        :raise self.MultipleDocumentsReturned: More than one document matches the search criteria
-        """
-        state = self._get_document_by_attr("eduPersonPrincipalName", eppn)
-        if state:
-            return self.init_state(state)
-        return None
-
-    @staticmethod
-    def init_state(
-        data: Mapping[str, Any],
-    ) -> PasswordResetEmailState | PasswordResetEmailAndPhoneState | None:
-        _data = dict(copy.deepcopy(data))  # to not modify callers data
-        method = _data.pop("method", None)
-        if method == "email":
-            return PasswordResetEmailState.from_dict(_data)
-        if method == "email_and_phone":
-            return PasswordResetEmailAndPhoneState.from_dict(_data)
-        return None
-
-    def save(self, state: PasswordResetState, is_in_database: bool) -> SaveResult:
-        """
-        Save state to the database.
-
-        :param state: The state to save
-        :param is_in_database: True if the state is already in the database. TODO: Remove when state have Meta.
-        """
-
-        data = state.to_dict()
-        # Remember what type of state this is, used when loading state above in init_state()
-        if isinstance(state, PasswordResetEmailAndPhoneState):
-            data["method"] = "email_and_phone"
-        elif isinstance(state, PasswordResetEmailState):
-            data["method"] = "email"
-
-        if state.modified_ts is None:
-            # Remove old reset password state
-            old_state = self.get_state_by_eppn(state.eppn)
-            if old_state:
-                self.remove_state(old_state)
-
-        spec: dict[str, Any] = {"eduPersonPrincipalName": state.eppn}
-
-        result = self._save(state.to_dict(), spec, is_in_database=is_in_database)
-        state.modified_ts = result.ts
-
-        return result
-
-    def remove_state(self, state: PasswordResetState) -> None:
-        self.remove_document({"eduPersonPrincipalName": state.eppn})