cr fixes

Author: gal432

lms/lmsweb/__init__.py

@@ -3,8 +3,8 @@
 import typing
 
 from flask import Flask
-from flask_httpauth import HTTPBasicAuth
 from flask_babel import Babel  # type: ignore
+from flask_httpauth import HTTPBasicAuth
 from flask_limiter import Limiter  # type: ignore
 from flask_limiter.util import get_remote_address  # type: ignore
 from flask_mail import Mail  # type: ignore
@@ -68,8 +68,6 @@ def get_password(username: str) -> typing.Optional[str]:
 def verify_password(username: str, client_password: str):
     username_username = models.User.username == username
     login_user = models.User.get_or_none(username_username)
-    if login_user is None:
-        return False
-    if not login_user.is_password_valid(client_password):
+    if login_user is None or not login_user.is_password_valid(client_password):
         return False
     return login_user

lms/lmsweb/config.py.example

@@ -67,8 +67,4 @@ LIMITS_PER_HOUR = 50
 # Change password settings
 MAX_INVALID_PASSWORD_TRIES = 5
 
-_REPOSITORY_FOLDER = os.getenv("REPOSITORY_FOLDER", os.path.abspath(os.path.join(os.curdir, "repositories")))
-
-
-def get_repository_folder():
-    return _REPOSITORY_FOLDER
+REPOSITORY_FOLDER = os.getenv("REPOSITORY_FOLDER", os.path.abspath(os.path.join(os.curdir, "repositories")))

lms/lmsweb/git_service.py

@@ -3,15 +3,14 @@
 import subprocess  # noqa: S404
 import tempfile
 import typing
-import logging
+import pathlib
 
 import flask
 
 from lms.lmsdb import models
 from lms.models import upload
 from lms.utils import hashing
-
-_logger = logging.getLogger(__name__)
+from lms.utils.log import log
 
 
 class _GitOperation(typing.NamedTuple):
@@ -23,6 +22,9 @@ class _GitOperation(typing.NamedTuple):
 
 
 class GitService:
+    _GIT_PROCESS_TIMEOUT = 20
+    _GIT_VALID_EXIT_CODE = 0
+
     def __init__(
             self,
             user: models.User,
@@ -40,19 +42,19 @@ def project_name(self) -> str:
         return f'{self._exercise_id}-{self._user.id}'
 
     @property
-    def repository_folder(self) -> str:
-        return os.path.join(self._base_repository_folder, self.project_name)
+    def repository_folder(self) -> pathlib.Path:
+        return pathlib.Path(self._base_repository_folder) / self.project_name
 
     def handle_operation(self) -> flask.Response:
         git_operation = self._extract_git_operation()
-        git_repository_folder = os.path.join(self.repository_folder, 'config')
+        repository_folder = self.repository_folder / 'config'
 
-        first_time_repository = not os.path.exists(git_repository_folder)
-        if first_time_repository:
+        new_repository = not repository_folder.exists()
+        if new_repository:
             self._initialize_bare_repository()
 
         if not git_operation.supported:
-            raise EnvironmentError
+            raise OSError
 
         data_out = self._execute_git_operation(git_operation)
 
@@ -71,38 +73,42 @@ def handle_operation(self) -> flask.Response:
 
         return self.build_response(data_out, git_operation)
 
-    def _execute_git_operation(self, git_operation: _GitOperation) -> bytes:
+    def _execute_command(
+            self,
+            args: typing.List[str],
+            cwd: typing.Union[str, pathlib.Path],
+            proc_input: typing.Optional[bytes] = None,
+    ):
         proc = subprocess.Popen(  # noqa: S603
-            args=git_operation.service_command,
+            args=args,
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
-            cwd=self._base_repository_folder,
+            cwd=cwd,
         )
-        data_out, _ = proc.communicate(self._request.data, 20)
-        if proc.wait() != 0:
-            _logger.error(
-                'Failed to execute command. stdout=%s\nstderr=%s',
-                proc.stdout.read(), proc.stderr.read(),
+        data_out, _ = proc.communicate(proc_input, self._GIT_PROCESS_TIMEOUT)
+        operation_failed = proc.wait() != self._GIT_VALID_EXIT_CODE
+        if operation_failed:
+            log.error(
+                'Failed to execute command %s. stdout=%s\nstderr=%s',
+                args, proc.stdout.read(), proc.stderr.read(),
             )
-            raise EnvironmentError
+            raise OSError
         return data_out
 
+    def _execute_git_operation(self, git_operation: _GitOperation) -> bytes:
+        return self._execute_command(
+            args=git_operation.service_command,
+            cwd=self._base_repository_folder,
+            proc_input=self._request.data,
+        )
+
     def _initialize_bare_repository(self) -> None:
         os.makedirs(self.repository_folder, exist_ok=True)
-        proc = subprocess.Popen(  # noqa: S603
+        self._execute_command(
             args=['git', 'init', '--bare'],
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
             cwd=self.repository_folder,
         )
-        if proc.wait() != 0:
-            _logger.error(
-                'Failed to execute command. stdout=%s\nstderr=%s',
-                proc.stdout.read(), proc.stderr.read(),
-            )
-            raise EnvironmentError
 
     @staticmethod
     def build_response(
@@ -167,7 +173,7 @@ def format_response_callback(response_bytes: bytes) -> bytes:
             format_response = format_response_callback
 
         else:
-            _logger.error(
+            log.error(
                 'Failed to find the git command for route %s',
                 self._request.path,
             )
@@ -182,33 +188,29 @@ def format_response_callback(response_bytes: bytes) -> bytes:
         )
 
     def _load_files_from_repository(self) -> typing.List[upload.File]:
+        """
+        Since the remote server is a git bare repository
+        we need to 'clone' the bare repository to resolve the files.
+        We are not changing the remote at any end - that is why we
+        don't care about git files here.
+        """
         with tempfile.TemporaryDirectory() as tempdir:
-            proc = subprocess.Popen(  # noqa: S603
+            self._execute_command(
                 args=['git', 'clone', self.repository_folder, '.'],
-                stdin=subprocess.PIPE,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
                 cwd=tempdir,
             )
-            return_code = proc.wait()
-            if return_code != 0:
-                _logger.error(
-                    'Failed to execute command. stdout=%s\nstderr=%s',
-                    proc.stdout.read(), proc.stderr.read(),
-                )
-                raise EnvironmentError
             to_return = []
             # remove git internal files
-            shutil.rmtree(os.path.join(tempdir, '.git'))
+            shutil.rmtree(pathlib.Path(tempdir) / '.git')
             for root, _, files in os.walk(tempdir):
                 for file in files:
                     upload_file = self._load_file(file, root, tempdir)
                     to_return.append(upload_file)
             return to_return
 
     @staticmethod
-    def _load_file(file_path: str, root: str, tempdir: str) -> upload.File:
-        with open(os.path.join(root, file_path)) as f:
-            file_path = os.path.join(os.path.relpath(root, tempdir), file_path)
+    def _load_file(file_name: str, root: str, tempdir: str) -> upload.File:
+        file_path = str(pathlib.Path(root).relative_to(tempdir) / file_name)
+        with open(pathlib.Path(root) / file_name) as f:
             upload_file = upload.File(path=file_path, code=f.read())
         return upload_file

lms/lmsweb/views.py

@@ -2,7 +2,7 @@
 
 import arrow  # type: ignore
 from flask import (
-    jsonify, make_response, render_template, request,
+    Response, jsonify, make_response, render_template, request,
     send_from_directory, session, url_for,
 )
 from flask_babel import gettext as _  # type: ignore
@@ -24,7 +24,7 @@
 )
 from lms.lmsweb.config import (
     CONFIRMATION_TIME, LANGUAGES, LIMITS_PER_HOUR,
-    LIMITS_PER_MINUTE, LOCALE, MAX_UPLOAD_SIZE, get_repository_folder,
+    LIMITS_PER_MINUTE, LOCALE, MAX_UPLOAD_SIZE, REPOSITORY_FOLDER,
 )
 from lms.lmsweb.forms.change_password import ChangePasswordForm
 from lms.lmsweb.forms.register import RegisterForm
@@ -547,14 +547,14 @@ def download(download_id: str):
 
 @webapp.route(f'{routes.GIT}/info/refs')
 @webapp.route(f'{routes.GIT}/git-receive-pack', methods=['POST'])
-@webapp.route(f'{routes.GIT}/git-upload-pack', methods=('POST',))
+@webapp.route(f'{routes.GIT}/git-upload-pack', methods=['POST'])
 @http_basic_auth.login_required
-def git_handler(exercise_id: int):
+def git_handler(exercise_id: int) -> Response:
     git_service = GitService(
         user=http_basic_auth.current_user(),
         exercise_id=exercise_id,
         request=request,
-        base_repository_folder=get_repository_folder(),
+        base_repository_folder=REPOSITORY_FOLDER,
     )
     return git_service.handle_operation()
 

tests/test_git_solution.py

@@ -2,11 +2,12 @@
 import os.path
 import shutil
 import tempfile
+from unittest import mock
 
 from flask.testing import FlaskClient
 
 from lms.lmsdb import models
-from lms.lmsweb import config, webapp
+from lms.lmsweb import webapp
 from tests import conftest
 
 
@@ -38,26 +39,21 @@ class TestSendSolutionFromGit:
     GET_METHOD = FlaskClient.get.__name__
     POST_METHOD = FlaskClient.post.__name__
 
-    temp_folder = None
+    temp_folder = ''
 
-    @classmethod
-    def setup_class(cls):
-        cls.temp_folder = tempfile.mkdtemp()
+    def setup_method(self, _method: str) -> None:
+        self.temp_folder = tempfile.mkdtemp()
 
-    def setup_method(self, method):
-        config._REPOSITORY_FOLDER = os.path.join(self.temp_folder, method.__name__)
-
-    @classmethod
-    def teardown_class(cls):
-        if cls.temp_folder and os.path.exists(cls.temp_folder):
-            shutil.rmtree(cls.temp_folder)
+    def teardown_method(self, _method: str) -> None:
+        if self.temp_folder and os.path.exists(self.temp_folder):
+            shutil.rmtree(self.temp_folder)
 
     @staticmethod
     def _get_formatted_git_url(exercise: models.Exercise, rel_path: str) -> str:
         return f'/git/{exercise.id}.git/{rel_path}'
 
-    @staticmethod
     def _send_git_request(
+            self,
             username: str,
             method_name: str,
             url: str,
@@ -71,7 +67,10 @@ def _send_git_request(
             ('Authorization', f'Basic {encoded_credentials}'),
         )
         query_string = {'service': service} if service is not None else None
-        return getattr(client, method_name)(url, query_string=query_string, headers=headers, data=data)
+
+        # patch the REPOSITORY_FOLDER to make new repository every test
+        with mock.patch('lms.lmsweb.views.REPOSITORY_FOLDER', self.temp_folder):
+            return getattr(client, method_name)(url, query_string=query_string, headers=headers, data=data)
 
     def test_not_authorized_access(self, exercise: models.Exercise, student_user: models.User):
         client = conftest.get_logged_user(student_user.username)