Merge pull request #195 from Normal-OJ/feat/force-batch-signup

feat: force batch signup

model/auth.py

@@ -275,23 +275,27 @@ def add_user(
 
 
 @auth_api.route('/batch-signup', methods=['POST'])
-@Request.json('new_users: str', 'course')
+@Request.json('new_users: str', 'course', 'force')
 @Request.doc('course', 'course', Course, null=True)
 @identity_verify(0)
 def batch_signup(
     user,
     new_users: str,
     course: Optional[Course],
+    force: Optional[bool],
 ):
     try:
         new_users = [*csv.DictReader(io.StringIO(new_users))]
     except csv.Error as e:
         current_app.logger.info(f'Error parse csv file [err={e}]')
         return HTTPError('Invalid file content', 400)
+    if force is None:
+        force = False
     try:
         new_users = User.batch_signup(
             new_users=new_users,
             course=course,
+            force=force,
         )
     except ValueError as e:
         return HTTPError(str(e), 400)

mongo/user.py

@@ -25,27 +25,32 @@
 class User(MongoBase, engine=engine.User):
 
     @classmethod
-    def signup(cls, username, password, email):
+    def signup(
+        cls,
+        username: str,
+        password: str,
+        email: str,
+    ):
         if re.match(r'^[a-zA-Z0-9_\-]+$', username) is None:
             raise ValueError(f'Invalid username [username={username}]')
-        user = cls(username)
-        user_id = hash_id(user.username, password)
+        user_id = hash_id(username, password)
         email = email.lower().strip()
-        cls.engine(
+        user = cls.engine(
             user_id=user_id,
             user_id2=user_id,
-            username=user.username,
+            username=username,
             email=email,
             md5=hashlib.md5(email.encode()).hexdigest(),
             active=False,
         ).save(force_insert=True)
-        return user.reload()
+        return cls(user).reload()
 
     @classmethod
     def batch_signup(
         cls,
         new_users: List[Dict[str, str]],
         course: Optional['Course'] = None,
+        force: bool = False,
     ):
         '''
         Register multiple students with course
@@ -68,22 +73,26 @@ def batch_signup(
         registered_users = []
         for u in new_users:
             try:
-                displayed_name = u.pop('displayedName', None)
-                if displayed_name is not None:
-                    activate_payload = {'displayedName': displayed_name}
-                else:
-                    activate_payload = {}
-                role = u.pop('role', None)
-                new_user = cls.signup(**u)
+                new_user = cls.signup(
+                    username=u['username'],
+                    password=u['password'],
+                    email=u['email'],
+                )
+                activate_payload = drop_none({
+                    'displayedName':
+                    u.get('displayedName'),
+                })
                 new_user.activate(activate_payload)
-                if role is not None:
+                if (role := u.get('role')) is not None:
                     new_user.update(role=role)
                     new_user.reload('role')
             except engine.NotUniqueError:
                 try:
                     new_user = cls.get_by_username(u['username'])
                 except engine.DoesNotExist:
                     new_user = cls.get_by_email(u['email'])
+                if force:
+                    new_user.force_update(u)
             registered_users.append(new_user)
         if course is not None:
             new_student_nicknames = {
@@ -94,6 +103,18 @@ def batch_signup(
             course.update_student_namelist(new_student_nicknames)
         return new_users
 
+    def force_update(self, new_user: Dict[str, Any]):
+        '''
+        Force update an existent user in batch update procedure
+        '''
+        if (displayed_name := new_user.get('displayedName')) is not None:
+            self.update(profile__displayed_name=displayed_name)
+        if (role := new_user.get('role')) is not None:
+            self.update(role=role)
+        if (password := new_user.get('password')) is not None:
+            self.change_password(password)
+        self.reload()
+
     @classmethod
     def login(cls, username, password):
         try:

mongo/utils.py

@@ -15,6 +15,7 @@
     'perm',
     'RedisCache',
     'doc_required',
+    'drop_none',
 )
 
 

tests/base_tester.py

@@ -1,10 +1,14 @@
 import secrets
-from typing import Union, Literal
+import typing
+from typing import Literal, Tuple, Dict, Any, Union
 from mongoengine import connect
 from mongo import *
 from flask.testing import FlaskClient
 from .conftest import *
 
+if typing.TYPE_CHECKING:
+    from flask.testing import TestResponse
+
 
 def random_string(k=None):
     '''
@@ -76,9 +80,9 @@ def request(
         method: Literal['get', 'post', 'put', 'patch', 'delete'],
         url: str,
         **ks,
-    ):
+    ) -> Tuple['TestResponse', Union[Any, Dict[str, Any]], Union[Any, None]]:
         func = getattr(client, method)
-        rv = func(url, **ks)
+        rv: 'TestResponse' = func(url, **ks)
         rv_json = rv.get_json()
         if isinstance(rv_json, dict):
             rv_data = rv_json.get('data')

tests/conftest.py

@@ -1,5 +1,6 @@
-from typing import Dict, List
+from typing import Dict, List, Protocol
 from flask import Flask
+from flask.testing import FlaskClient
 from mongo import *
 from mongo import engine
 import mongomock.gridfs
@@ -10,7 +11,6 @@
 from zipfile import ZipFile
 from collections import defaultdict
 from tests.base_tester import random_string
-from tests.test_homework import CourseData
 from tests.test_problem import get_file
 from tests import utils
 
@@ -35,28 +35,34 @@ def client(app: Flask):
     return app.test_client()
 
 
+class ForgeClient(Protocol):
+
+    def __call__(self, username: str) -> FlaskClient:
+        ...
+
+
 @pytest.fixture
-def forge_client(client):
+def forge_client(client: FlaskClient):
 
-    def seted_cookie(username):
+    def seted_cookie(username: str) -> FlaskClient:
         client.set_cookie('test.test', 'piann', User(username).secret)
         return client
 
     return seted_cookie
 
 
 @pytest.fixture
-def client_admin(forge_client):
+def client_admin(forge_client: ForgeClient):
     return forge_client('admin')
 
 
 @pytest.fixture
-def client_teacher(forge_client):
+def client_teacher(forge_client: ForgeClient):
     return forge_client('teacher')
 
 
 @pytest.fixture
-def client_student(forge_client):
+def client_student(forge_client: ForgeClient):
     return forge_client('student')
 
 
@@ -74,6 +80,7 @@ def test2_token():
 
 @pytest.fixture
 def make_course(forge_client):
+    from tests.test_homework import CourseData
 
     def make_course(username, students={}, tas=[]):
         '''

tests/test_auth.py

@@ -6,6 +6,8 @@
 import secrets
 from mongo import *
 from mongo import engine
+from tests import utils
+from tests.conftest import ForgeClient
 
 
 class TestSignup:
@@ -310,13 +312,27 @@ def row(self):
             values = [('' if v is None else v) for v in values]
             return ','.join(map(str, values))
 
+    @staticmethod
+    def cmp_payload_and_user(
+        user: User,
+        payload: SignupInput,
+    ):
+        assert user.username == payload.username
+        assert user.email == payload.email
+        if payload.displayed_name is not None:
+            assert user.profile.displayed_name == payload.displayed_name
+        if payload.role is not None:
+            assert user.role == payload.role
+        login = User.login(payload.username, payload.password)
+        assert login.username == payload.username
+
     @classmethod
     def signup_input(
         cls,
         *,
         displayed_name: Optional[Union[str, bool]] = None,
         role: Optional[int] = None,
-    ):
+    ) -> SignupInput:
         '''
         Generate random signup input data
         '''
@@ -456,3 +472,50 @@ def test_signup_with_invalid_input_format(self, forge_client):
         )
         assert rv.status_code == 400, rv.get_json()
         assert 'input' in rv.get_json()['message']
+
+    def test_force_signup_should_override_existent_users(
+        self,
+        forge_client: ForgeClient,
+    ):
+        existent_users = [
+            self.signup_input(
+                displayed_name=True,
+                role=int(engine.User.Role.TEACHER),
+            ) for _ in range(5)
+        ]
+        for eu in existent_users:
+            u = dataclasses.asdict(eu)
+            del u['displayed_name']
+            del u['role']
+            User.signup(**u)
+            eu.password += secrets.token_hex(8)
+
+        # ensure they can't login with updated payload
+        for u in existent_users:
+            with pytest.raises(engine.DoesNotExist):
+                User.login(u.username, u.password)
+            with pytest.raises(engine.DoesNotExist):
+                User.login(u.email, u.password)
+
+        excepted_users = [
+            *(self.signup_input() for _ in range(5)),
+            *existent_users,
+        ]
+
+        course = utils.course.create_course(teacher='first_admin')
+        client = forge_client('first_admin')
+        rv = client.post(
+            '/auth/batch-signup',
+            json={
+                'newUsers': self.convert_to_csv(excepted_users),
+                'course': course.course_name,
+                'force': True,
+            },
+        )
+        assert rv.status_code == 200, rv.get_json()
+
+        course.reload()
+        for u in excepted_users:
+            login = User.login(u.username, u.password)
+            self.cmp_payload_and_user(login, u)
+            assert u.username in course.student_nicknames

tests/test_homework.py

@@ -1,8 +1,8 @@
-from typing import Callable
-from flask.testing import FlaskClient
+from datetime import datetime
 import pytest
+from flask.testing import FlaskClient
 from tests.base_tester import BaseTester, random_string
-from datetime import datetime
+from tests.conftest import ForgeClient
 from mongo import *
 
 
@@ -96,7 +96,7 @@ class TestIPFilter(BaseTester):
     )
     def test_valid_filter(
         self,
-        forge_client: Callable[[str], FlaskClient],
+        forge_client: ForgeClient,
         course_data: CourseData,
         _filter: str,
     ):
@@ -127,7 +127,7 @@ def test_valid_filter(
     )
     def test_invalid_filter(
         self,
-        forge_client: Callable[[str], FlaskClient],
+        forge_client: ForgeClient,
         course_data: CourseData,
         _filter: str,
     ):