diff --git a/mod_api/__init__.py b/mod_api/__init__.py index a54a9413..3fb527b5 100644 --- a/mod_api/__init__.py +++ b/mod_api/__init__.py @@ -36,6 +36,7 @@ # Route modules register themselves against the blueprint; the rest of # the stack adds one module per PR. from mod_api.routes import auth as auth_routes # noqa: E402, F401 +from mod_api.routes import results as results_routes # noqa: E402, F401 from mod_api.routes import runs as runs_routes # noqa: E402, F401 from mod_api.routes import samples as samples_routes # noqa: E402, F401 from mod_api.routes import system as system_routes # noqa: E402, F401 diff --git a/mod_api/routes/results.py b/mod_api/routes/results.py new file mode 100644 index 00000000..57a49078 --- /dev/null +++ b/mod_api/routes/results.py @@ -0,0 +1,475 @@ +""" +Expected/actual output, diffs, and baseline approval routes. + +GET /runs/{id}/samples/{sid}/regression-tests/{rid}/outputs/{oid}/expected +GET /runs/{id}/samples/{sid}/regression-tests/{rid}/outputs/{oid}/actual +GET /runs/{id}/samples/{sid}/regression-tests/{rid}/outputs/{oid}/diff +POST /runs/{id}/samples/{sid}/baseline-approval Approve a new baseline +""" + +import base64 +import os + +from flask import current_app, g, redirect, request, url_for + +from mod_api import mod_api +from mod_api.middleware.auth import require_roles, require_scope +from mod_api.middleware.error_handler import make_error_response +from mod_api.middleware.validation import validate_body, validate_path_id +from mod_api.models.api_token import Scope +from mod_api.schemas.results import (BaselineApprovalRequestSchema, + BaselineApprovalSchema, + OutputFileContentSchema) +from mod_api.services.diff_service import compute_diff, file_sha256, read_lines +from mod_api.services.status import is_dummy_row +from mod_api.services.storage import (get_test_results_base_path, + resolve_artifact) +from mod_api.utils import safe_resolve, single_response +from mod_auth.models import Role +from mod_regression.models import RegressionTestOutputFiles +from mod_test.models import Test, TestResultFile + +INVALID_PATH_MSG = 'Invalid file path.' +READ_ERROR_MSG = 'Failed to read file.' + + +def _find_result_file(run_id, regression_test_id, output_id=None): + """ + Look up the right TestResultFile row. + + Uses run_id + regression_test_id from the path. If output_id is + given as a query param, narrow to that specific output file. + """ + query = TestResultFile.query.filter_by( + test_id=run_id, + regression_test_id=regression_test_id, + ) + + if output_id is not None: + query = query.filter_by(regression_test_output_id=output_id) + + return query.first() + + +def _validate_result_file_access(run_id, sample_id, regression_id, output_id): + """Validate access to a result file and return it, or an error response.""" + test = Test.query.filter(Test.id == run_id).first() + if test is None: + return None, make_error_response('not_found', f'Run {run_id} not found.', http_status=404) + + result_file = _find_result_file(run_id, regression_id, output_id) + + if result_file is None: + return None, make_error_response( + 'not_found', + f'No result for regression test {regression_id}.', + http_status=404, + ) + + actual_sample_id = ( + result_file.regression_test.sample_id + if result_file.regression_test else None + ) + if actual_sample_id != sample_id: + return None, make_error_response( + 'not_found', + f'Regression test {regression_id} does not belong to sample {sample_id}.', + http_status=404, + ) + + return result_file, None + + +def _read_output_file(file_path, fmt, is_expected=True): + """Read an output file into the response dict. + + The sha256 in the result is always computed over the complete file, + even when content is truncated to the 1 MiB inline limit. + """ + if not os.path.isfile(file_path): + type_str = 'Expected' if is_expected else 'Actual' + return None, make_error_response( + 'not_found', + f'{type_str} output file not found on disk.', + http_status=404, + ) + + sha256 = file_sha256(file_path) + file_size = os.path.getsize(file_path) + truncated = False + download_url = None + + if file_size > 1048576: + truncated = True + filename = os.path.basename(file_path) + download_url, _ = resolve_artifact(f'TestResults/{filename}') + + if fmt == 'text': + try: + with open(file_path, 'r', encoding='utf-8', errors='replace') as f: + content = f.read(1048576) + encoding = 'utf-8' + except Exception: + return None, make_error_response('internal_error', READ_ERROR_MSG, http_status=500) + else: + try: + with open(file_path, 'rb') as f: + content = base64.b64encode(f.read(1048576)).decode('ascii') + encoding = 'base64' + except Exception: + return None, make_error_response('internal_error', READ_ERROR_MSG, http_status=500) + + return { + 'content': content, + 'encoding': encoding, + 'sha256': sha256, + 'truncated': truncated, + 'download_url': download_url, + }, None + + +@mod_api.route( + '/runs//samples//regression-tests//outputs//expected', + methods=['GET'] +) +@require_scope(Scope.RESULTS_READ) +@validate_path_id('run_id') +@validate_path_id('sample_id') +@validate_path_id('regression_id') +@validate_path_id('output_id') +def get_expected_output(run_id, sample_id, regression_id, output_id): + """Return the expected output file for a regression test result.""" + result_file, err = _validate_result_file_access( + run_id, sample_id, regression_id, output_id) + if err: + return err + + if is_dummy_row(result_file): + return make_error_response('not_found', 'Expected output not found.', http_status=404) + + base_path = get_test_results_base_path() + expected_filename = result_file.expected + ext = '' + if result_file.regression_test_output: + ext = result_file.regression_test_output.correct_extension + if ext: + ext = ext.replace('/', '').replace('\\', '').replace('..', '') + expected_filename += ext + + file_path = safe_resolve(base_path, expected_filename) + if file_path is None: + return make_error_response('forbidden', INVALID_PATH_MSG, http_status=403) + + fmt = request.args.get('format', 'base64') + + data, err = _read_output_file(file_path, fmt, is_expected=True) + if err: + return err + + content = data['content'] + encoding = data['encoding'] + sha256 = data['sha256'] + truncated = data['truncated'] + download_url = data['download_url'] + + _, storage_status = resolve_artifact(f'TestResults/{expected_filename}') + + return single_response({ + 'run_id': run_id, + 'sample_id': sample_id, + 'regression_id': result_file.regression_test_id, + 'output_id': result_file.regression_test_output_id, + 'filename': expected_filename, + 'content_type': 'application/octet-stream', + 'encoding': encoding, + 'content': content, + 'truncated': truncated, + 'download_url': download_url, + 'sha256': sha256, + 'storage_status': storage_status, + }, schema=OutputFileContentSchema()) + + +@mod_api.route( + '/runs//samples//regression-tests//outputs//actual', + methods=['GET'] +) +@require_scope(Scope.RESULTS_READ) +@validate_path_id('run_id') +@validate_path_id('sample_id') +@validate_path_id('regression_id') +@validate_path_id('output_id') +def get_actual_output(run_id, sample_id, regression_id, output_id): + """ + Return the actual output file for a regression test result. + + got=null in the DB means the output matched expected — not that it's + missing. We return 303 (redirect to expected) in that case. Missing + output (the dummy sentinel row) returns 404. + """ + result_file, err = _validate_result_file_access( + run_id, sample_id, regression_id, output_id) + if err: + return err + + if is_dummy_row(result_file): + return make_error_response( + 'missing_output', + 'Test produced no output when output was expected.', + http_status=404, + ) + + if result_file.got is None: + # Relative redirect: clients only resend Authorization headers on + # same-origin redirects, and an absolute URL would break behind a + # reverse proxy anyway. + return redirect(url_for( + 'api.get_expected_output', + run_id=run_id, + sample_id=sample_id, + regression_id=regression_id, + output_id=output_id, + format=request.args.get('format', 'base64'), + ), code=303) + + base_path = get_test_results_base_path() + actual_filename = result_file.got + if result_file.regression_test_output: + ext = result_file.regression_test_output.correct_extension + if ext: + ext = ext.replace('/', '').replace('\\', '').replace('..', '') + actual_filename += ext + + file_path = safe_resolve(base_path, actual_filename) + if file_path is None: + return make_error_response('forbidden', INVALID_PATH_MSG, http_status=403) + + fmt = request.args.get('format', 'base64') + + data, err = _read_output_file(file_path, fmt, is_expected=False) + if err: + return err + + content = data['content'] + encoding = data['encoding'] + sha256 = data['sha256'] + truncated = data['truncated'] + download_url = data['download_url'] + + _, storage_status = resolve_artifact(f'TestResults/{actual_filename}') + + return single_response({ + 'run_id': run_id, + 'sample_id': sample_id, + 'regression_id': result_file.regression_test_id, + 'output_id': result_file.regression_test_output_id, + 'filename': actual_filename, + 'content_type': 'application/octet-stream', + 'encoding': encoding, + 'content': content, + 'truncated': truncated, + 'download_url': download_url, + 'sha256': sha256, + 'storage_status': storage_status, + }, schema=OutputFileContentSchema()) + + +def _handle_missing_diff(result_file, format_type, diff_ids): + if is_dummy_row(result_file): + if format_type == 'unified': + return single_response({**diff_ids, 'format': 'unified', 'content': ''}) + return single_response({ + **diff_ids, + 'status': 'missing_actual', + 'format': 'structured', + 'summary': {'added_lines': 0, 'removed_lines': 0, 'changed_hunks': 0}, + 'hunks': [], + }) + + if result_file.got is None: + if format_type == 'unified': + return single_response({**diff_ids, 'format': 'unified', 'content': ''}) + return single_response({ + **diff_ids, + 'status': 'identical', + 'format': 'structured', + 'summary': {'added_lines': 0, 'removed_lines': 0, 'changed_hunks': 0}, + 'hunks': [], + }) + return None + + +@mod_api.route( + '/runs//samples//regression-tests//outputs//diff', + methods=['GET'] +) +@require_scope(Scope.RESULTS_READ) +@validate_path_id('run_id') +@validate_path_id('sample_id') +@validate_path_id('regression_id') +@validate_path_id('output_id') +def get_diff(run_id, sample_id, regression_id, output_id): + """Structured diff between expected and actual output.""" + result_file, err = _validate_result_file_access( + run_id, sample_id, regression_id, output_id) + if err: + return err + + diff_ids = { + 'run_id': run_id, + 'sample_id': sample_id, + 'regression_id': result_file.regression_test_id, + 'output_id': result_file.regression_test_output_id, + } + + format_type = request.args.get('format', 'structured') + + missing_response = _handle_missing_diff(result_file, format_type, diff_ids) + if missing_response: + return missing_response + + base_path = get_test_results_base_path() + ext = result_file.regression_test_output.correct_extension if result_file.regression_test_output else '' + if ext: + ext = ext.replace('/', '').replace('\\', '').replace('..', '') + expected_path = safe_resolve(base_path, result_file.expected + ext) + actual_path = safe_resolve(base_path, result_file.got + ext) + + if expected_path is None or actual_path is None: + return make_error_response('forbidden', INVALID_PATH_MSG, http_status=403) + + if not os.path.isfile(expected_path): + return make_error_response('not_found', 'Expected output file not found on disk.', http_status=404) + if not os.path.isfile(actual_path): + return make_error_response('not_found', 'Actual output file not found on disk.', http_status=404) + + max_diff_bytes = 10 * 1024 * 1024 # 10 MiB + if os.path.getsize(expected_path) > max_diff_bytes or os.path.getsize(actual_path) > max_diff_bytes: + return make_error_response('unprocessable', 'File too large for diff. Use download_url.', http_status=422) + + if format_type == 'unified': + import difflib + expected_lines = read_lines(expected_path) + actual_lines = read_lines(actual_path) + differ = list(difflib.unified_diff( + expected_lines, + actual_lines, + fromfile='expected', + tofile='actual', + lineterm='' + )) + if len(differ) > 10000: + differ = differ[:10000] + differ.append("\n... Diff truncated due to length ...") + unified_content = '\n'.join(differ) + return single_response({ + **diff_ids, + 'format': 'unified', + 'content': unified_content + }) + + context_lines = request.args.get('context_lines', 3, type=int) + context_lines = max(1, min(context_lines, 50)) + + diff_result = compute_diff( + expected_path, actual_path, context_lines=context_lines) + diff_result.update(diff_ids) + diff_result['format'] = 'structured' + return single_response(diff_result) + + +@mod_api.route('/runs//samples//baseline-approval', methods=['POST']) +@require_roles([Role.admin]) +@require_scope(Scope.BASELINES_WRITE) +@validate_path_id('run_id') +@validate_path_id('sample_id') +@validate_body(BaselineApprovalRequestSchema) +def create_baseline_approval(run_id, sample_id, validated_data=None): + """ + Record intent to approve actual output as the new expected baseline. + + WARNING: When remove_variants is set to true, this action will remove all + platform-specific variants, making this output the single source of truth + across all platforms. Care should be taken as this applies globally. + """ + test = Test.query.filter(Test.id == run_id).first() + if test is None: + return make_error_response('not_found', f'Run {run_id} not found.', http_status=404) + + regression_id = validated_data['regression_id'] + output_id = validated_data['output_id'] + + result_file = TestResultFile.query.filter_by( + test_id=run_id, + regression_test_id=regression_id, + regression_test_output_id=output_id, + ).first() + + if result_file is None: + return make_error_response('not_found', 'Result file not found.', http_status=404) + + actual_sample_id = ( + result_file.regression_test.sample_id + if result_file.regression_test else None + ) + if actual_sample_id != sample_id: + return make_error_response( + 'not_found', + f'Regression test {regression_id} does not belong to sample {sample_id}.', + http_status=404, + ) + + if is_dummy_row(result_file): + return make_error_response('unprocessable', 'Cannot approve a dummy row.', http_status=422) + + if result_file.got is None: + return make_error_response('unprocessable', 'Output already matches expected.', http_status=422) + + # The actual output file (named by its hash) is already in TestResults/. + # We just need to update the RegressionTestOutput to point to this new hash. + rto = result_file.regression_test_output + if rto is None: + return make_error_response('internal_error', 'No RegressionTestOutput linked.', http_status=500) + + new_baseline = result_file.got + + base_path = get_test_results_base_path() + ext = rto.correct_extension or '' + if ext: + ext = ext.replace('/', '').replace('\\', '').replace('..', '') + actual_filename = new_baseline + ext + file_path = safe_resolve(base_path, actual_filename) + if not file_path or not os.path.isfile(file_path): + return make_error_response('unprocessable', 'Actual output file not found in storage.', http_status=422) + + old_baseline = rto.correct + rto.correct = new_baseline + + remove_variants = validated_data.get('remove_variants', False) + if remove_variants: + RegressionTestOutputFiles.query.filter_by( + regression_test_output_id=rto.id).delete() + + g.db.commit() + + # Audit log: this mutates the global expected baseline (and optionally + # deletes all variants), so record who approved what. + approver = getattr(g, 'api_user', None) + current_app.logger.info( + 'Baseline approved by %s (user_id=%s): run=%s regression=%s ' + 'output=%s old_hash=%s new_hash=%s remove_variants=%s', + approver.name if approver else 'unknown', + approver.id if approver else None, + run_id, regression_id, output_id, old_baseline, new_baseline, + remove_variants) + + import datetime + return single_response({ + 'status': 'approved', + 'run_id': run_id, + 'sample_id': sample_id, + 'regression_id': regression_id, + 'output_id': output_id, + 'requested_by': getattr(g, 'api_user').name if getattr(g, 'api_user', None) else 'unknown', + 'created_at': datetime.datetime.now(datetime.timezone.utc) + }, schema=BaselineApprovalSchema()) diff --git a/mod_api/schemas/results.py b/mod_api/schemas/results.py new file mode 100644 index 00000000..fe054dc2 --- /dev/null +++ b/mod_api/schemas/results.py @@ -0,0 +1,64 @@ +"""Schemas for output file content and baseline approvals.""" + +from marshmallow import RAISE, Schema, fields, validate + +from mod_api.schemas.common import DATETIME_FORMAT + + +class OutputFileContentSchema(Schema): + """File content blob returned for expected or actual output.""" + + run_id = fields.Integer(allow_none=True) + sample_id = fields.Integer(required=True) + regression_id = fields.Integer(required=True) + output_id = fields.Integer(required=True) + filename = fields.String(required=True) + content_type = fields.String(required=True) + encoding = fields.String( + required=True, validate=validate.OneOf(['utf-8', 'base64'])) + content = fields.String(required=True) + # sha256 always covers the complete file, even when content is truncated. + sha256 = fields.String(allow_none=True) + truncated = fields.Boolean(load_default=False) + download_url = fields.String(allow_none=True) + storage_status = fields.String( + required=True, + validate=validate.OneOf(['ok', 'degraded', 'missing']), + ) + + +class BaselineApprovalRequestSchema(Schema): + """POST /runs/{id}/samples/{sid}/baseline-approval body.""" + + regression_id = fields.Integer( + required=True, + validate=validate.Range(min=1), + ) + output_id = fields.Integer( + required=True, + validate=validate.Range(min=1), + ) + + remove_variants = fields.Boolean( + load_default=False, + ) + + class Meta: + """Reject unknown fields.""" + + unknown = RAISE + + +class BaselineApprovalSchema(Schema): + """Response after a baseline approval is applied.""" + + status = fields.String( + required=True, + validate=validate.OneOf( + ['approved'])) + run_id = fields.Integer(required=True) + sample_id = fields.Integer(required=True) + regression_id = fields.Integer(required=True) + output_id = fields.Integer(required=True) + requested_by = fields.String(required=True) + created_at = fields.DateTime(required=True, format=DATETIME_FORMAT) diff --git a/mod_api/services/diff_service.py b/mod_api/services/diff_service.py new file mode 100644 index 00000000..f527c57b --- /dev/null +++ b/mod_api/services/diff_service.py @@ -0,0 +1,220 @@ +""" +Structured diff computation between expected and actual output files. + +Produces JSON hunks with line-level detail instead of the legacy HTML +diff output. Uses difflib.unified_diff internally. +""" + +import difflib +import hashlib +import os +import re +from typing import Any, Dict, List, Optional, Tuple + +from mod_api.services.storage import get_test_results_base_path + + +def compute_diff( + expected_path: str, + actual_path: str, + context_lines: int = 3, + max_hunks: int = 500, +) -> Dict[str, Any]: + """ + Compute a structured diff between two files. + + Returns a dict matching the Diff schema: status, summary (added_lines, + removed_lines, changed_hunks), and a list of hunks. + """ + context_lines = max(1, min(context_lines, 50)) + + if not os.path.isfile(expected_path): + return { + 'status': 'missing_expected', + 'summary': {'added_lines': 0, 'removed_lines': 0, 'changed_hunks': 0}, + 'hunks': [], + } + + if not os.path.isfile(actual_path): + return { + 'status': 'missing_actual', + 'summary': {'added_lines': 0, 'removed_lines': 0, 'changed_hunks': 0}, + 'hunks': [], + } + + expected_lines = read_lines(expected_path) + actual_lines = read_lines(actual_path) + + if expected_lines == actual_lines: + return { + 'status': 'identical', + 'summary': {'added_lines': 0, 'removed_lines': 0, 'changed_hunks': 0}, + 'hunks': [], + } + + hunks = _compute_hunks(expected_lines, actual_lines, + context_lines, max_hunks) + added = sum( + 1 for h in hunks for line in h['lines'] if line['kind'] == 'added') + removed = sum( + 1 for h in hunks for line in h['lines'] if line['kind'] == 'removed') + + return { + 'status': 'different', + 'summary': { + 'added_lines': added, + 'removed_lines': removed, + 'changed_hunks': len(hunks), + }, + 'hunks': hunks, + } + + +# Matches the @@ -a,b +c,d @@ header line from unified_diff. +_HUNK_RE = re.compile(r'^@@ -(\d+)(?:,\d+)? \+(\d+)(?:,\d+)? @@') + + +def _process_diff_line(line, current_hunk, expected_line_num, actual_line_num): + if line.startswith('+'): + current_hunk['lines'].append({ + 'kind': 'added', + 'expected_line': None, + 'actual_line': actual_line_num, + 'text': line[1:], + }) + actual_line_num += 1 + elif line.startswith('-'): + current_hunk['lines'].append({ + 'kind': 'removed', + 'expected_line': expected_line_num, + 'actual_line': None, + 'text': line[1:], + }) + expected_line_num += 1 + else: + content = line[1:] if line.startswith(' ') else line + current_hunk['lines'].append({ + 'kind': 'context', + 'expected_line': expected_line_num, + 'actual_line': actual_line_num, + 'text': content, + }) + expected_line_num += 1 + actual_line_num += 1 + return expected_line_num, actual_line_num + + +def _process_hunk_header( + line: str, + current_hunk: Optional[Dict[str, Any]], + hunks: List[Dict[str, Any]], + max_hunks: int +) -> Tuple[Optional[Dict[str, Any]], int, int, bool]: + if current_hunk and len(hunks) >= max_hunks: + return None, 0, 0, True + if current_hunk: + hunks.append(current_hunk) + + m = _HUNK_RE.match(line) + if m: + expected_line_num = int(m.group(1)) + actual_line_num = int(m.group(2)) + else: + expected_line_num = 0 + actual_line_num = 0 + + new_hunk = { + 'expected_start': expected_line_num, + 'actual_start': actual_line_num, + 'lines': [], + } + return new_hunk, expected_line_num, actual_line_num, False + + +def _compute_hunks( + expected_lines: List[str], + actual_lines: List[str], + context_lines: int, + max_hunks: int, +) -> List[Dict[str, Any]]: + """Parse unified_diff output into structured hunk dicts.""" + differ = difflib.unified_diff( + expected_lines, + actual_lines, + lineterm='', + n=context_lines, + ) + + hunks: List[Dict[str, Any]] = [] + current_hunk: Optional[Dict[str, Any]] = None + expected_line_num = 0 + actual_line_num = 0 + + for line in differ: + if line.startswith(('---', '+++')): + continue + + if line.startswith('@@'): + current_hunk, expected_line_num, actual_line_num, stop = _process_hunk_header( + line, current_hunk, hunks, max_hunks + ) + if stop: + break + continue + + if current_hunk is None: + continue + + expected_line_num, actual_line_num = _process_diff_line( + line, current_hunk, expected_line_num, actual_line_num) + + if current_hunk: + hunks.append(current_hunk) + + return hunks[:max_hunks] + + +def _enforce_safe_path(file_path: str) -> bool: + base = os.path.realpath(get_test_results_base_path()) + target = os.path.realpath(file_path) + return target.startswith(base + os.sep) or target == base + + +def read_lines(file_path: str, max_size_bytes: int = 10 * 1024 * 1024) -> List[str]: + """Read file lines with a cp1252 fallback, matching legacy behavior. + + Parameters + ---------- + file_path : str + Absolute path to the file to read. + max_size_bytes : int + Maximum file size in bytes. Raises ValueError if exceeded. + """ + if not _enforce_safe_path(file_path): + raise ValueError("Unsafe file path") + file_size = os.path.getsize(file_path) + if file_size > max_size_bytes: + raise ValueError( + f"File too large ({file_size} bytes > {max_size_bytes} limit)") + try: + with open(file_path, encoding='utf8') as f: + return [line.rstrip('\n\r') for line in f.readlines()] + except UnicodeDecodeError: + # errors='replace' because cp1252 leaves five byte values undefined — + # without it the fallback can itself raise on binary output files. + with open(file_path, encoding='cp1252', errors='replace') as f: + return [line.rstrip('\n\r') for line in f.readlines()] + + +def file_sha256(file_path: str) -> Optional[str]: + """Compute SHA-256 of a file. Returns None if the file can't be read.""" + if not _enforce_safe_path(file_path): + return None + try: + sha = hashlib.sha256() + with open(file_path, 'rb') as f: + for block in iter(lambda: f.read(8192), b''): + sha.update(block) + return sha.hexdigest() + except (OSError, IOError): + return None diff --git a/tests/api/test_middleware_auth.py b/tests/api/test_middleware_auth.py index 983600ba..a2e16013 100644 --- a/tests/api/test_middleware_auth.py +++ b/tests/api/test_middleware_auth.py @@ -125,6 +125,12 @@ def test_scope_boundary_write_endpoints_fail_on_read_only_scopes(self): self.assertEqual(res.status_code, 403) self.assertEqual(res.json['code'], 'forbidden') + # 3. POST /runs/1/samples/1/baseline-approval + res = self.client.post('/api/v1/runs/1/samples/1/baseline-approval', + headers={'Authorization': f'Bearer {plaintext}'}) + self.assertEqual(res.status_code, 403) + self.assertEqual(res.json['code'], 'forbidden') + def test_multiple_candidates_same_prefix(self): plaintext1, token1 = self.generate_db_token(self.user, scopes=['system:read']) plaintext2, token2 = self.generate_db_token(self.user, scopes=['system:read']) diff --git a/tests/api/test_routes_results.py b/tests/api/test_routes_results.py new file mode 100644 index 00000000..74555253 --- /dev/null +++ b/tests/api/test_routes_results.py @@ -0,0 +1,365 @@ +import base64 +import json +import os +import tempfile +from unittest.mock import patch + +from flask import g + +from mod_api.middleware.rate_limit import _rate_limit_store +from mod_auth.models import Role, User +from mod_regression.models import (Category, InputType, OutputType, + RegressionTest, RegressionTestOutput) +from mod_test.models import TestResult, TestResultFile +from tests.api.base import ApiTestCase + + +class TestRoutesResults(ApiTestCase): + def setUp(self): + super().setUp() + self.setup_run_data('res') + + category = Category('Test Category', 'Description') + g.db.add(category) + g.db.commit() + + self.reg_test = RegressionTest( + 1, 'command', InputType.file, OutputType.file, category.id, 0) + g.db.add(self.reg_test) + g.db.commit() + self.reg_test_id = self.reg_test.id + + self.reg_out = RegressionTestOutput( + self.reg_test_id, 'expected_hash', '.txt', 'exp_file') + g.db.add(self.reg_out) + g.db.commit() + self.reg_out_id = self.reg_out.id + + self.test_result = TestResult(self.test_id, self.reg_test_id, 0, 0, 0) + g.db.add(self.test_result) + g.db.commit() + + self.result_file = TestResultFile( + self.test_id, self.reg_test_id, self.reg_out_id, 'expected_hash', 'actual_hash') + g.db.add(self.result_file) + g.db.commit() + + self.test_dir = tempfile.TemporaryDirectory() + self.dir_path = self.test_dir.name + + # Create TestResults directory + self.test_results_dir = os.path.join(self.dir_path, 'TestResults') + os.makedirs(self.test_results_dir, exist_ok=True) + + # Configure app to use our temp dir + self.original_sample_repo = self.app.config.get('SAMPLE_REPOSITORY') + self.app.config['SAMPLE_REPOSITORY'] = self.dir_path + + _rate_limit_store.clear() + + def tearDown(self): + if self.original_sample_repo is not None: + self.app.config['SAMPLE_REPOSITORY'] = self.original_sample_repo + else: + self.app.config.pop('SAMPLE_REPOSITORY', None) + self.test_dir.cleanup() + super().tearDown() + + def test_get_expected_output_base64(self): + expected_file_path = os.path.join( + self.test_results_dir, 'expected_hash.txt') + with open(expected_file_path, 'wb') as f: + f.write(b'expected data') + + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + token = self.get_token( + 'res_user@local.com', 'userpass123', 't1', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/{self.reg_test_id}' + f'/outputs/{self.reg_out_id}/expected', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['encoding'], 'base64') + self.assertEqual(res.json['content'], base64.b64encode( + b'expected data').decode('ascii')) + self.assertEqual(res.json['filename'], 'expected_hash.txt') + + def test_get_expected_output_text(self): + expected_file_path = os.path.join( + self.test_results_dir, 'expected_hash.txt') + with open(expected_file_path, 'wb') as f: + f.write(b'line1\nline2') + + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + token = self.get_token( + 'res_user@local.com', 'userpass123', 't2', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/{self.reg_test_id}' + f'/outputs/{self.reg_out_id}/expected?format=text', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['encoding'], 'utf-8') + self.assertEqual(res.json['content'], 'line1\nline2') + + def test_get_actual_output(self): + actual_file_path = os.path.join( + self.test_results_dir, 'actual_hash.txt') + with open(actual_file_path, 'wb') as f: + f.write(b'actual data') + + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + token = self.get_token( + 'res_user@local.com', 'userpass123', 't3', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/{self.reg_test_id}' + f'/outputs/{self.reg_out_id}/actual', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['filename'], 'actual_hash.txt') + self.assertEqual(res.json['content'], base64.b64encode( + b'actual data').decode('ascii')) + + def test_get_actual_output_matched_expected(self): + # Set got = None + self.result_file.got = None + g.db.commit() + + expected_file_path = os.path.join( + self.test_results_dir, 'expected_hash.txt') + with open(expected_file_path, 'wb') as f: + f.write(b'expected data') + + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + token = self.get_token( + 'res_user@local.com', 'userpass123', 't4', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/{self.reg_test_id}' + f'/outputs/{self.reg_out_id}/actual', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 303) + redirect_url = res.headers['Location'] + res2 = self.client.get(redirect_url, headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res2.status_code, 200) + + import base64 + self.assertEqual(res2.json['content'], base64.b64encode( + b'expected data').decode('ascii')) + + def test_get_diff(self): + expected_file_path = os.path.join( + self.test_results_dir, 'expected_hash.txt') + with open(expected_file_path, 'wb') as f: + f.write(b'line1\nline2\n') + + actual_file_path = os.path.join( + self.test_results_dir, 'actual_hash.txt') + with open(actual_file_path, 'wb') as f: + f.write(b'line1\nline_new\n') + + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + token = self.get_token( + 'res_user@local.com', 'userpass123', 't5', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/{self.reg_test_id}' + f'/outputs/{self.reg_out_id}/diff', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['status'], 'different') + self.assertEqual(res.json['summary']['added_lines'], 1) + + def test_get_diff_unified_format(self): + expected_file_path = os.path.join( + self.test_results_dir, 'expected_hash.txt') + with open(expected_file_path, 'wb') as f: + f.write(b'line1\nline2\n') + + actual_file_path = os.path.join( + self.test_results_dir, 'actual_hash.txt') + with open(actual_file_path, 'wb') as f: + f.write(b'line1\nline_new\n') + + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + token = self.get_token( + 'res_user@local.com', 'userpass123', 't5_uni', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/{self.reg_test_id}' + f'/outputs/{self.reg_out_id}/diff?format=unified', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['format'], 'unified') + self.assertIn('content', res.json) + self.assertIsInstance(res.json['content'], str) + + def test_get_diff_identical_files(self): + # When got is None, diff returns status 'identical' + self.result_file.got = None + g.db.commit() + + expected_file_path = os.path.join( + self.test_results_dir, 'expected_hash.txt') + with open(expected_file_path, 'wb') as f: + f.write(b'expected data\n') + + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + token = self.get_token( + 'res_user@local.com', 'userpass123', 't5_id', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/{self.reg_test_id}' + f'/outputs/{self.reg_out_id}/diff', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['status'], 'identical') + + def test_create_baseline_approval(self): + token = self.get_token('res_admin@local.com', + 'adminpass123', 't6', scopes=['baselines:write']) + + actual_file_path = os.path.join(self.test_results_dir, 'actual_hash.txt') + with open(actual_file_path, 'wb') as f: + f.write(b'actual data') + + payload = { + 'regression_id': self.reg_test_id, + 'output_id': self.reg_out_id, + 'remove_variants': False + } + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + res = self.client.post(f'/api/v1/runs/{self.test_id}/samples/1/baseline-approval', data=json.dumps( + payload), content_type='application/json', headers={'Authorization': f'Bearer {token}'}) + + if res.status_code != 200: + print("ERROR JSON:", res.json) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['status'], 'approved') + + # Verify db change + reg_out_after = RegressionTestOutput.query.get(self.reg_out_id) + self.assertEqual(reg_out_after.correct, 'actual_hash') + + def test_create_baseline_approval_forbidden_role(self): + # Create token directly in DB to bypass token creation limitations + from mod_api.models.api_token import ApiToken + plaintext = ApiToken.generate_token() + token = ApiToken( + user_id=self.user.id, # res_user has user role + token_name='t7_forbidden', + token_hash=ApiToken.hash_token(plaintext), + token_prefix=ApiToken.extract_prefix(plaintext), + scopes=['baselines:write'], + expires_in_days=7 + ) + g.db.add(token) + g.db.commit() + + payload = { + 'regression_id': self.reg_test_id, + 'output_id': self.reg_out_id + } + res = self.client.post(f'/api/v1/runs/{self.test_id}/samples/1/baseline-approval', data=json.dumps( + payload), content_type='application/json', headers={'Authorization': f'Bearer {plaintext}'}) + + self.assertEqual(res.status_code, 403) + self.assertEqual(res.json['code'], 'forbidden') + + def test_create_baseline_approval_contributor_forbidden(self): + # Baseline approval is admin-only: a contributor must be rejected + # even when holding a baselines:write token. + from mod_api.models.api_token import ApiToken + from mod_auth.models import Role, User + contributor = User( + 'res_contrib', Role.contributor, 'res_contrib@local.com', + User.generate_hash('contribpass123')) + g.db.add(contributor) + g.db.commit() + + plaintext = ApiToken.generate_token() + token = ApiToken( + user_id=contributor.id, + token_name='t_contrib_forbidden', + token_hash=ApiToken.hash_token(plaintext), + token_prefix=ApiToken.extract_prefix(plaintext), + scopes=['baselines:write'], + expires_in_days=7, + ) + g.db.add(token) + g.db.commit() + + payload = { + 'regression_id': self.reg_test_id, + 'output_id': self.reg_out_id, + } + res = self.client.post( + f'/api/v1/runs/{self.test_id}/samples/1/baseline-approval', + data=json.dumps(payload), content_type='application/json', + headers={'Authorization': f'Bearer {plaintext}'}) + + self.assertEqual(res.status_code, 403) + self.assertEqual(res.json['code'], 'forbidden') + + def test_create_baseline_approval_remove_variants(self): + token = self.get_token('res_admin@local.com', + 'adminpass123', 't8', scopes=['baselines:write']) + + actual_file_path = os.path.join(self.test_results_dir, 'actual_hash.txt') + with open(actual_file_path, 'wb') as f: + f.write(b'actual data') + + payload = { + 'regression_id': self.reg_test_id, + 'output_id': self.reg_out_id, + 'remove_variants': True + } + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + res = self.client.post(f'/api/v1/runs/{self.test_id}/samples/1/baseline-approval', data=json.dumps( + payload), content_type='application/json', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['status'], 'approved') + + # Verify db change + from mod_regression.models import RegressionTestOutputFiles + variants = RegressionTestOutputFiles.query.filter_by( + regression_test_output_id=self.reg_out_id).count() + self.assertEqual(variants, 0) + + def test_create_baseline_approval_output_already_matches(self): + # got=None means the actual output already matches the baseline, + # so there is nothing to approve. + self.result_file.got = None + g.db.commit() + + token = self.get_token('res_admin@local.com', + 'adminpass123', 't9', scopes=['baselines:write']) + payload = { + 'regression_id': self.reg_test_id, + 'output_id': self.reg_out_id + } + res = self.client.post( + f'/api/v1/runs/{self.test_id}/samples/1/baseline-approval', + data=json.dumps(payload), content_type='application/json', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 422) + self.assertIn('matches expected', res.json['message']) + + def test_get_actual_output_missing_storage(self): + # We don't write the file 'actual_hash.txt', so it will not be found on the filesystem + with patch.dict('run.config', {'SAMPLE_REPOSITORY': self.dir_path}): + token = self.get_token( + 'res_user@local.com', 'userpass123', 't9', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/{self.reg_test_id}' + f'/outputs/{self.reg_out_id}/actual', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 404) + self.assertIn('not found', res.json['message'].lower()) + + def test_get_output_nonexistent_resource_404(self): + token = self.get_token('res_user@local.com', + 'userpass123', 't10', scopes=['results:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/1/regression-tests/999999' + f'/outputs/{self.reg_out_id}/expected', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 404) + self.assertEqual(res.json['code'], 'not_found') diff --git a/tests/api/test_services_diff_service.py b/tests/api/test_services_diff_service.py new file mode 100644 index 00000000..4beb34da --- /dev/null +++ b/tests/api/test_services_diff_service.py @@ -0,0 +1,126 @@ +import os +import tempfile + +from mod_api.services.diff_service import (_compute_hunks, compute_diff, + file_sha256, read_lines) +from tests.api.base import ApiTestCase + + +class TestDiffService(ApiTestCase): + def setUp(self): + super().setUp() + self.test_dir = tempfile.TemporaryDirectory() + self.dir_path = self.test_dir.name + from unittest.mock import patch + patcher = patch( + 'mod_api.services.diff_service._enforce_safe_path', return_value=True) + self.addCleanup(patcher.stop) + self.mock_safe = patcher.start() + + def tearDown(self): + self.test_dir.cleanup() + super().tearDown() + + def create_file(self, filename, content, encoding='utf-8'): + path = os.path.join(self.dir_path, filename) + with open(path, 'w', encoding=encoding) as f: + f.write(content) + return path + + def test_compute_diff_identical(self): + content = "line1\nline2\n" + path1 = self.create_file("file1.txt", content) + path2 = self.create_file("file2.txt", content) + + diff = compute_diff(path1, path2) + self.assertEqual(diff['status'], 'identical') + self.assertEqual(diff['summary']['added_lines'], 0) + self.assertEqual(diff['summary']['removed_lines'], 0) + self.assertEqual(len(diff['hunks']), 0) + + def test_compute_diff_missing_expected(self): + path2 = self.create_file("file2.txt", "content") + + diff = compute_diff(os.path.join(self.dir_path, "missing.txt"), path2) + self.assertEqual(diff['status'], 'missing_expected') + + def test_compute_diff_missing_actual(self): + path1 = self.create_file("file1.txt", "content") + + diff = compute_diff(path1, os.path.join(self.dir_path, "missing.txt")) + self.assertEqual(diff['status'], 'missing_actual') + + def test_compute_diff_different(self): + content1 = "line1\nline2\nline3\n" + content2 = "line1\nline_new\nline3\n" + path1 = self.create_file("file1.txt", content1) + path2 = self.create_file("file2.txt", content2) + + diff = compute_diff(path1, path2) + self.assertEqual(diff['status'], 'different') + self.assertEqual(diff['summary']['added_lines'], 1) + self.assertEqual(diff['summary']['removed_lines'], 1) + self.assertEqual(diff['summary']['changed_hunks'], 1) + self.assertEqual(len(diff['hunks']), 1) + + hunk = diff['hunks'][0] + self.assertEqual(hunk['expected_start'], 1) + self.assertEqual(hunk['actual_start'], 1) + + def test_compute_diff_context_lines_clamped(self): + content1 = "\n".join(str(i) for i in range(1, 201)) + "\n" + content2 = content1.replace("\n100\n", "\n100_new\n") + path1 = self.create_file("file1.txt", content1) + path2 = self.create_file("file2.txt", content2) + + diff = compute_diff(path1, path2, context_lines=200) + self.assertEqual(diff['status'], 'different') + hunk = diff['hunks'][0] + # max context is 50 before and 50 after, plus 1 removed and 1 added = 102 lines total + self.assertEqual(len(hunk['lines']), 102) + + def test_compute_hunks_max_hunks(self): + lines1 = ["1", "2", "3", "4", "5"] + lines2 = ["1a", "2", "3a", "4", "5a"] + # With context_lines=0 we should get 3 separate hunks + hunks = _compute_hunks(lines1, lines2, context_lines=0, max_hunks=2) + self.assertEqual(len(hunks), 2) # bounded to 2 + + def test_compute_hunks_parsing(self): + lines1 = ["common", "remove_me", "common"] + lines2 = ["common", "add_me", "common"] + hunks = _compute_hunks(lines1, lines2, context_lines=1, max_hunks=10) + self.assertEqual(len(hunks), 1) + lines = hunks[0]['lines'] + self.assertEqual(lines[0]['kind'], 'context') + self.assertEqual(lines[1]['kind'], 'removed') + self.assertEqual(lines[2]['kind'], 'added') + self.assertEqual(lines[3]['kind'], 'context') + + def test_read_lines_utf8(self): + path = os.path.join(self.dir_path, "utf8.txt") + with open(path, 'w', encoding='utf-8', newline='') as f: + f.write("line1\r\nline2\n") + lines = read_lines(path) + self.assertEqual(lines, ["line1", "line2"]) + + def test_read_lines_cp1252(self): + path = os.path.join(self.dir_path, "cp1252.txt") + # Write bytes that are valid cp1252 but invalid utf-8 + with open(path, 'wb') as f: + # \x80 is euro sign in cp1252, invalid start byte in utf-8 + f.write(b"line1\r\n\x80line2") + + lines = read_lines(path) + # \x80 maps to \u20ac + self.assertEqual(lines, ["line1", "\u20acline2"]) + + def test_file_sha256(self): + path = self.create_file("sha.txt", "hello") + sha = file_sha256(path) + # sha256("hello") = 2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824 + self.assertEqual( + sha, "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824") + + self.assertIsNone(file_sha256( + os.path.join(self.dir_path, "nonexistent.txt")))