Update run_tests() to return results directly (fixes #17)

Changes run_tests() to accept parameters and return List[TaskTestResult]
instead of writing results to a file and reading args from environment.
This enables direct module invocation from task.py.

- Rename _TestResult to TaskTestResult (public API)
- Add result_will_be_submitted and only_run_these_tasks parameters
- Update checks.py to use config module for paths
- Add SUCCESS/FAILURE constants to utils.py

Author: Nils Bars

ref_utils/checks.py

@@ -1,11 +1,12 @@
 """Various checks you may want to run during submission tests"""
-from pathlib import Path
-from typing import List, Optional
-
 import os
+import subprocess
+from pathlib import Path
+from typing import List
 
-from .utils import print_ok, print_warn, print_err, SUCCESS, FAILURE
+from .config import get_config
 from .process import run
+from .utils import FAILURE, SUCCESS, print_err, print_ok, print_warn
 
 _NO_LINT_ENV_VAR = "NO_LINT"
 _ENV_VAL_TRUE = "1"
@@ -16,10 +17,9 @@ def contains_flag(flag: str, python_script: Path, silent: bool = False) -> bool:
     Run submitted file and match whether it contains the flag value.
     """
     cmd: List[str] = ["python3", python_script.as_posix()]
-    output: Optional[str] = run(cmd, check_returncode=False, timeout=10)
-    if output is None:
-        return FAILURE
-    if not flag in output:
+    result = run(cmd, check_signal=False, timeout=10, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    output = result.stdout.decode() if result.stdout else ""
+    if flag not in output:
         if not silent:
             print_err("[!] Failed to find flag")
         return FAILURE
@@ -34,10 +34,10 @@ def run_pylint(python_files: List[Path]) -> bool:
     """
     if not python_files or os.environ.get(_NO_LINT_ENV_VAR, '') == _ENV_VAL_TRUE:
         return SUCCESS
-    lint_output = run(["pylint", "--exit-zero", "--rcfile", "/etc/pylintrc"] +
-                      [str(f.resolve()) for f in python_files])
-    if lint_output is None:
-        return FAILURE
+    result = run(["pylint", "--exit-zero", "--rcfile", str(get_config().pylint_config_path)] +
+                 [str(f.resolve()) for f in python_files],
+                 stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    lint_output = result.stdout.decode() if result.stdout else ""
     if lint_output != "":
         print_warn("[!] pylint's syntax and coding style checks failed:")
         print_warn('    ' + '\n    '.join(lint_output.split('\n')))
@@ -52,9 +52,10 @@ def run_mypy(python_files: List[Path]) -> bool:
     """
     if not python_files or os.environ.get(_NO_LINT_ENV_VAR, '') == _ENV_VAL_TRUE:
         return SUCCESS
-    lint_output = run(["mypy", "--config-file", "/etc/mypyrc"] + [str(f.resolve()) for f in python_files])
-    if lint_output is None:
-        return FAILURE
+    cmd = ["mypy", "--config-file", str(get_config().mypy_config_path)]
+    cmd += [str(f.resolve()) for f in python_files]
+    result = run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    lint_output = result.stdout.decode() if result.stdout else ""
     if lint_output != "":
         print_warn("[!] mypy's type checks failed:")
         print_warn('    ' + '\n    '.join(lint_output.split('\n')))
@@ -68,7 +69,7 @@ def check_all_python_files() -> bool:
     Run checks only suited for Python files (mypy + pylint)
     """
     tests_passed = True
-    python_files = [f for f in Path("/home/user").glob("**/*.py") if not f.name.startswith(".")]
+    python_files = [f for f in get_config().user_home_path.glob("**/*.py") if not f.name.startswith(".")]
     if not python_files or os.environ.get(_NO_LINT_ENV_VAR, '') == _ENV_VAL_TRUE:
         return tests_passed
     print_ok(f'[+] Testing {len(python_files)} Python source code files')

ref_utils/decorator.py

@@ -1,17 +1,14 @@
-from functools import wraps
-from collections import defaultdict
-from pathlib import Path
-from typing import Any, Callable, List
+import os
 import typing as ty
+import warnings
+from dataclasses import dataclass
+from functools import wraps
+from typing import Any, Callable
 
 from ref_utils.error import RefUtilsError
-from .utils import print_ok, print_err
-from dataclasses import dataclass, asdict
-import warnings
-import json
-import os
 
-TEST_RESULT_PATH = Path("/var/test_result")
+from .utils import print_err, print_ok
+
 DEFAULT_TASK_NAME = 'default'
 __registered_tasks: ty.Dict[str, '_Task'] = {}
 
@@ -29,9 +26,10 @@ class TestResult():
 
 
 @dataclass
-class _TestResult():
+class TaskTestResult():
     """
-    Class used to serialize data before sending it to the webserver.
+    The result of a task's test, including the task name.
+    Used for serialization when sending results to the webserver.
     """
     task_name: str
     success: bool
@@ -111,30 +109,38 @@ def wrapper(*args: str, **kwargs: Any) -> Any:
 
 
 
-def run_tests() -> None:
+def run_tests(
+    *,
+    result_will_be_submitted: bool = False,
+    only_run_these_tasks: ty.Optional[ty.Sequence[str]] = None,
+) -> ty.List[TaskTestResult]:
     """
-    Must be called by the test script to execute all tests.
+    Execute all registered tests.
+
+    Args:
+        result_will_be_submitted: If True, indicates this is a final submission.
+        only_run_these_tasks: Optional list of task names to run. If None, runs all tasks.
+
+    Returns:
+        A list of TaskTestResult objects containing the results of each task.
     """
+    # Set env var so test_result_will_be_submitted() works within tests
+    if result_will_be_submitted:
+        os.environ["RESULT_WILL_BE_SUBMITTED"] = "1"
+
     print_ok('[+] Running tests..')
     all_tests_passed = True
     has_multiple_tasks = len(__registered_tasks) > 1
-    task_test_results: ty.List[_TestResult] = []
-
-    # This is set by task.py if the user only wants to run a subset of tests.
-    only_run_these_tasks = os.environ.get("ONLY_RUN_THESE_TASKS")
-    if only_run_these_tasks:
-        only_run_these_tasks = only_run_these_tasks.split(":")
+    task_test_results: ty.List[TaskTestResult] = []
 
     # Run all sub-tasks one after another.
     for task_name, tests in __registered_tasks.items():
         task_passed = True
 
-        TEST_RESULT_PATH.unlink(missing_ok=True)
-
         if has_multiple_tasks:
             print_ok(f'[+] *** Running tests for task \"{task_name}\" ***')
             if only_run_these_tasks and task_name not in only_run_these_tasks:
-                print_ok(f"[+] User requested to exclude task, skipping...")
+                print_ok("[+] User requested to exclude task, skipping...")
                 continue
 
         if tests.env_tests and not tests.submission_test and not tests.extended_submission_test:
@@ -150,7 +156,7 @@ def run_tests() -> None:
 
         #Do not run submission tests if the environ is invalid
         if not task_passed:
-            task_test_results.append(_TestResult(task_name, False, None))
+            task_test_results.append(TaskTestResult(task_name, False, None))
             if has_multiple_tasks:
                 # Only print this if we have multiple tasks. If we only have one,
                 # the would just duplicate the error printed at the end.
@@ -170,9 +176,9 @@ def run_tests() -> None:
                 ret = False
 
             if isinstance(ret, bool):
-                ret = _TestResult(task_name, ret, None)
+                ret = TaskTestResult(task_name, ret, None)
             elif isinstance(ret, TestResult):
-                ret = _TestResult(task_name, ret.success, ret.score)
+                ret = TaskTestResult(task_name, ret.success, ret.score)
             else:
                 raise RefUtilsError(f"Submission test returned unexpected type: {type(ret)}")
 
@@ -181,7 +187,7 @@ def run_tests() -> None:
             all_tests_passed &= ret.success
         else:
             # If there is no test, we consider this to be an success.
-            task_test_results.append(_TestResult(task_name, True, None))
+            task_test_results.append(TaskTestResult(task_name, True, None))
             print_ok("[+] No test found")
 
         if not task_passed and has_multiple_tasks:
@@ -197,5 +203,4 @@ def run_tests() -> None:
     else:
         print_ok('[+] All tests passed! Good job. Ready to submit!')
 
-    results = json.dumps([asdict(e) for e in task_test_results])
-    TEST_RESULT_PATH.write_text(results)
+    return task_test_results

ref_utils/utils.py

@@ -1,12 +1,18 @@
 """ Utility functions including colored printing or subprocess.run wrapper dropping privileges"""
-import sys
 import os
+import sys
 import typing as t
+from pathlib import Path
 from typing import Any, AnyStr, List, Union
 
-from pathlib import Path
 from colorama import Fore, Style
 
+from .config import get_config
+
+# Constants for return values
+SUCCESS: bool = True
+FAILURE: bool = False
+
 
 def print_ok(*args: str, **kwargs: Any) -> None:
     """Print green to signal correctness"""
@@ -46,7 +52,7 @@ def get_user_environment() -> t.Dict[str, Union[str, bytes]]:
         defined during submission.
     """
     ret: t.Dict[str, Union[str, bytes]] = {}
-    content = Path('/tmp/.user_environ').read_text()
+    content = get_config().user_environ_path.read_text()
     lines = content.split('\x00')
     for line in lines:
         if line == '':
@@ -88,4 +94,4 @@ def map_path_as_posix(cmd: List[Union[Path, str, bytes]]) -> List[Union[str, byt
             ret.append(c.as_posix())
         else:
             ret.append(c)
-    return ret
+    return ret