Add mypy types and pathlib (#4)

Author: niosus

homework_checker/check_homework.py

@@ -34,7 +34,7 @@ def main():
     # Read the job file.
     log.debug('Reading from file "%s"', args.input)
     checker = Checker(args.input)
-    results = checker.check_homework()
+    results = checker.check_all_homeworks()
     md_writer = MdWriter()
     md_writer.update(results)
     # Write the resulting markdown file.

homework_checker/checker.py

@@ -1,9 +1,9 @@
 """Check the homework."""
 
-from os import path
-
 import logging
 from datetime import datetime
+from typing import Dict
+from pathlib import Path
 
 from . import tools
 from .schema_manager import SchemaManager
@@ -13,47 +13,51 @@
 
 log = logging.getLogger("GHC")
 
+HomeworkResultDict = Dict[str, Dict[str, tools.CmdResult]]
+
 
 class Checker:
     """Check homework."""
 
     TESTS_TAG = "tests"
 
-    def __init__(self, job_file_path):
+    def __init__(self: "Checker", job_file_path: Path):
         """Initialize the checker from file."""
         self._job_file_path = tools.expand_if_needed(job_file_path)
         schema_manager = SchemaManager(self._job_file_path)
         self._base_node = schema_manager.validated_yaml
         self._checked_code_folder = tools.expand_if_needed(
-            self._base_node[Tags.FOLDER_TAG]
+            Path(self._base_node[Tags.FOLDER_TAG])
         )
-        # The results of all tests will be kept here.
-        self._results = {}
 
-    def check_homework(self):
-        """Run over all Tasks in all homeworks."""
-        results = {}
-        for homework_node in self._base_node[Tags.HOMEWORKS_TAG]:
-            current_folder = path.join(
-                self._checked_code_folder, homework_node[Tags.FOLDER_TAG]
+    def check_homework(self: "Checker", homework_node: dict) -> HomeworkResultDict:
+        """Run over all Tasks in a single homework."""
+        results: HomeworkResultDict = {}
+        current_folder = Path(self._checked_code_folder, homework_node[Tags.FOLDER_TAG])
+        if not current_folder.exists():
+            log.warning("Folder '%s' does not exist. Skiping.", current_folder)
+            return results
+        deadline_str = homework_node[Tags.DEADLINE_TAG]
+        deadline_datetime = datetime.strptime(deadline_str, tools.DATE_PATTERN)
+        if datetime.now() > deadline_datetime:
+            results[tools.EXPIRED_TAG] = {}
+        for task_node in homework_node[Tags.TASKS_TAG]:
+            task = Task.from_yaml_node(
+                task_node=task_node,
+                student_hw_folder=current_folder,
+                job_file=self._job_file_path,
             )
-            if not path.exists(current_folder):
-                log.warning("Folder '%s' does not exist. Skiping.", current_folder)
+            if not task:
                 continue
-            hw_name = homework_node[Tags.NAME_TAG]
-            results[hw_name] = {}
-            deadline_str = homework_node[Tags.DEADLINE_TAG]
-            deadline_datetime = datetime.strptime(deadline_str, tools.DATE_PATTERN)
-            if datetime.now() > deadline_datetime:
-                results[hw_name][tools.EXPIRED_TAG] = True
-            for task_node in homework_node[Tags.TASKS_TAG]:
-                task = Task.from_yaml_node(
-                    task_node=task_node,
-                    student_hw_folder=current_folder,
-                    job_file=self._job_file_path,
-                )
-                if not task:
-                    continue
-                results[hw_name][task.name] = task.check_all_tests()
+            results[task.name] = task.check_all_tests()
+        return results
 
+    def check_all_homeworks(self: "Checker") -> Dict[str, HomeworkResultDict]:
+        """Run over all Tasks in all homeworks."""
+        results: Dict[str, HomeworkResultDict] = {}
+        for homework_node in self._base_node[Tags.HOMEWORKS_TAG]:
+            hw_name = homework_node[Tags.NAME_TAG]
+            current_homework_results = self.check_homework(homework_node)
+            if current_homework_results:
+                results[hw_name] = current_homework_results
         return results

homework_checker/md_writer.py

@@ -1,6 +1,10 @@
 """Write test results into a markdown file."""
+from typing import Dict
+from pathlib import Path
 
 from .tools import EXPIRED_TAG
+from .tools import CmdResult
+from .checker import HomeworkResultDict
 
 TABLE_TEMPLATE = "| {hw_name} | {task_name} | {test_name} | {result_sign} |\n"
 TABLE_SEPARATOR = "|---|---|---|:---:|\n"
@@ -34,7 +38,7 @@
 class MdWriter:
     """Write given tests results into a markdown file."""
 
-    def __init__(self):
+    def __init__(self: "MdWriter"):
         """Initialize the writer."""
         self._md_table = TABLE_TEMPLATE.format(
             hw_name="Homework Name",
@@ -45,7 +49,7 @@ def __init__(self):
         self._md_table += TABLE_SEPARATOR
         self._errors = ""  # Markdown part with errors.
 
-    def update(self, hw_results):
+    def update(self: "MdWriter", hw_results: Dict[str, HomeworkResultDict]):
         """Update the table of completion."""
         for hw_name, hw_dict in sorted(hw_results.items()):
             need_hw_name = True
@@ -74,7 +78,7 @@ def update(self, hw_results):
                     need_hw_name = False  # We only print homework name once.
                     need_task_name = False  # We only print Task name once.
 
-    def write_md_file(self, md_file_path):
+    def write_md_file(self: "MdWriter", md_file_path: Path):
         """Write all the added content to the md file."""
         md_file_content = "# Test results\n"
         md_file_content += self._md_table
@@ -86,7 +90,14 @@ def write_md_file(self, md_file_path):
         with open(md_file_path, "w") as md_file:
             md_file.write(md_file_content)
 
-    def _add_error(self, hw_name, task_name, test_name, test_result, expired):
+    def _add_error(
+        self: "MdWriter",
+        hw_name: str,
+        task_name: str,
+        test_name: str,
+        test_result: CmdResult,
+        expired: bool,
+    ):
         """Add a section of errors to the md file."""
         if test_result.succeeded():
             return

homework_checker/schema_manager.py

@@ -1,25 +1,31 @@
 """Manage creation of schema."""
+from __future__ import annotations
 import sys
 import logging
 import operator
-from os import path
-from schema import Schema, SchemaError, Or, Optional
-from ruamel.yaml import YAML
-from ruamel.yaml.comments import CommentedMap, CommentedSeq
+from pathlib import Path
+from typing import Union, List, Dict, Any
+from ruamel.yaml import YAML  # type: ignore[import]
+from ruamel.yaml.comments import CommentedMap, CommentedSeq  # type: ignore[import]
+from schema import Schema, SchemaError, Or, Optional  # type: ignore[import]
 
 from .tools import MAX_DATE_STR
 from .schema_tags import Tags, OutputTags, BuildTags, LangTags
 
 log = logging.getLogger("GHC")
 
-SCHEMA_FILE = path.join(path.dirname(path.dirname(__file__)), "schema", "schema.yml")
+SCHEMA_FILE = Path(__file__).parent.parent / "schema" / "schema.yml"
 
 
 class SchemaManager:
     """Manage schema creation."""
 
-    def __init__(self, file_name):
+    GenericListType = List[Union[Any, Dict[str, Any], List[Any]]]
+    GenericDictType = Dict[str, Union[Any, Dict[str, Any], List[Any]]]
+
+    def __init__(self: SchemaManager, file_name: Path):
         """Create a schema for my tests."""
+        # TODO(igor): simplify by moving parts out of here into separate variables.
         self.__schema = Schema(
             {
                 Tags.FOLDER_TAG: str,
@@ -63,8 +69,9 @@ def __init__(self, file_name):
             }
         )
         yaml = YAML()
-        yaml.width = 4096  # big enough value to prevent wrapping
-        yaml.explicit_start = True
+        # big enough value to prevent wrapping
+        yaml.width = 4096  # type: ignore[assignment]
+        yaml.explicit_start = True  # type: ignore[assignment]
         yaml.indent(mapping=2, sequence=4, offset=2)
         with open(file_name, "r") as stream:
             contents = SchemaManager.__to_simple_dict(yaml.load(stream))
@@ -83,8 +90,11 @@ def __init__(self, file_name):
         except OSError:
             log.debug("Cannot write schema file. We only use this while developing.")
 
-    def __to_simple_list(commented_seq):
-        simple_list = []
+    @staticmethod
+    def __to_simple_list(
+        commented_seq: List[Union[Any, CommentedSeq, CommentedMap]]
+    ) -> SchemaManager.GenericListType:
+        simple_list: SchemaManager.GenericListType = []
         for value in commented_seq:
             if isinstance(value, CommentedSeq):
                 simple_list.append(SchemaManager.__to_simple_list(value))
@@ -94,8 +104,11 @@ def __to_simple_list(commented_seq):
                 simple_list.append(value)
         return simple_list
 
-    def __to_simple_dict(commented_map):
-        simple_dict = {}
+    @staticmethod
+    def __to_simple_dict(
+        commented_map: Dict[str, Union[Any, CommentedSeq, CommentedMap]],
+    ) -> SchemaManager.GenericDictType:
+        simple_dict: SchemaManager.GenericDictType = {}
         for key, value in commented_map.items():
             if isinstance(value, CommentedMap):
                 simple_dict[key] = SchemaManager.__to_simple_dict(value)
@@ -106,7 +119,7 @@ def __to_simple_dict(commented_map):
         return simple_dict
 
     @property
-    def validated_yaml(self):
+    def validated_yaml(self) -> dict:
         """Return validated yaml."""
         return self.__validated_yaml
 
@@ -115,9 +128,13 @@ def schema(self):
         """Return schema."""
         return self.__schema
 
+    # pylint: disable=R0911
+    # This method needs this many returns.
     @staticmethod
-    def __sanitize_value(input_var):
+    def __sanitize_value(input_var: Union[dict, Optional, Or, Any]):
         """Use the schema and create an example file from it."""
+        # pylint: disable=W0212
+        # This seems to be the only way to get to schema value.
         if isinstance(input_var, dict):
             new_dict = {}
             for key, val in input_var.items():

homework_checker/schema_tags.py

@@ -1,4 +1,5 @@
 """Collection of schema related tags."""
+from typing import Any
 
 
 class Tags:
@@ -49,14 +50,14 @@ class LangTags:
 class OneOf:
     """Check that an item is one of the list."""
 
-    def __init__(self, some_list):
+    def __init__(self: "OneOf", some_list: list):
         """Set the list to choose from."""
         self.__items = some_list
 
-    def __call__(self, item):
+    def __call__(self: "OneOf", item: Any):
         """Check that the list contains what is needed."""
         return item in self.__items
 
-    def __str__(self):
+    def __str__(self: "OneOf"):
         """Override str for this class."""
         return "Possible values: {}".format(self.__items)