split test.py into test.py and testfactory.py

Miguel Barão
1 parent adb8522e
Showing 4 changed files with 303 additions and 300 deletions Show diff stats
perguntations/app.py
perguntations/main.py
perguntations/test.py
perguntations/testfactory.py
@@ -20,7 +20,7 @@ from sqlalchemy.orm import sessionmaker
 # this project
 from perguntations.models import Student, Test, Question
 from perguntations.tools import load_yaml
-from perguntations.test import TestFactory, TestFactoryException
+from perguntations.testfactory import TestFactory, TestFactoryException
  
 logger = logging.getLogger(__name__)
  
@@ -182,22 +182,16 @@ class App():
         testconf.update(conf)
  
         # start test factory
-        logger.info('Making test factory...')
+        logger.info('Running test factory...')
         try:
             self.testfactory = TestFactory(testconf)
         except TestFactoryException as exc:
             logger.critical(exc)
             raise AppException('Failed to create test factory!') from exc
  
-        logger.info('Test factory ready. No errors found.')
-
     # ------------------------------------------------------------------------
     def _pregenerate_tests(self, num):
         event_loop = asyncio.get_event_loop()
-        # for _ in range(num):
-        #     test = event_loop.run_until_complete(self.testfactory.generate())
-        #     self.pregenerated_tests.append(test)
-
         self.pregenerated_tests += [
                 event_loop.run_until_complete(self.testfactory.generate())
                 for _ in range(num)]
@@ -99,13 +99,12 @@ def get_logger_config(debug=False):
                 },
             },
         }
-    default_config['loggers'].update({
-        f'{APP_NAME}.{module}': {
-            'handlers': ['default'],
-            'level': level,
-            'propagate': False,
-            } for module in ['app', 'models', 'factory', 'questions',
-                             'test', 'tools']})
+
+    modules = ['app', 'models', 'questions', 'test', 'testfactory', 'tools']
+    logger = {'handlers': ['default'], 'level': level, 'propagate': False}
+
+    default_config['loggers'].update({f'{APP_NAME}.{module}': logger
+                                      for module in modules})
  
     return load_yaml(config_file, default=default_config)
  
 '''
-TestFactory - generates tests for students
 Test - instances of this class are individual tests
 '''
  
-
 # python standard library
-from os import path
-import random
 from datetime import datetime
 import logging
-import re
-from typing import Any, Dict
-
-# this project
-from perguntations.questions import QFactory, QuestionException
-from perguntations.tools import load_yaml
  
 # Logger configuration
 logger = logging.getLogger(__name__)
  
  
 # ============================================================================
-class TestFactoryException(Exception):
-    '''exception raised in this module'''
-
-
-# ============================================================================
-class TestFactory(dict):
-    '''
-    Each instance of TestFactory() is a test generator.
-    For example, if we want to serve two different tests, then we need two
-    instances of TestFactory(), one for each test.
-    '''
-
-    # ------------------------------------------------------------------------
-    def __init__(self, conf: Dict[str, Any]) -> None:
-        '''
-        Loads configuration from yaml file, then overrides some configurations
-        using the conf argument.
-        Base questions are added to a pool of questions factories.
-        '''
-
-        # --- set test defaults and then use given configuration
-        super().__init__({  # defaults
-            'title': '',
-            'show_points': True,
-            'scale': None,  # or [0, 20]
-            'duration': 0,  # 0=infinite
-            'autosubmit': False,
-            'debug': False,
-            'show_ref': False,
-            })
-        self.update(conf)
-
-        # --- perform sanity checks and normalize the test questions
-        self.sanity_checks()
-        logger.info('Sanity checks PASSED.')
-
-        # --- find refs of all questions used in the test
-        qrefs = {r for qq in self['questions'] for r in qq['ref']}
-        logger.info('Declared %d questions (each test uses %d).',
-                    len(qrefs), len(self["questions"]))
-
-        # --- for review, we are done. no factories needed
-        if self['review']:
-            logger.info('Review mode. No questions loaded. No factories.')
-            return
-
-        # --- load and build question factories
-        self.question_factory = {}
-
-        counter = 1
-        for file in self["files"]:
-            fullpath = path.normpath(path.join(self["questions_dir"], file))
-            (dirname, filename) = path.split(fullpath)
-
-            logger.info('Loading "%s"...', fullpath)
-            questions = load_yaml(fullpath)   # , default=[])
-
-            for i, question in enumerate(questions):
-                # make sure every question in the file is a dictionary
-                if not isinstance(question, dict):
-                    msg = f'Question {i} in {file} is not a dictionary'
-                    raise TestFactoryException(msg)
-
-                # check if ref is missing, then set to '/path/file.yaml:3'
-                if 'ref' not in question:
-                    question['ref'] = f'{file}:{i:04}'
-                    logger.warning('Missing ref set to "%s"', question["ref"])
-
-                # check for duplicate refs
-                if question['ref'] in self.question_factory:
-                    other = self.question_factory[question['ref']]
-                    otherfile = path.join(other.question['path'],
-                                          other.question['filename'])
-                    msg = (f'Duplicate reference "{question["ref"]}" in files '
-                           f'"{otherfile}" and "{fullpath}".')
-                    raise TestFactoryException(msg)
-
-                # make factory only for the questions used in the test
-                if question['ref'] in qrefs:
-                    question.setdefault('type', 'information')
-                    question.update({
-                        'filename': filename,
-                        'path': dirname,
-                        'index': i            # position in the file, 0 based
-                        })
-
-                    self.question_factory[question['ref']] = QFactory(question)
-
-                    # check if all the questions can be correctly generated
-                    try:
-                        self.question_factory[question['ref']].generate()
-                    except Exception as exc:
-                        msg = f'Failed to generate "{question["ref"]}"'
-                        raise TestFactoryException(msg) from exc
-                    else:
-                        logger.info('%4d.  "%s" Ok.', counter, question["ref"])
-                    counter += 1
-
-        qmissing = qrefs.difference(set(self.question_factory.keys()))
-        if qmissing:
-            raise TestFactoryException(f'Could not find questions {qmissing}.')
-
-    # ------------------------------------------------------------------------
-    def check_test_ref(self) -> None:
-        '''Test must have a `ref`'''
-        if 'ref' not in self:
-            raise TestFactoryException('Missing "ref" in configuration!')
-        if not re.match(r'^[a-zA-Z0-9_-]+$', self['ref']):
-            raise TestFactoryException('Test "ref" can only contain the '
-                                       'characters a-zA-Z0-9_-')
-
-    def check_missing_database(self) -> None:
-        '''Test must have a database'''
-        if 'database' not in self:
-            raise TestFactoryException('Missing "database" in configuration')
-        if not path.isfile(path.expanduser(self['database'])):
-            msg = f'Database "{self["database"]}" not found!'
-            raise TestFactoryException(msg)
-
-    def check_missing_answers_directory(self) -> None:
-        '''Test must have a answers directory'''
-        if 'answers_dir' not in self:
-            msg = 'Missing "answers_dir" in configuration'
-            raise TestFactoryException(msg)
-
-    def check_answers_directory_writable(self) -> None:
-        '''Answers directory must be writable'''
-        testfile = path.join(path.expanduser(self['answers_dir']), 'REMOVE-ME')
-        try:
-            with open(testfile, 'w') as file:
-                file.write('You can safely remove this file.')
-        except OSError as exc:
-            msg = f'Cannot write answers to directory "{self["answers_dir"]}"'
-            raise TestFactoryException(msg) from exc
-
-    def check_questions_directory(self) -> None:
-        '''Check if questions directory is missing or not accessible.'''
-        if 'questions_dir' not in self:
-            logger.warning('Missing "questions_dir". Using "%s"',
-                           path.abspath(path.curdir))
-            self['questions_dir'] = path.curdir
-        elif not path.isdir(path.expanduser(self['questions_dir'])):
-            raise TestFactoryException(f'Can\'t find questions directory '
-                                       f'"{self["questions_dir"]}"')
-
-    def check_import_files(self) -> None:
-        '''Check if there are files to import (with questions)'''
-        if 'files' not in self:
-            msg = ('Missing "files" in configuration with the list of '
-                   'question files to import!')
-            raise TestFactoryException(msg)
-
-        if isinstance(self['files'], str):
-            self['files'] = [self['files']]
-
-    def check_question_list(self) -> None:
-        '''normalize question list'''
-        if 'questions' not in self:
-            raise TestFactoryException('Missing "questions" in configuration')
-
-        for i, question in enumerate(self['questions']):
-            # normalize question to a dict and ref to a list of references
-            if isinstance(question, str):  # e.g.,    - some_ref
-                question = {'ref': [question]}    # becomes  - ref: [some_ref]
-            elif isinstance(question, dict) and isinstance(question['ref'], str):
-                question['ref'] = [question['ref']]
-            elif isinstance(question, list):
-                question = {'ref': [str(a) for a in question]}
-
-            self['questions'][i] = question
-
-    def check_missing_title(self) -> None:
-        '''Warns if title is missing'''
-        if not self['title']:
-            logger.warning('Title is undefined!')
-
-    def check_grade_scaling(self) -> None:
-        '''Just informs the scale limits'''
-        if 'scale_points' in self:
-            msg = ('*** DEPRECATION WARNING: *** scale_points, scale_min, '
-                   'scale_max were replaced by "scale: [min, max]".')
-            logger.warning(msg)
-            self['scale'] = [self['scale_min'], self['scale_max']]
-
-
-    # ------------------------------------------------------------------------
-    def sanity_checks(self) -> None:
-        '''
-        Checks for valid keys and sets default values.
-        Also checks if some files and directories exist
-        '''
-        self.check_test_ref()
-        self.check_missing_database()
-        self.check_missing_answers_directory()
-        self.check_answers_directory_writable()
-        self.check_questions_directory()
-        self.check_import_files()
-        self.check_question_list()
-        self.check_missing_title()
-        self.check_grade_scaling()
-
-    # ------------------------------------------------------------------------
-    async def generate(self):
-        '''
-        Given a dictionary with a student dict {'name':'john', 'number': 123}
-        returns instance of Test() for that particular student
-        '''
-
-        # make list of questions
-        questions = []
-        qnum = 1  # track question number
-        nerr = 0  # count errors during questions generation
-
-        for qlist in self['questions']:
-            # choose list of question variants
-            choose = qlist.get('choose', 1)
-            qrefs = random.sample(qlist['ref'], k=choose)
-
-            for qref in qrefs:
-                # generate instance of question
-                try:
-                    question = await self.question_factory[qref].gen_async()
-                except QuestionException:
-                    logger.error('Can\'t generate question "%s". Skipping.', qref)
-                    nerr += 1
-                    continue
-
-                # some defaults
-                if question['type'] in ('information', 'success', 'warning',
-                                        'alert'):
-                    question['points'] = qlist.get('points', 0.0)
-                else:
-                    question['points'] = qlist.get('points', 1.0)
-                    question['number'] = qnum  # counter for non informative panels
-                    qnum += 1
-
-                questions.append(question)
-
-        # setup scale
-        total_points = sum(q['points'] for q in questions)
-
-        if total_points > 0:
-            # normalize question points to scale
-            if self['scale'] is not None:
-                scale_min, scale_max = self['scale']
-                for question in questions:
-                    question['points'] *= (scale_max - scale_min) / total_points
-            else:
-                self['scale'] = [0, total_points]
-        else:
-            logger.warning('Total points is **ZERO**.')
-            if self['scale'] is None:
-                self['scale'] = [0, 20]  # default
-
-        if nerr > 0:
-            logger.error('%s errors found!', nerr)
-
-        # copy these from the test configuratoin to each test instance
-        inherit = {'ref', 'title', 'database', 'answers_dir',
-                   'questions_dir', 'files',
-                   'duration', 'autosubmit',
-                   'scale', 'show_points',
-                   'show_ref', 'debug', }
-        # NOT INCLUDED:  testfile, allow_all, review
-
-        return Test({'questions': questions, **{k:self[k] for k in inherit}})
-
-    # ------------------------------------------------------------------------
-    def __repr__(self):
-        testsettings = '\n'.join(f'  {k:14s}: {v}' for k, v in self.items())
-        return '{\n' + testsettings + '\n}'
-
-
-# ============================================================================
 class Test(dict):
     '''
     Each instance Test() is a concrete test of a single student.
@@ -335,7 +51,6 @@ class Test(dict):
         '''
         for ref, ans in answers_dict.items():
             self['questions'][ref].set_answer(ans)
-            # self['questions'][ref]['answer'] = ans
  
     # ------------------------------------------------------------------------
     async def correct(self) -> float:
@@ -0,0 +1,295 @@
+'''
+TestFactory - generates tests for students
+'''
+
+# python standard library
+from os import path
+import random
+import logging
+import re
+from typing import Any, Dict
+
+# this project
+from perguntations.questions import QFactory, QuestionException
+from perguntations.test import Test
+from perguntations.tools import load_yaml
+
+# Logger configuration
+logger = logging.getLogger(__name__)
+
+
+# ============================================================================
+class TestFactoryException(Exception):
+    '''exception raised in this module'''
+
+
+# ============================================================================
+class TestFactory(dict):
+    '''
+    Each instance of TestFactory() is a test generator.
+    For example, if we want to serve two different tests, then we need two
+    instances of TestFactory(), one for each test.
+    '''
+
+    # ------------------------------------------------------------------------
+    def __init__(self, conf: Dict[str, Any]) -> None:
+        '''
+        Loads configuration from yaml file, then overrides some configurations
+        using the conf argument.
+        Base questions are added to a pool of questions factories.
+        '''
+
+        # --- set test defaults and then use given configuration
+        super().__init__({  # defaults
+            'title': '',
+            'show_points': True,
+            'scale': None,  # or [0, 20]
+            'duration': 0,  # 0=infinite
+            'autosubmit': False,
+            'debug': False,
+            'show_ref': False,
+            })
+        self.update(conf)
+
+        # --- for review, we are done. no factories needed
+        if self['review']:
+            logger.info('Review mode. No questions loaded. No factories.')
+            return
+
+        # --- perform sanity checks and normalize the test questions
+        self.sanity_checks()
+        logger.info('Sanity checks PASSED.')
+
+        # --- find refs of all questions used in the test
+        qrefs = {r for qq in self['questions'] for r in qq['ref']}
+        logger.info('Declared %d questions (each test uses %d).',
+                    len(qrefs), len(self["questions"]))
+
+        # --- load and build question factories
+        self.question_factory = {}
+
+        counter = 1
+        for file in self["files"]:
+            fullpath = path.normpath(path.join(self["questions_dir"], file))
+            (dirname, filename) = path.split(fullpath)
+
+            logger.info('Loading "%s"...', fullpath)
+            questions = load_yaml(fullpath)   # , default=[])
+
+            for i, question in enumerate(questions):
+                # make sure every question in the file is a dictionary
+                if not isinstance(question, dict):
+                    msg = f'Question {i} in {file} is not a dictionary'
+                    raise TestFactoryException(msg)
+
+                # check if ref is missing, then set to '/path/file.yaml:3'
+                if 'ref' not in question:
+                    question['ref'] = f'{file}:{i:04}'
+                    logger.warning('Missing ref set to "%s"', question["ref"])
+
+                # check for duplicate refs
+                if question['ref'] in self.question_factory:
+                    other = self.question_factory[question['ref']]
+                    otherfile = path.join(other.question['path'],
+                                          other.question['filename'])
+                    msg = (f'Duplicate reference "{question["ref"]}" in files '
+                           f'"{otherfile}" and "{fullpath}".')
+                    raise TestFactoryException(msg)
+
+                # make factory only for the questions used in the test
+                if question['ref'] in qrefs:
+                    question.setdefault('type', 'information')
+                    question.update({
+                        'filename': filename,
+                        'path': dirname,
+                        'index': i            # position in the file, 0 based
+                        })
+
+                    self.question_factory[question['ref']] = QFactory(question)
+
+                    # check if all the questions can be correctly generated
+                    try:
+                        self.question_factory[question['ref']].generate()
+                    except Exception as exc:
+                        msg = f'Failed to generate "{question["ref"]}"'
+                        raise TestFactoryException(msg) from exc
+                    else:
+                        logger.info('%4d.  "%s" Ok.', counter, question["ref"])
+                    counter += 1
+
+        qmissing = qrefs.difference(set(self.question_factory.keys()))
+        if qmissing:
+            raise TestFactoryException(f'Could not find questions {qmissing}.')
+
+        logger.info('Test factory ready. No errors found.')
+
+
+    # ------------------------------------------------------------------------
+    def check_test_ref(self) -> None:
+        '''Test must have a `ref`'''
+        if 'ref' not in self:
+            raise TestFactoryException('Missing "ref" in configuration!')
+        if not re.match(r'^[a-zA-Z0-9_-]+$', self['ref']):
+            raise TestFactoryException('Test "ref" can only contain the '
+                                       'characters a-zA-Z0-9_-')
+
+    def check_missing_database(self) -> None:
+        '''Test must have a database'''
+        if 'database' not in self:
+            raise TestFactoryException('Missing "database" in configuration')
+        if not path.isfile(path.expanduser(self['database'])):
+            msg = f'Database "{self["database"]}" not found!'
+            raise TestFactoryException(msg)
+
+    def check_missing_answers_directory(self) -> None:
+        '''Test must have a answers directory'''
+        if 'answers_dir' not in self:
+            msg = 'Missing "answers_dir" in configuration'
+            raise TestFactoryException(msg)
+
+    def check_answers_directory_writable(self) -> None:
+        '''Answers directory must be writable'''
+        testfile = path.join(path.expanduser(self['answers_dir']), 'REMOVE-ME')
+        try:
+            with open(testfile, 'w') as file:
+                file.write('You can safely remove this file.')
+        except OSError as exc:
+            msg = f'Cannot write answers to directory "{self["answers_dir"]}"'
+            raise TestFactoryException(msg) from exc
+
+    def check_questions_directory(self) -> None:
+        '''Check if questions directory is missing or not accessible.'''
+        if 'questions_dir' not in self:
+            logger.warning('Missing "questions_dir". Using "%s"',
+                           path.abspath(path.curdir))
+            self['questions_dir'] = path.curdir
+        elif not path.isdir(path.expanduser(self['questions_dir'])):
+            raise TestFactoryException(f'Can\'t find questions directory '
+                                       f'"{self["questions_dir"]}"')
+
+    def check_import_files(self) -> None:
+        '''Check if there are files to import (with questions)'''
+        if 'files' not in self:
+            msg = ('Missing "files" in configuration with the list of '
+                   'question files to import!')
+            raise TestFactoryException(msg)
+
+        if isinstance(self['files'], str):
+            self['files'] = [self['files']]
+
+    def check_question_list(self) -> None:
+        '''normalize question list'''
+        if 'questions' not in self:
+            raise TestFactoryException('Missing "questions" in configuration')
+
+        for i, question in enumerate(self['questions']):
+            # normalize question to a dict and ref to a list of references
+            if isinstance(question, str):  # e.g.,    - some_ref
+                question = {'ref': [question]}    # becomes  - ref: [some_ref]
+            elif isinstance(question, dict) and isinstance(question['ref'], str):
+                question['ref'] = [question['ref']]
+            elif isinstance(question, list):
+                question = {'ref': [str(a) for a in question]}
+
+            self['questions'][i] = question
+
+    def check_missing_title(self) -> None:
+        '''Warns if title is missing'''
+        if not self['title']:
+            logger.warning('Title is undefined!')
+
+    def check_grade_scaling(self) -> None:
+        '''Just informs the scale limits'''
+        if 'scale_points' in self:
+            msg = ('*** DEPRECATION WARNING: *** scale_points, scale_min, '
+                   'scale_max were replaced by "scale: [min, max]".')
+            logger.warning(msg)
+            self['scale'] = [self['scale_min'], self['scale_max']]
+
+
+    # ------------------------------------------------------------------------
+    def sanity_checks(self) -> None:
+        '''
+        Checks for valid keys and sets default values.
+        Also checks if some files and directories exist
+        '''
+        self.check_test_ref()
+        self.check_missing_database()
+        self.check_missing_answers_directory()
+        self.check_answers_directory_writable()
+        self.check_questions_directory()
+        self.check_import_files()
+        self.check_question_list()
+        self.check_missing_title()
+        self.check_grade_scaling()
+
+    # ------------------------------------------------------------------------
+    async def generate(self):
+        '''
+        Given a dictionary with a student dict {'name':'john', 'number': 123}
+        returns instance of Test() for that particular student
+        '''
+
+        # make list of questions
+        questions = []
+        qnum = 1  # track question number
+        nerr = 0  # count errors during questions generation
+
+        for qlist in self['questions']:
+            # choose list of question variants
+            choose = qlist.get('choose', 1)
+            qrefs = random.sample(qlist['ref'], k=choose)
+
+            for qref in qrefs:
+                # generate instance of question
+                try:
+                    question = await self.question_factory[qref].gen_async()
+                except QuestionException:
+                    logger.error('Can\'t generate question "%s". Skipping.', qref)
+                    nerr += 1
+                    continue
+
+                # some defaults
+                if question['type'] in ('information', 'success', 'warning',
+                                        'alert'):
+                    question['points'] = qlist.get('points', 0.0)
+                else:
+                    question['points'] = qlist.get('points', 1.0)
+                    question['number'] = qnum  # counter for non informative panels
+                    qnum += 1
+
+                questions.append(question)
+
+        # setup scale
+        total_points = sum(q['points'] for q in questions)
+
+        if total_points > 0:
+            # normalize question points to scale
+            if self['scale'] is not None:
+                scale_min, scale_max = self['scale']
+                for question in questions:
+                    question['points'] *= (scale_max - scale_min) / total_points
+            else:
+                self['scale'] = [0, total_points]
+        else:
+            logger.warning('Total points is **ZERO**.')
+            if self['scale'] is None:
+                self['scale'] = [0, 20]  # default
+
+        if nerr > 0:
+            logger.error('%s errors found!', nerr)
+
+        # copy these from the test configuratoin to each test instance
+        inherit = {'ref', 'title', 'database', 'answers_dir',
+                   'questions_dir', 'files',
+                   'duration', 'autosubmit',
+                   'scale', 'show_points',
+                   'show_ref', 'debug', }
+        # NOT INCLUDED:  testfile, allow_all, review
+
+        return Test({'questions': questions, **{k:self[k] for k in inherit}})
+
+    # ------------------------------------------------------------------------
+    def __repr__(self):
+        testsettings = '\n'.join(f'  {k:14s}: {v}' for k, v in self.items())
+        return '{\n' + testsettings + '\n}'
...	...	@@ -20,7 +20,7 @@ from sqlalchemy.orm import sessionmaker
20	20	# this project
21	21	from perguntations.models import Student, Test, Question
22	22	from perguntations.tools import load_yaml
23		-from perguntations.test import TestFactory, TestFactoryException
	23	+from perguntations.testfactory import TestFactory, TestFactoryException
24	24
25	25	logger = logging.getLogger(__name__)
26	26
...	...	@@ -182,22 +182,16 @@ class App():
182	182	testconf.update(conf)
183	183
184	184	# start test factory
185		- logger.info('Making test factory...')
	185	+ logger.info('Running test factory...')
186	186	try:
187	187	self.testfactory = TestFactory(testconf)
188	188	except TestFactoryException as exc:
189	189	logger.critical(exc)
190	190	raise AppException('Failed to create test factory!') from exc
191	191
192		- logger.info('Test factory ready. No errors found.')
193		-
194	192	# ------------------------------------------------------------------------
195	193	def _pregenerate_tests(self, num):
196	194	event_loop = asyncio.get_event_loop()
197		- # for _ in range(num):
198		- # test = event_loop.run_until_complete(self.testfactory.generate())
199		- # self.pregenerated_tests.append(test)
200		-
201	195	self.pregenerated_tests += [
202	196	event_loop.run_until_complete(self.testfactory.generate())
203	197	for _ in range(num)]
...	...
...	...	@@ -99,13 +99,12 @@ def get_logger_config(debug=False):
99	99	},
100	100	},
101	101	}
102		- default_config['loggers'].update({
103		- f'{APP_NAME}.{module}': {
104		- 'handlers': ['default'],
105		- 'level': level,
106		- 'propagate': False,
107		- } for module in ['app', 'models', 'factory', 'questions',
108		- 'test', 'tools']})
	102	+
	103	+ modules = ['app', 'models', 'questions', 'test', 'testfactory', 'tools']
	104	+ logger = {'handlers': ['default'], 'level': level, 'propagate': False}
	105	+
	106	+ default_config['loggers'].update({f'{APP_NAME}.{module}': logger
	107	+ for module in modules})
109	108
110	109	return load_yaml(config_file, default=default_config)
111	110
...	...
1	1	'''
2		-TestFactory - generates tests for students
3	2	Test - instances of this class are individual tests
4	3	'''
5	4
6		-
7	5	# python standard library
8		-from os import path
9		-import random
10	6	from datetime import datetime
11	7	import logging
12		-import re
13		-from typing import Any, Dict
14		-
15		-# this project
16		-from perguntations.questions import QFactory, QuestionException
17		-from perguntations.tools import load_yaml
18	8
19	9	# Logger configuration
20	10	logger = logging.getLogger(__name__)
21	11
22	12
23	13	# ============================================================================
24		-class TestFactoryException(Exception):
25		- '''exception raised in this module'''
26		-
27		-
28		-# ============================================================================
29		-class TestFactory(dict):
30		- '''
31		- Each instance of TestFactory() is a test generator.
32		- For example, if we want to serve two different tests, then we need two
33		- instances of TestFactory(), one for each test.
34		- '''
35		-
36		- # ------------------------------------------------------------------------
37		- def __init__(self, conf: Dict[str, Any]) -> None:
38		- '''
39		- Loads configuration from yaml file, then overrides some configurations
40		- using the conf argument.
41		- Base questions are added to a pool of questions factories.
42		- '''
43		-
44		- # --- set test defaults and then use given configuration
45		- super().__init__({ # defaults
46		- 'title': '',
47		- 'show_points': True,
48		- 'scale': None, # or [0, 20]
49		- 'duration': 0, # 0=infinite
50		- 'autosubmit': False,
51		- 'debug': False,
52		- 'show_ref': False,
53		- })
54		- self.update(conf)
55		-
56		- # --- perform sanity checks and normalize the test questions
57		- self.sanity_checks()
58		- logger.info('Sanity checks PASSED.')
59		-
60		- # --- find refs of all questions used in the test
61		- qrefs = {r for qq in self['questions'] for r in qq['ref']}
62		- logger.info('Declared %d questions (each test uses %d).',
63		- len(qrefs), len(self["questions"]))
64		-
65		- # --- for review, we are done. no factories needed
66		- if self['review']:
67		- logger.info('Review mode. No questions loaded. No factories.')
68		- return
69		-
70		- # --- load and build question factories
71		- self.question_factory = {}
72		-
73		- counter = 1
74		- for file in self["files"]:
75		- fullpath = path.normpath(path.join(self["questions_dir"], file))
76		- (dirname, filename) = path.split(fullpath)
77		-
78		- logger.info('Loading "%s"...', fullpath)
79		- questions = load_yaml(fullpath) # , default=[])
80		-
81		- for i, question in enumerate(questions):
82		- # make sure every question in the file is a dictionary
83		- if not isinstance(question, dict):
84		- msg = f'Question {i} in {file} is not a dictionary'
85		- raise TestFactoryException(msg)
86		-
87		- # check if ref is missing, then set to '/path/file.yaml:3'
88		- if 'ref' not in question:
89		- question['ref'] = f'{file}:{i:04}'
90		- logger.warning('Missing ref set to "%s"', question["ref"])
91		-
92		- # check for duplicate refs
93		- if question['ref'] in self.question_factory:
94		- other = self.question_factory[question['ref']]
95		- otherfile = path.join(other.question['path'],
96		- other.question['filename'])
97		- msg = (f'Duplicate reference "{question["ref"]}" in files '
98		- f'"{otherfile}" and "{fullpath}".')
99		- raise TestFactoryException(msg)
100		-
101		- # make factory only for the questions used in the test
102		- if question['ref'] in qrefs:
103		- question.setdefault('type', 'information')
104		- question.update({
105		- 'filename': filename,
106		- 'path': dirname,
107		- 'index': i # position in the file, 0 based
108		- })
109		-
110		- self.question_factory[question['ref']] = QFactory(question)
111		-
112		- # check if all the questions can be correctly generated
113		- try:
114		- self.question_factory[question['ref']].generate()
115		- except Exception as exc:
116		- msg = f'Failed to generate "{question["ref"]}"'
117		- raise TestFactoryException(msg) from exc
118		- else:
119		- logger.info('%4d. "%s" Ok.', counter, question["ref"])
120		- counter += 1
121		-
122		- qmissing = qrefs.difference(set(self.question_factory.keys()))
123		- if qmissing:
124		- raise TestFactoryException(f'Could not find questions {qmissing}.')
125		-
126		- # ------------------------------------------------------------------------
127		- def check_test_ref(self) -> None:
128		- '''Test must have a `ref`'''
129		- if 'ref' not in self:
130		- raise TestFactoryException('Missing "ref" in configuration!')
131		- if not re.match(r'^[a-zA-Z0-9_-]+$', self['ref']):
132		- raise TestFactoryException('Test "ref" can only contain the '
133		- 'characters a-zA-Z0-9_-')
134		-
135		- def check_missing_database(self) -> None:
136		- '''Test must have a database'''
137		- if 'database' not in self:
138		- raise TestFactoryException('Missing "database" in configuration')
139		- if not path.isfile(path.expanduser(self['database'])):
140		- msg = f'Database "{self["database"]}" not found!'
141		- raise TestFactoryException(msg)
142		-
143		- def check_missing_answers_directory(self) -> None:
144		- '''Test must have a answers directory'''
145		- if 'answers_dir' not in self:
146		- msg = 'Missing "answers_dir" in configuration'
147		- raise TestFactoryException(msg)
148		-
149		- def check_answers_directory_writable(self) -> None:
150		- '''Answers directory must be writable'''
151		- testfile = path.join(path.expanduser(self['answers_dir']), 'REMOVE-ME')
152		- try:
153		- with open(testfile, 'w') as file:
154		- file.write('You can safely remove this file.')
155		- except OSError as exc:
156		- msg = f'Cannot write answers to directory "{self["answers_dir"]}"'
157		- raise TestFactoryException(msg) from exc
158		-
159		- def check_questions_directory(self) -> None:
160		- '''Check if questions directory is missing or not accessible.'''
161		- if 'questions_dir' not in self:
162		- logger.warning('Missing "questions_dir". Using "%s"',
163		- path.abspath(path.curdir))
164		- self['questions_dir'] = path.curdir
165		- elif not path.isdir(path.expanduser(self['questions_dir'])):
166		- raise TestFactoryException(f'Can\'t find questions directory '
167		- f'"{self["questions_dir"]}"')
168		-
169		- def check_import_files(self) -> None:
170		- '''Check if there are files to import (with questions)'''
171		- if 'files' not in self:
172		- msg = ('Missing "files" in configuration with the list of '
173		- 'question files to import!')
174		- raise TestFactoryException(msg)
175		-
176		- if isinstance(self['files'], str):
177		- self['files'] = [self['files']]
178		-
179		- def check_question_list(self) -> None:
180		- '''normalize question list'''
181		- if 'questions' not in self:
182		- raise TestFactoryException('Missing "questions" in configuration')
183		-
184		- for i, question in enumerate(self['questions']):
185		- # normalize question to a dict and ref to a list of references
186		- if isinstance(question, str): # e.g., - some_ref
187		- question = {'ref': [question]} # becomes - ref: [some_ref]
188		- elif isinstance(question, dict) and isinstance(question['ref'], str):
189		- question['ref'] = [question['ref']]
190		- elif isinstance(question, list):
191		- question = {'ref': [str(a) for a in question]}
192		-
193		- self['questions'][i] = question
194		-
195		- def check_missing_title(self) -> None:
196		- '''Warns if title is missing'''
197		- if not self['title']:
198		- logger.warning('Title is undefined!')
199		-
200		- def check_grade_scaling(self) -> None:
201		- '''Just informs the scale limits'''
202		- if 'scale_points' in self:
203		- msg = ('* DEPRECATION WARNING: * scale_points, scale_min, '
204		- 'scale_max were replaced by "scale: [min, max]".')
205		- logger.warning(msg)
206		- self['scale'] = [self['scale_min'], self['scale_max']]
207		-
208		-
209		- # ------------------------------------------------------------------------
210		- def sanity_checks(self) -> None:
211		- '''
212		- Checks for valid keys and sets default values.
213		- Also checks if some files and directories exist
214		- '''
215		- self.check_test_ref()
216		- self.check_missing_database()
217		- self.check_missing_answers_directory()
218		- self.check_answers_directory_writable()
219		- self.check_questions_directory()
220		- self.check_import_files()
221		- self.check_question_list()
222		- self.check_missing_title()
223		- self.check_grade_scaling()
224		-
225		- # ------------------------------------------------------------------------
226		- async def generate(self):
227		- '''
228		- Given a dictionary with a student dict {'name':'john', 'number': 123}
229		- returns instance of Test() for that particular student
230		- '''
231		-
232		- # make list of questions
233		- questions = []
234		- qnum = 1 # track question number
235		- nerr = 0 # count errors during questions generation
236		-
237		- for qlist in self['questions']:
238		- # choose list of question variants
239		- choose = qlist.get('choose', 1)
240		- qrefs = random.sample(qlist['ref'], k=choose)
241		-
242		- for qref in qrefs:
243		- # generate instance of question
244		- try:
245		- question = await self.question_factory[qref].gen_async()
246		- except QuestionException:
247		- logger.error('Can\'t generate question "%s". Skipping.', qref)
248		- nerr += 1
249		- continue
250		-
251		- # some defaults
252		- if question['type'] in ('information', 'success', 'warning',
253		- 'alert'):
254		- question['points'] = qlist.get('points', 0.0)
255		- else:
256		- question['points'] = qlist.get('points', 1.0)
257		- question['number'] = qnum # counter for non informative panels
258		- qnum += 1
259		-
260		- questions.append(question)
261		-
262		- # setup scale
263		- total_points = sum(q['points'] for q in questions)
264		-
265		- if total_points > 0:
266		- # normalize question points to scale
267		- if self['scale'] is not None:
268		- scale_min, scale_max = self['scale']
269		- for question in questions:
270		- question['points'] *= (scale_max - scale_min) / total_points
271		- else:
272		- self['scale'] = [0, total_points]
273		- else:
274		- logger.warning('Total points is ZERO.')
275		- if self['scale'] is None:
276		- self['scale'] = [0, 20] # default
277		-
278		- if nerr > 0:
279		- logger.error('%s errors found!', nerr)
280		-
281		- # copy these from the test configuratoin to each test instance
282		- inherit = {'ref', 'title', 'database', 'answers_dir',
283		- 'questions_dir', 'files',
284		- 'duration', 'autosubmit',
285		- 'scale', 'show_points',
286		- 'show_ref', 'debug', }
287		- # NOT INCLUDED: testfile, allow_all, review
288		-
289		- return Test({'questions': questions, **{k:self[k] for k in inherit}})
290		-
291		- # ------------------------------------------------------------------------
292		- def __repr__(self):
293		- testsettings = '\n'.join(f' {k:14s}: {v}' for k, v in self.items())
294		- return '{\n' + testsettings + '\n}'
295		-
296		-
297		-# ============================================================================
298	14	class Test(dict):
299	15	'''
300	16	Each instance Test() is a concrete test of a single student.
...	...	@@ -335,7 +51,6 @@ class Test(dict):
335	51	'''
336	52	for ref, ans in answers_dict.items():
337	53	self['questions'][ref].set_answer(ans)
338		- # self['questions'][ref]['answer'] = ans
339	54
340	55	# ------------------------------------------------------------------------
341	56	async def correct(self) -> float:
...	...
...	...	@@ -0,0 +1,295 @@
	1	+'''
	2	+TestFactory - generates tests for students
	3	+'''
	4	+
	5	+# python standard library
	6	+from os import path
	7	+import random
	8	+import logging
	9	+import re
	10	+from typing import Any, Dict
	11	+
	12	+# this project
	13	+from perguntations.questions import QFactory, QuestionException
	14	+from perguntations.test import Test
	15	+from perguntations.tools import load_yaml
	16	+
	17	+# Logger configuration
	18	+logger = logging.getLogger(__name__)
	19	+
	20	+
	21	+# ============================================================================
	22	+class TestFactoryException(Exception):
	23	+ '''exception raised in this module'''
	24	+
	25	+
	26	+# ============================================================================
	27	+class TestFactory(dict):
	28	+ '''
	29	+ Each instance of TestFactory() is a test generator.
	30	+ For example, if we want to serve two different tests, then we need two
	31	+ instances of TestFactory(), one for each test.
	32	+ '''
	33	+
	34	+ # ------------------------------------------------------------------------
	35	+ def __init__(self, conf: Dict[str, Any]) -> None:
	36	+ '''
	37	+ Loads configuration from yaml file, then overrides some configurations
	38	+ using the conf argument.
	39	+ Base questions are added to a pool of questions factories.
	40	+ '''
	41	+
	42	+ # --- set test defaults and then use given configuration
	43	+ super().__init__({ # defaults
	44	+ 'title': '',
	45	+ 'show_points': True,
	46	+ 'scale': None, # or [0, 20]
	47	+ 'duration': 0, # 0=infinite
	48	+ 'autosubmit': False,
	49	+ 'debug': False,
	50	+ 'show_ref': False,
	51	+ })
	52	+ self.update(conf)
	53	+
	54	+ # --- for review, we are done. no factories needed
	55	+ if self['review']:
	56	+ logger.info('Review mode. No questions loaded. No factories.')
	57	+ return
	58	+
	59	+ # --- perform sanity checks and normalize the test questions
	60	+ self.sanity_checks()
	61	+ logger.info('Sanity checks PASSED.')
	62	+
	63	+ # --- find refs of all questions used in the test
	64	+ qrefs = {r for qq in self['questions'] for r in qq['ref']}
	65	+ logger.info('Declared %d questions (each test uses %d).',
	66	+ len(qrefs), len(self["questions"]))
	67	+
	68	+ # --- load and build question factories
	69	+ self.question_factory = {}
	70	+
	71	+ counter = 1
	72	+ for file in self["files"]:
	73	+ fullpath = path.normpath(path.join(self["questions_dir"], file))
	74	+ (dirname, filename) = path.split(fullpath)
	75	+
	76	+ logger.info('Loading "%s"...', fullpath)
	77	+ questions = load_yaml(fullpath) # , default=[])
	78	+
	79	+ for i, question in enumerate(questions):
	80	+ # make sure every question in the file is a dictionary
	81	+ if not isinstance(question, dict):
	82	+ msg = f'Question {i} in {file} is not a dictionary'
	83	+ raise TestFactoryException(msg)
	84	+
	85	+ # check if ref is missing, then set to '/path/file.yaml:3'
	86	+ if 'ref' not in question:
	87	+ question['ref'] = f'{file}:{i:04}'
	88	+ logger.warning('Missing ref set to "%s"', question["ref"])
	89	+
	90	+ # check for duplicate refs
	91	+ if question['ref'] in self.question_factory:
	92	+ other = self.question_factory[question['ref']]
	93	+ otherfile = path.join(other.question['path'],
	94	+ other.question['filename'])
	95	+ msg = (f'Duplicate reference "{question["ref"]}" in files '
	96	+ f'"{otherfile}" and "{fullpath}".')
	97	+ raise TestFactoryException(msg)
	98	+
	99	+ # make factory only for the questions used in the test
	100	+ if question['ref'] in qrefs:
	101	+ question.setdefault('type', 'information')
	102	+ question.update({
	103	+ 'filename': filename,
	104	+ 'path': dirname,
	105	+ 'index': i # position in the file, 0 based
	106	+ })
	107	+
	108	+ self.question_factory[question['ref']] = QFactory(question)
	109	+
	110	+ # check if all the questions can be correctly generated
	111	+ try:
	112	+ self.question_factory[question['ref']].generate()
	113	+ except Exception as exc:
	114	+ msg = f'Failed to generate "{question["ref"]}"'
	115	+ raise TestFactoryException(msg) from exc
	116	+ else:
	117	+ logger.info('%4d. "%s" Ok.', counter, question["ref"])
	118	+ counter += 1
	119	+
	120	+ qmissing = qrefs.difference(set(self.question_factory.keys()))
	121	+ if qmissing:
	122	+ raise TestFactoryException(f'Could not find questions {qmissing}.')
	123	+
	124	+ logger.info('Test factory ready. No errors found.')
	125	+
	126	+
	127	+ # ------------------------------------------------------------------------
	128	+ def check_test_ref(self) -> None:
	129	+ '''Test must have a `ref`'''
	130	+ if 'ref' not in self:
	131	+ raise TestFactoryException('Missing "ref" in configuration!')
	132	+ if not re.match(r'^[a-zA-Z0-9_-]+$', self['ref']):
	133	+ raise TestFactoryException('Test "ref" can only contain the '
	134	+ 'characters a-zA-Z0-9_-')
	135	+
	136	+ def check_missing_database(self) -> None:
	137	+ '''Test must have a database'''
	138	+ if 'database' not in self:
	139	+ raise TestFactoryException('Missing "database" in configuration')
	140	+ if not path.isfile(path.expanduser(self['database'])):
	141	+ msg = f'Database "{self["database"]}" not found!'
	142	+ raise TestFactoryException(msg)
	143	+
	144	+ def check_missing_answers_directory(self) -> None:
	145	+ '''Test must have a answers directory'''
	146	+ if 'answers_dir' not in self:
	147	+ msg = 'Missing "answers_dir" in configuration'
	148	+ raise TestFactoryException(msg)
	149	+
	150	+ def check_answers_directory_writable(self) -> None:
	151	+ '''Answers directory must be writable'''
	152	+ testfile = path.join(path.expanduser(self['answers_dir']), 'REMOVE-ME')
	153	+ try:
	154	+ with open(testfile, 'w') as file:
	155	+ file.write('You can safely remove this file.')
	156	+ except OSError as exc:
	157	+ msg = f'Cannot write answers to directory "{self["answers_dir"]}"'
	158	+ raise TestFactoryException(msg) from exc
	159	+
	160	+ def check_questions_directory(self) -> None:
	161	+ '''Check if questions directory is missing or not accessible.'''
	162	+ if 'questions_dir' not in self:
	163	+ logger.warning('Missing "questions_dir". Using "%s"',
	164	+ path.abspath(path.curdir))
	165	+ self['questions_dir'] = path.curdir
	166	+ elif not path.isdir(path.expanduser(self['questions_dir'])):
	167	+ raise TestFactoryException(f'Can\'t find questions directory '
	168	+ f'"{self["questions_dir"]}"')
	169	+
	170	+ def check_import_files(self) -> None:
	171	+ '''Check if there are files to import (with questions)'''
	172	+ if 'files' not in self:
	173	+ msg = ('Missing "files" in configuration with the list of '
	174	+ 'question files to import!')
	175	+ raise TestFactoryException(msg)
	176	+
	177	+ if isinstance(self['files'], str):
	178	+ self['files'] = [self['files']]
	179	+
	180	+ def check_question_list(self) -> None:
	181	+ '''normalize question list'''
	182	+ if 'questions' not in self:
	183	+ raise TestFactoryException('Missing "questions" in configuration')
	184	+
	185	+ for i, question in enumerate(self['questions']):
	186	+ # normalize question to a dict and ref to a list of references
	187	+ if isinstance(question, str): # e.g., - some_ref
	188	+ question = {'ref': [question]} # becomes - ref: [some_ref]
	189	+ elif isinstance(question, dict) and isinstance(question['ref'], str):
	190	+ question['ref'] = [question['ref']]
	191	+ elif isinstance(question, list):
	192	+ question = {'ref': [str(a) for a in question]}
	193	+
	194	+ self['questions'][i] = question
	195	+
	196	+ def check_missing_title(self) -> None:
	197	+ '''Warns if title is missing'''
	198	+ if not self['title']:
	199	+ logger.warning('Title is undefined!')
	200	+
	201	+ def check_grade_scaling(self) -> None:
	202	+ '''Just informs the scale limits'''
	203	+ if 'scale_points' in self:
	204	+ msg = ('* DEPRECATION WARNING: * scale_points, scale_min, '
	205	+ 'scale_max were replaced by "scale: [min, max]".')
	206	+ logger.warning(msg)
	207	+ self['scale'] = [self['scale_min'], self['scale_max']]
	208	+
	209	+
	210	+ # ------------------------------------------------------------------------
	211	+ def sanity_checks(self) -> None:
	212	+ '''
	213	+ Checks for valid keys and sets default values.
	214	+ Also checks if some files and directories exist
	215	+ '''
	216	+ self.check_test_ref()
	217	+ self.check_missing_database()
	218	+ self.check_missing_answers_directory()
	219	+ self.check_answers_directory_writable()
	220	+ self.check_questions_directory()
	221	+ self.check_import_files()
	222	+ self.check_question_list()
	223	+ self.check_missing_title()
	224	+ self.check_grade_scaling()
	225	+
	226	+ # ------------------------------------------------------------------------
	227	+ async def generate(self):
	228	+ '''
	229	+ Given a dictionary with a student dict {'name':'john', 'number': 123}
	230	+ returns instance of Test() for that particular student
	231	+ '''
	232	+
	233	+ # make list of questions
	234	+ questions = []
	235	+ qnum = 1 # track question number
	236	+ nerr = 0 # count errors during questions generation
	237	+
	238	+ for qlist in self['questions']:
	239	+ # choose list of question variants
	240	+ choose = qlist.get('choose', 1)
	241	+ qrefs = random.sample(qlist['ref'], k=choose)
	242	+
	243	+ for qref in qrefs:
	244	+ # generate instance of question
	245	+ try:
	246	+ question = await self.question_factory[qref].gen_async()
	247	+ except QuestionException:
	248	+ logger.error('Can\'t generate question "%s". Skipping.', qref)
	249	+ nerr += 1
	250	+ continue
	251	+
	252	+ # some defaults
	253	+ if question['type'] in ('information', 'success', 'warning',
	254	+ 'alert'):
	255	+ question['points'] = qlist.get('points', 0.0)
	256	+ else:
	257	+ question['points'] = qlist.get('points', 1.0)
	258	+ question['number'] = qnum # counter for non informative panels
	259	+ qnum += 1
	260	+
	261	+ questions.append(question)
	262	+
	263	+ # setup scale
	264	+ total_points = sum(q['points'] for q in questions)
	265	+
	266	+ if total_points > 0:
	267	+ # normalize question points to scale
	268	+ if self['scale'] is not None:
	269	+ scale_min, scale_max = self['scale']
	270	+ for question in questions:
	271	+ question['points'] *= (scale_max - scale_min) / total_points
	272	+ else:
	273	+ self['scale'] = [0, total_points]
	274	+ else:
	275	+ logger.warning('Total points is ZERO.')
	276	+ if self['scale'] is None:
	277	+ self['scale'] = [0, 20] # default
	278	+
	279	+ if nerr > 0:
	280	+ logger.error('%s errors found!', nerr)
	281	+
	282	+ # copy these from the test configuratoin to each test instance
	283	+ inherit = {'ref', 'title', 'database', 'answers_dir',
	284	+ 'questions_dir', 'files',
	285	+ 'duration', 'autosubmit',
	286	+ 'scale', 'show_points',
	287	+ 'show_ref', 'debug', }
	288	+ # NOT INCLUDED: testfile, allow_all, review
	289	+
	290	+ return Test({'questions': questions, **{k:self[k] for k in inherit}})
	291	+
	292	+ # ------------------------------------------------------------------------
	293	+ def __repr__(self):
	294	+ testsettings = '\n'.join(f' {k:14s}: {v}' for k, v in self.items())
	295	+ return '{\n' + testsettings + '\n}'
...	...