add endpoint to enable/disable workspace notifications for user

backend/tracim_backend/tests/functional/test_user.py

@@ -2382,6 +2382,444 @@ class TestUserSetWorkspaceAsRead(FunctionalTest):
         )
 
 
+class TestUserEnableWorkspaceNotification(FunctionalTest):
+    """
+    Tests for /api/v2/users/{user_id}/workspaces/{workspace_id}/notify
+    """
+    def test_api_enable_user_workspace_notification__ok__200__admin(self):
+        # init DB
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        admin = dbsession.query(models.User) \
+            .filter(models.User.email == 'admin@admin.admin') \
+            .one()
+        workspace_api = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config
+
+        )
+        workspace = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        ).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        uapi = UserApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        gapi = GroupApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        groups = [gapi.get_one_with_name('users')]
+        test_user = uapi.create_user(
+            email='test@test.test',
+            password='pass',
+            name='bob',
+            groups=groups,
+            timezone='Europe/Paris',
+            lang='fr',
+            do_save=True,
+            do_notify=False,
+        )
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        rapi.create_one(test_user, workspace, UserRoleInWorkspace.READER, with_notif=False)  # nopep8
+        transaction.commit()
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is False
+        self.testapp.authorization = (
+            'Basic',
+            (
+                'admin@admin.admin',
+                'admin@admin.admin'
+            )
+        )
+        self.testapp.put_json('/api/v2/users/{user_id}/workspaces/{workspace_id}/notify'.format(  # nopep8
+            user_id=test_user.user_id,
+            workspace_id=workspace.workspace_id
+        ), status=204)
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is True
+
+    def test_api_enable_user_workspace_notification__ok__200__user_itself(self):
+        # init DB
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        admin = dbsession.query(models.User) \
+            .filter(models.User.email == 'admin@admin.admin') \
+            .one()
+        workspace_api = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config
+
+        )
+        workspace = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        ).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        uapi = UserApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        gapi = GroupApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        groups = [gapi.get_one_with_name('users')]
+        test_user = uapi.create_user(
+            email='test@test.test',
+            password='pass',
+            name='bob',
+            groups=groups,
+            timezone='Europe/Paris',
+            lang='fr',
+            do_save=True,
+            do_notify=False,
+        )
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        rapi.create_one(test_user, workspace, UserRoleInWorkspace.READER, with_notif=False)  # nopep8
+        transaction.commit()
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is False
+        self.testapp.authorization = (
+            'Basic',
+            (
+                'test@test.test',
+                'pass',
+            )
+        )
+        self.testapp.put_json('/api/v2/users/{user_id}/workspaces/{workspace_id}/notify'.format(  # nopep8
+            user_id=test_user.user_id,
+            workspace_id=workspace.workspace_id
+        ), status=204)
+
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is True
+
+    def test_api_enable_user_workspace_notification__err__403__other_user(self):
+        # init DB
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        admin = dbsession.query(models.User) \
+            .filter(models.User.email == 'admin@admin.admin') \
+            .one()
+        workspace_api = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config
+
+        )
+        workspace = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        ).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        uapi = UserApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        gapi = GroupApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        groups = [gapi.get_one_with_name('users')]
+        test_user = uapi.create_user(
+            email='test@test.test',
+            password='pass',
+            name='bob',
+            groups=groups,
+            timezone='Europe/Paris',
+            lang='fr',
+            do_save=True,
+            do_notify=False,
+        )
+        test_user2 = uapi.create_user(
+            email='test2@test2.test2',
+            password='pass',
+            name='boby',
+            groups=groups,
+            timezone='Europe/Paris',
+            lang='fr',
+            do_save=True,
+            do_notify=False,
+        )
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        rapi.create_one(test_user, workspace, UserRoleInWorkspace.READER, with_notif=False)  # nopep8
+        rapi.create_one(test_user2, workspace, UserRoleInWorkspace.READER, with_notif=False)  # nopep8
+        transaction.commit()
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is False
+        self.testapp.authorization = (
+            'Basic',
+            (
+                'test2@test2.test2',
+                'pass',
+            )
+        )
+        self.testapp.put_json('/api/v2/users/{user_id}/workspaces/{workspace_id}/notify'.format(  # nopep8
+            user_id=test_user.user_id,
+            workspace_id=workspace.workspace_id
+        ), status=403)
+
+
+class TestUserDisableWorkspaceNotification(FunctionalTest):
+    """
+    Tests for /api/v2/users/{user_id}/workspaces/{workspace_id}/unnotify
+    """
+    def test_api_disable_user_workspace_notification__ok__200__admin(self):
+        # init DB
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        admin = dbsession.query(models.User) \
+            .filter(models.User.email == 'admin@admin.admin') \
+            .one()
+        workspace_api = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config
+
+        )
+        workspace = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        ).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        uapi = UserApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        gapi = GroupApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        groups = [gapi.get_one_with_name('users')]
+        test_user = uapi.create_user(
+            email='test@test.test',
+            password='pass',
+            name='bob',
+            groups=groups,
+            timezone='Europe/Paris',
+            lang='fr',
+            do_save=True,
+            do_notify=True,
+        )
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        rapi.create_one(test_user, workspace, UserRoleInWorkspace.READER, with_notif=True)  # nopep8
+        transaction.commit()
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is True
+        self.testapp.authorization = (
+            'Basic',
+            (
+                'admin@admin.admin',
+                'admin@admin.admin'
+            )
+        )
+        self.testapp.put_json('/api/v2/users/{user_id}/workspaces/{workspace_id}/unnotify'.format(  # nopep8
+            user_id=test_user.user_id,
+            workspace_id=workspace.workspace_id
+        ), status=204)
+
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is False
+
+    def test_api_enable_user_workspace_notification__ok__200__user_itself(self):
+        # init DB
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        admin = dbsession.query(models.User) \
+            .filter(models.User.email == 'admin@admin.admin') \
+            .one()
+        workspace_api = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config
+
+        )
+        workspace = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        ).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        uapi = UserApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        gapi = GroupApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        groups = [gapi.get_one_with_name('users')]
+        test_user = uapi.create_user(
+            email='test@test.test',
+            password='pass',
+            name='bob',
+            groups=groups,
+            timezone='Europe/Paris',
+            lang='fr',
+            do_save=True,
+            do_notify=False,
+        )
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        rapi.create_one(test_user, workspace, UserRoleInWorkspace.READER, with_notif=True)  # nopep8
+        transaction.commit()
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is True
+        self.testapp.authorization = (
+            'Basic',
+            (
+                'test@test.test',
+                'pass',
+            )
+        )
+        self.testapp.put_json('/api/v2/users/{user_id}/workspaces/{workspace_id}/unnotify'.format(  # nopep8
+            user_id=test_user.user_id,
+            workspace_id=workspace.workspace_id
+        ), status=204)
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is False
+
+    def test_api_disable_user_workspace_notification__err__403__other_user(self):
+        # init DB
+        dbsession = get_tm_session(self.session_factory, transaction.manager)
+        admin = dbsession.query(models.User) \
+            .filter(models.User.email == 'admin@admin.admin') \
+            .one()
+        workspace_api = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config
+
+        )
+        workspace = WorkspaceApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        ).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        uapi = UserApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        gapi = GroupApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        groups = [gapi.get_one_with_name('users')]
+        test_user = uapi.create_user(
+            email='test@test.test',
+            password='pass',
+            name='bob',
+            groups=groups,
+            timezone='Europe/Paris',
+            lang='fr',
+            do_save=True,
+            do_notify=False,
+        )
+        test_user2 = uapi.create_user(
+            email='test2@test2.test2',
+            password='pass',
+            name='boby',
+            groups=groups,
+            timezone='Europe/Paris',
+            lang='fr',
+            do_save=True,
+            do_notify=False,
+        )
+        rapi = RoleApi(
+            current_user=admin,
+            session=dbsession,
+            config=self.app_config,
+        )
+        rapi.create_one(test_user, workspace, UserRoleInWorkspace.READER, with_notif=True)  # nopep8
+        rapi.create_one(test_user2, workspace, UserRoleInWorkspace.READER, with_notif=False)  # nopep8
+        transaction.commit()
+        role = rapi.get_one(test_user.user_id, workspace.workspace_id)
+        assert role.do_notify is True
+        self.testapp.authorization = (
+            'Basic',
+            (
+                'test2@test2.test2',
+                'pass',
+            )
+        )
+        self.testapp.put_json('/api/v2/users/{user_id}/workspaces/{workspace_id}/unnotify'.format(  # nopep8
+            user_id=test_user.user_id,
+            workspace_id=workspace.workspace_id
+        ), status=403)
+
+
 class TestUserWorkspaceEndpoint(FunctionalTest):
     """
     Tests for /api/v2/users/{user_id}/workspaces
@@ -4235,7 +4673,7 @@ class TestSetUserInfoEndpoint(FunctionalTest):
         params = {
             'public_name': 'updated',
             'timezone': 'Europe/London',
-            'lang' : 'en',
+            'lang': 'en',
         }
         self.testapp.put_json(
             '/api/v2/users/{}'.format(user_id),

backend/tracim_backend/views/core_api/user_controller.py

@@ -1,4 +1,5 @@
 from pyramid.config import Configurator
+from tracim_backend.lib.core.userworkspace import RoleApi
 from tracim_backend.lib.utils.utils import password_generator
 
 try:  # Python 3.5+
@@ -458,6 +459,60 @@ class UserController(Controller):
         api.mark_read__workspace(request.current_workspace)
         return
 
+    @hapic.with_api_doc(tags=[SWAGGER_TAG__USER_ENDPOINTS])
+    @require_same_user_or_profile(Group.TIM_ADMIN)
+    @hapic.input_path(UserWorkspaceIdPathSchema())
+    @hapic.output_body(NoContentSchema(), default_http_code=HTTPStatus.NO_CONTENT)  # nopep8
+    def enable_workspace_notification(self, context, request: TracimRequest, hapic_data=None):  # nopep8
+        """
+        enable workspace notification
+        """
+        app_config = request.registry.settings['CFG']
+        api = ContentApi(
+            current_user=request.candidate_user,  # User
+            session=request.dbsession,
+            config=app_config,
+        )
+        wapi = WorkspaceApi(
+            current_user=request.candidate_user,  # User
+            session=request.dbsession,
+            config=app_config,
+        )
+        workspace = wapi.get_one(hapic_data.path.workspace_id)
+        wapi.enable_notifications(request.candidate_user, workspace)
+        rapi = RoleApi(
+            current_user=request.candidate_user,  # User
+            session=request.dbsession,
+            config=app_config,
+        )
+        role = rapi.get_one(request.candidate_user.user_id, workspace.workspace_id)
+        wapi.save(workspace)
+        return
+
+    @hapic.with_api_doc(tags=[SWAGGER_TAG__USER_ENDPOINTS])
+    @require_same_user_or_profile(Group.TIM_ADMIN)
+    @hapic.input_path(UserWorkspaceIdPathSchema())
+    @hapic.output_body(NoContentSchema(), default_http_code=HTTPStatus.NO_CONTENT)  # nopep8
+    def disable_workspace_notification(self, context, request: TracimRequest, hapic_data=None):  # nopep8
+        """
+        disable workspace notification
+        """
+        app_config = request.registry.settings['CFG']
+        api = ContentApi(
+            current_user=request.candidate_user,  # User
+            session=request.dbsession,
+            config=app_config,
+        )
+        wapi = WorkspaceApi(
+            current_user=request.candidate_user,  # User
+            session=request.dbsession,
+            config=app_config,
+        )
+        workspace = wapi.get_one(hapic_data.path.workspace_id)
+        wapi.disable_notifications(request.candidate_user, workspace)
+        wapi.save(workspace)
+        return
+
     def bind(self, configurator: Configurator) -> None:
         """
         Create all routes and views using pyramid configurator
@@ -532,3 +587,11 @@ class UserController(Controller):
         # set workspace as read
         configurator.add_route('read_workspace', '/users/{user_id}/workspaces/{workspace_id}/read', request_method='PUT')  # nopep8
         configurator.add_view(self.set_workspace_as_read, route_name='read_workspace')  # nopep8
+
+        # enable workspace notification
+        configurator.add_route('enable_workspace_notification', '/users/{user_id}/workspaces/{workspace_id}/notify', request_method='PUT')  # nopep8
+        configurator.add_view(self.enable_workspace_notification, route_name='enable_workspace_notification')  # nopep8
+
+        # enable workspace notification
+        configurator.add_route('disable_workspace_notification', '/users/{user_id}/workspaces/{workspace_id}/unnotify', request_method='PUT')  # nopep8
+        configurator.add_view(self.disable_workspace_notification, route_name='disable_workspace_notification')  # nopep8