irrelevant, for testing

Miguel Barão
1 parent 7c49db42
Showing 4 changed files with 145 additions and 114 deletions Show diff stats
mypy.ini
perguntations/app.py
perguntations/initdb.py
update.sh
 [mypy]
 python_version = 3.8
-# warn_return_any = True
-# warn_unused_configs = True
 [mypy-setuptools.*]
 ignore_missing_imports = True
@@ -6,7 +6,6 @@ Description:    Main application logic.
 # python standard libraries
 import asyncio
-# from contextlib import contextmanager  # `with` statement in db sessions
 import csv
 import io
 import json
@@ -17,8 +16,7 @@ from os import path
 import bcrypt
 from sqlalchemy import create_engine, select, func
 from sqlalchemy.orm import Session
-from sqlalchemy.exc import NoResultFound
-# from sqlalchemy.orm import sessionmaker
+# from sqlalchemy.exc import NoResultFound
 # this project
 from perguntations.models import Student, Test, Question
@@ -113,8 +111,6 @@ class App():
             logger.info('Generating %d tests. May take awhile...',
                         len(self.allowed))
             self._pregenerate_tests(len(self.allowed))
-        else:
-            logger.info('No tests generated yet.')
         if conf['correct']:
             self._correct_tests()
@@ -139,94 +135,91 @@ class App():
         logger.info('Database "%s" has %s students.', dbfile, num)
-    # ------------------------------------------------------------------------
-    def _correct_tests(self):
-        with Session(self._engine, future=True) as session:
-            # Find which tests have to be corrected
-            dbtests = session.execute(
-                select(Test).
-                where(Test.ref == self.testfactory['ref']).
-                where(Test.state == "SUBMITTED")
-            ).all()
-            # dbtests = session.query(Test)\
-            #         .filter(Test.ref == self.testfactory['ref'])\
-            #         .filter(Test.state == "SUBMITTED")\
-            #         .all()
-
-            logger.info('Correcting %d tests...', len(dbtests))
-            for dbtest in dbtests:
-                try:
-                    with open(dbtest.filename) as file:
-                        testdict = json.load(file)
-                except FileNotFoundError:
-                    logger.error('File not found: %s', dbtest.filename)
-                    continue
-
-                # creates a class Test with the methods to correct it
-                # the questions are still dictionaries, so we have to call
-                # question_from() to produce Question() instances that can be
-                # corrected. Finally the test can be corrected.
-                test = perguntations.test.Test(testdict)
-                test['questions'] = [question_from(q) for q in test['questions']]
-                test.correct()
-                logger.info('Student %s:  grade = %f', test['student']['number'], test['grade'])
-
-                # save JSON file (overwriting the old one)
-                uid = test['student']['number']
-                ref = test['ref']
-                finish_time = test['finish_time']
-                answers_dir = test['answers_dir']
-                fname = f'{uid}--{ref}--{finish_time}.json'
-                fpath = path.join(answers_dir, fname)
-                test.save_json(fpath)
-                logger.info('%s saved JSON file.', uid)
-
-                # update database
-                dbtest.grade = test['grade']
-                dbtest.state = test['state']
-                dbtest.questions = [
-                    Question(
-                        number=n,
-                        ref=q['ref'],
-                        grade=q['grade'],
-                        comment=q.get('comment', ''),
-                        starttime=str(test['start_time']),
-                        finishtime=str(test['finish_time']),
-                        test_id=test['ref']
-                        )
-                    for n, q in enumerate(test['questions'])
-                    ]
-                logger.info('%s database updated.', uid)
+#     # ------------------------------------------------------------------------
+#     # FIXME not working
+#     def _correct_tests(self):
+#         with Session(self._engine, future=True) as session:
+#             # Find which tests have to be corrected
+#             dbtests = session.execute(
+#                 select(Test).
+#                 where(Test.ref == self.testfactory['ref']).
+#                 where(Test.state == "SUBMITTED")
+#             ).all()
+#             # dbtests = session.query(Test)\
+#             #         .filter(Test.ref == self.testfactory['ref'])\
+#             #         .filter(Test.state == "SUBMITTED")\
+#             #         .all()
+
+#             logger.info('Correcting %d tests...', len(dbtests))
+#             for dbtest in dbtests:
+#                 try:
+#                     with open(dbtest.filename) as file:
+#                         testdict = json.load(file)
+#                 except FileNotFoundError:
+#                     logger.error('File not found: %s', dbtest.filename)
+#                     continue
+
+#                 # creates a class Test with the methods to correct it
+#                 # the questions are still dictionaries, so we have to call
+#                 # question_from() to produce Question() instances that can be
+#                 # corrected. Finally the test can be corrected.
+#                 test = perguntations.test.Test(testdict)
+#                 test['questions'] = [question_from(q) for q in test['questions']]
+#                 test.correct()
+#                 logger.info('Student %s:  grade = %f', test['student']['number'], test['grade'])
+
+#                 # save JSON file (overwriting the old one)
+#                 uid = test['student']['number']
+#                 ref = test['ref']
+#                 finish_time = test['finish_time']
+#                 answers_dir = test['answers_dir']
+#                 fname = f'{uid}--{ref}--{finish_time}.json'
+#                 fpath = path.join(answers_dir, fname)
+#                 test.save_json(fpath)
+#                 logger.info('%s saved JSON file.', uid)
+
+#                 # update database
+#                 dbtest.grade = test['grade']
+#                 dbtest.state = test['state']
+#                 dbtest.questions = [
+#                     Question(
+#                         number=n,
+#                         ref=q['ref'],
+#                         grade=q['grade'],
+#                         comment=q.get('comment', ''),
+#                         starttime=str(test['start_time']),
+#                         finishtime=str(test['finish_time']),
+#                         test_id=test['ref']
+#                         )
+#                     for n, q in enumerate(test['questions'])
+#                     ]
+#                 logger.info('%s database updated.', uid)
     # ------------------------------------------------------------------------
-    async def login(self, uid, try_pw, headers=None):
+    async def login(self, uid, password, headers=None):
         '''login authentication'''
-        if uid not in self.allowed and uid != '0':      # not allowed
+        if uid != '0' and uid not in self.allowed:      # not allowed
             logger.warning('"%s" unauthorized.', uid)
             return 'unauthorized'
         with Session(self._engine, future=True) as session:
-        # with self._db_session() as sess:
-            name, hashed_pw = session.execute(
+            name, hashed = session.execute(
                     select(Student.name, Student.password).
-                    where(id == uid)
+                    where(Student.id == uid)
             ).one()
-            # name, hashed_pw = sess.query(Student.name, Student.password)\
-            #                      .filter_by(id=uid)\
-            #                      .one()
-        if hashed_pw == '':              # update password on first login
-            await self.update_student_password(uid, try_pw)
-            pw_ok = True
+        if hashed == '':              # update password on first login
+            logger.info('First login "%s"', name)
+            await self.update_password(uid, password)
+            ok = True
         else:                           # check password
             loop = asyncio.get_running_loop()
-            pw_ok = await loop.run_in_executor(None,
-                                                bcrypt.checkpw,
-                                                try_pw.encode('utf-8'),
-                                                hashed_pw.password)
-            # pw_ok = await check_password(try_pw, hashed_pw)  # async bcrypt
+            ok = await loop.run_in_executor(None,
+                                            bcrypt.checkpw,
+                                            password.encode('utf-8'),
+                                            hashed)
-        if not pw_ok:        # wrong password
+        if not ok:
             logger.info('"%s" wrong password.', uid)
             return 'wrong_password'
@@ -238,6 +231,7 @@ class App():
                            uid, headers['remote_ip'])
             # FIXME invalidate previous login
         else:
+            # first login
             self.online[uid] = {'student': {
                                     'name': name,
                                     'number': uid,
@@ -375,8 +369,9 @@ class App():
                 for n, q in enumerate(test['questions'])
                 ]
-        with self._db_session() as sess:
-            sess.add(test_row)
+        with Session(self._engine, future=True) as session:
+            session.add(test_row)
+            session.commit()
         logger.info('"%s" database updated.', uid)
     # ------------------------------------------------------------------------
@@ -432,12 +427,22 @@ class App():
     def get_questions_csv(self):
         '''generates a CSV with the grades of the test'''
         test_ref = self.testfactory['ref']
-        with self._db_session() as sess:
-            questions = sess.query(Test.id, Test.student_id, Test.starttime,
-                                   Question.number, Question.grade)\
-                            .join(Question)\
-                            .filter(Test.ref == test_ref)\
-                            .all()
+        with Session(self._engine, future=True) as session:
+            questions = session.execute(
+                select(Test.id, Test.student_id, Test.starttime,
+                       Question.number, Question.grade).
+                join(Question).
+                where(Test.ref == test_ref)
+            ).all()
+            print(questions)
+
+
+
+            # questions = sess.query(Test.id, Test.student_id, Test.starttime,
+            #                        Question.number, Question.grade)\
+            #                 .join(Question)\
+            #                 .filter(Test.ref == test_ref)\
+            #                 .all()
             qnums = set()  # keeps track of all the questions in the test
             tests = {}      # {test_id: {student_id, starttime, 0: grade, ...}}
@@ -463,14 +468,21 @@ class App():
     def get_test_csv(self):
         '''generates a CSV with the grades of the test currently running'''
         test_ref = self.testfactory['ref']
-        with self._db_session() as sess:
-            tests = sess.query(Test.student_id,
-                                Test.grade,
-                                Test.starttime, Test.finishtime)\
-                         .filter(Test.ref == test_ref)\
-                         .order_by(Test.student_id)\
-                         .all()
-
+        with Session(self._engine, future=True) as session:
+            tests = session.execute(
+                select(Test.student_id, Test.grade, Test.starttime, Test.finishtime).
+                where(Test.ref == test_ref).
+                order_by(Test.student_id)
+            ).all()
+        # with self._db_session() as sess:
+            # tests = sess.query(Test.student_id,
+            #                     Test.grade,
+            #                     Test.starttime, Test.finishtime)\
+            #              .filter(Test.ref == test_ref)\
+            #              .order_by(Test.student_id)\
+            #              .all()
+
+        print(tests)
         if not tests:
             logger.warning('Empty CSV: there are no tests!')
             return test_ref, ''
@@ -603,21 +615,40 @@ class App():
         logger.info('"%s" area=%g%%, window=%dx%d, screen=%dx%d',
                     uid, area, win_x, win_y, scr_x, scr_y)
-    async def update_student_password(self, uid, password=''):
+    async def update_password(self, uid, password=''):
         '''change password on the database'''
         if password:
-            password = await hash_password(password)
-        with self._db_session() as sess:
-            student = sess.query(Student).filter_by(id=uid).one()
+            # password = await hash_password(password)
+            loop = asyncio.get_running_loop()
+            password =  await loop.run_in_executor(None,
+                                                   bcrypt.hashpw,
+                                                   password.encode('utf-8'),
+                                                   bcrypt.gensalt())
+
+        # with self._db_session() as sess:
+        #     student = sess.query(Student).filter_by(id=uid).one()
+        with Session(self._engine, future=True) as session:
+            student = session.execute(
+                select(Student).
+                where(Student.id == uid)
+            ).scalar_one()
             student.password = password
         logger.info('"%s" password updated.', uid)
     def insert_new_student(self, uid, name):
         '''insert new student into the database'''
-        try:
-            with self._db_session() as sess:
-                sess.add(Student(id=uid, name=name, password=''))
-        except exc.SQLAlchemyError:
-            logger.error('Insert failed: student %s already exists?', uid)
-        else:
-            logger.info('New student: "%s", "%s"', uid, name)
+        with Session(self._engine, future=True) as session:
+            session.add(
+                Student(id=uid, name=name, password='')
+            )
+            session.commit()
+        # try:
+        #     with Session(self._engine, future=True) as session:
+        #         session.add(
+        #             Student(id=uid, name=name, password='')
+        #         )
+        #         session.commit()
+        # except Exception:
+        #     logger.error('Insert failed: student %s already exists?', uid)
+        # else:
+        #     logger.info('New student: "%s", "%s"', uid, name)
@@ -194,7 +194,7 @@ def main():
     if args.update_all:
         all_students = session.execute(
             select(Student).where(Student.id != '0')
-        ).all()
+        ).scalars().all()
         print(f'Updating password of {len(all_students)} users', end='')
         for student in all_students:
@@ -208,9 +208,12 @@ def main():
     else:
         for student_id in args.update:
             print(f'Updating password of {student_id}')
-            student = session.execute(select(Student.id))
-            password = (args.pw or student_id).encode('utf-8')
-            student.password = bcrypt.hashpw(password, bcrypt.gensalt())
+            student = session.execute(
+                select(Student).
+                where(Student.id == student_id)
+            ).scalar_one()
+            new_password = (args.pw or student_id).encode('utf-8')
+            student.password = bcrypt.hashpw(new_password, bcrypt.gensalt())
         session.commit()
     show_students_in_database(session, args.verbose)
@@ -3,4 +3,3 @@
 git pull
 npm update
 pip install -U .
-
1	[mypy]	1	[mypy]
2	python_version = 3.8	2	python_version = 3.8
3	-# warn_return_any = True
4	-# warn_unused_configs = True
5		3
6	[mypy-setuptools.*]	4	[mypy-setuptools.*]
7	ignore_missing_imports = True	5	ignore_missing_imports = True
	@@ -3,4 +3,3 @@		@@ -3,4 +3,3 @@
3	git pull	3	git pull
4	npm update	4	npm update
5	pip install -U .	5	pip install -U .
6	-