From 007ca0f7eeec3fe45bb3cdd694a89a41ee66b032 Mon Sep 17 00:00:00 2001 From: Pulkit Chauhan Date: Mon, 20 Jul 2026 15:24:45 +0530 Subject: [PATCH] PR 3: System Routes and Run Execution Endpoints --- mod_api/__init__.py | 2 + mod_api/routes/runs.py | 677 +++++++++++++++++++++++ mod_api/routes/system.py | 204 +++++++ mod_api/schemas/runs.py | 99 ++++ mod_api/services/error_service.py | 307 ++++++++++ mod_api/services/storage.py | 77 +++ mod_auth/controllers.py | 22 +- tests/api/test_middleware_auth.py | 167 ++++++ tests/api/test_routes_runs.py | 407 ++++++++++++++ tests/api/test_routes_system.py | 101 ++++ tests/api/test_services_error_service.py | 208 +++++++ tests/api/test_services_storage.py | 131 +++++ 12 files changed, 2396 insertions(+), 6 deletions(-) create mode 100644 mod_api/routes/runs.py create mode 100644 mod_api/routes/system.py create mode 100644 mod_api/schemas/runs.py create mode 100644 mod_api/services/error_service.py create mode 100644 mod_api/services/storage.py create mode 100644 tests/api/test_middleware_auth.py create mode 100644 tests/api/test_routes_runs.py create mode 100644 tests/api/test_routes_system.py create mode 100644 tests/api/test_services_error_service.py create mode 100644 tests/api/test_services_storage.py diff --git a/mod_api/__init__.py b/mod_api/__init__.py index a4b28241b..dc930a635 100644 --- a/mod_api/__init__.py +++ b/mod_api/__init__.py @@ -36,3 +36,5 @@ # 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 runs as runs_routes # noqa: E402, F401 +from mod_api.routes import system as system_routes # noqa: E402, F401 diff --git a/mod_api/routes/runs.py b/mod_api/routes/runs.py new file mode 100644 index 000000000..1ff75a91d --- /dev/null +++ b/mod_api/routes/runs.py @@ -0,0 +1,677 @@ +""" +Test run routes. + +GET /runs List runs (filtered, paginated, sorted) +POST /runs Trigger a new run +GET /runs/{id} Single run details +GET /runs/{id}/summary Pass/fail/skip counts +GET /runs/{id}/progress Progress event timeline +GET /runs/{id}/config Run configuration and test matrix +POST /runs/{id}/cancel Cancel a queued or running test +""" + +from collections import defaultdict + +from flask import g, request +from sqlalchemy import func, or_ +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import joinedload + +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_date_range, + validate_offset_pagination, + validate_path_id, validate_sort) +from mod_api.models.api_token import Scope +from mod_api.schemas.runs import (ProgressEventSchema, RunCreateRequestSchema, + RunSchema, RunSummarySchema) +from mod_api.services.error_service import derive_errors_for_run +from mod_api.services.status import (batch_get_run_data, derive_run_status, + derive_sample_status) +from mod_api.utils import get_sort_column, paginated_response, single_response +from mod_auth.models import Role +from mod_customized.models import CustomizedTest +from mod_regression.models import RegressionTest, RegressionTestOutput +from mod_test.models import (Fork, Test, TestPlatform, TestProgress, + TestResult, TestResultFile, TestStatus, TestType) + + +def _serialize_run(test): + """Turn a Test row into the Run response shape the spec expects.""" + return _batch_serialize([test])[0] + + +def _batch_serialize(tests, statuses=None, timestamps=None): + if statuses is None or timestamps is None: + statuses, timestamps = batch_get_run_data(tests) + return [ + { + 'run_id': t.id, + 'status': statuses.get(t.id, 'queued'), + 'platform': t.platform.value, + 'test_type': 'pr' if t.test_type == TestType.pull_request else 'commit', + 'repository': t.fork.github_name if t.fork else 'unknown', + 'branch': t.branch, + 'commit_sha': t.commit, + 'pr_number': t.pr_nr if t.pr_nr and t.pr_nr > 0 else None, + 'created_at': timestamps.get(t.id, {}).get('created_at'), + 'queued_at': timestamps.get(t.id, {}).get('queued_at'), + 'started_at': timestamps.get(t.id, {}).get('started_at'), + 'completed_at': timestamps.get(t.id, {}).get('completed_at'), + 'github_link': t.github_link if t.fork else None, + } + for t in tests + ] + + +def _apply_repository_filter(query, repository): + repo_field = RunCreateRequestSchema().fields.get('repository') + if repo_field: + try: + repo_field.deserialize(repository) + except Exception as e: + return None, make_error_response( + 'validation_error', + 'Invalid repository format.', + details={'fields': {'repository': str(e)}}, + http_status=400, + ) + fork_url = f'https://github.com/{repository}.git' + return query.join(Fork).filter(Fork.github == fork_url), None + + +def _apply_date_filters(query, created_after, created_before): + first_progress = ( + g.db.query( + TestProgress.test_id, func.min( + TestProgress.timestamp).label('min_ts')) .group_by( + TestProgress.test_id) .subquery()) + # LEFT JOIN so queued runs (no TestProgress rows yet, hence no known + # creation time) aren't silently dropped — otherwise combining a date + # filter with ?status=queued always returns an empty page. Runs without + # timestamps are treated as matching any requested window. + query = query.outerjoin( + first_progress, Test.id == first_progress.c.test_id) + if created_after: + query = query.filter(or_(first_progress.c.min_ts >= created_after, + first_progress.c.min_ts.is_(None))) + if created_before: + query = query.filter(or_(first_progress.c.min_ts <= created_before, + first_progress.c.min_ts.is_(None))) + return query + + +def _apply_run_filters(query, created_after, created_before): + platform = request.args.get('platform') + if platform: + try: + platform_enum = TestPlatform.from_string(platform) + query = query.filter(Test.platform == platform_enum) + except Exception: + valid_platforms = ', '.join(TestPlatform.values()) + return None, make_error_response( + 'validation_error', + f'Invalid platform: {platform}. Must be one of: {valid_platforms}.', + http_status=400, + ) + + branch = request.args.get('branch') + if branch: + query = query.filter(Test.branch == branch) + + commit_sha = request.args.get('commit_sha') + if commit_sha: + query = query.filter(Test.commit == commit_sha) + + repository = request.args.get('repository') + if repository: + query, err = _apply_repository_filter(query, repository) + if err: + return None, err + + if created_after or created_before: + query = _apply_date_filters(query, created_after, created_before) + + return query, None + + +def _validate_run_permissions(user, target_repo, main_repo_full): + # GitHub owner/repo names are case-insensitive. + if target_repo.lower() == main_repo_full.lower(): + allowed = (Role.admin, Role.tester, Role.contributor) + if user.role not in allowed: + return make_error_response( + 'forbidden', + 'Only admins, testers, and contributors can trigger runs for the main repository.', + details={ + 'required_roles': [role.value for role in allowed], + 'repository': target_repo, + }, + http_status=403, + ) + else: + owner = target_repo.split('/')[0] + github_login = getattr(user, 'github_login', None) + + if not github_login and getattr(user, 'github_token', None): + from mod_auth.controllers import fetch_username_from_token + github_login = fetch_username_from_token(user) + if github_login: + user.github_login = github_login + g.db.add(user) + + github_login = github_login or '' + + if not github_login or owner.lower() != github_login.lower(): + return make_error_response( + 'forbidden', + f'You can only trigger runs for your own repository (expected owner: {github_login}) ' + 'or the main repository.', + details={ + 'repository': target_repo, + 'owner_required': github_login, + }, + http_status=403, + ) + return None + + +def _validate_regression_test_ids(regression_test_ids): + if regression_test_ids is not None: + if not regression_test_ids: + return None, make_error_response( + 'validation_error', + 'regression_test_ids cannot be empty.', + details={'fields': { + 'regression_test_ids': 'Must contain at least one ID.'}}, + http_status=400, + ) + active_tests = RegressionTest.query.filter( + RegressionTest.id.in_(regression_test_ids), + RegressionTest.active == True, # noqa: E712 + ).all() + active_ids = {t.id for t in active_tests} + inactive_ids = [ + tid for tid in regression_test_ids if tid not in active_ids] + if inactive_ids: + return None, make_error_response( + 'unprocessable', + 'Some regression test IDs are inactive or do not exist.', + details={'inactive_ids': inactive_ids}, + http_status=422, + ) + else: + active_tests = RegressionTest.query.filter_by(active=True).all() + regression_test_ids = [t.id for t in active_tests] + return regression_test_ids, None + + +@mod_api.route('/runs', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_offset_pagination() +@validate_sort() +@validate_date_range +def list_runs( + limit=50, + offset=0, + sort='-created_at', + created_after=None, + created_before=None): + """List runs with filters for platform, branch, commit, repo, status, and date range.""" + query, err = _apply_run_filters(Test.query, created_after, created_before) + if err: + return err + + sort_map = { + 'run_id': Test.id, + 'created_at': Test.id, # best proxy - Test has no created_at column + } + order = get_sort_column(sort, sort_map) + if order is not None: + query = query.order_by(order) + else: + query = query.order_by(Test.id.desc()) + + status_filter = request.args.get('status') + if status_filter: + if status_filter not in ('queued', 'running', 'canceled'): + return make_error_response( + 'validation_error', + f'Filtering by status "{status_filter}" is not supported. Supported: queued, running, canceled.', + http_status=400, + ) + + latest_progress_sq = ( + g.db.query(func.max(TestProgress.id).label('max_id')) + .group_by(TestProgress.test_id) + .subquery() + ) + + if status_filter == 'queued': + query = query.outerjoin(TestProgress).filter( + TestProgress.id.is_(None)) + elif status_filter == 'running': + query = query.join( + TestProgress, + TestProgress.test_id == Test.id) .filter( + TestProgress.id.in_(latest_progress_sq)) .filter( + TestProgress.status.in_( + [ + TestStatus.preparation, + TestStatus.testing])) + elif status_filter == 'canceled': + query = query.join(TestProgress, TestProgress.test_id == Test.id)\ + .filter(TestProgress.id.in_(latest_progress_sq))\ + .filter(TestProgress.status == TestStatus.canceled) + + total = query.count() + tests = query.offset(offset).limit(limit).all() + serialized = _batch_serialize(tests) + return paginated_response( + serialized, + total, + limit, + offset, + schema=RunSchema()) + + +def _get_or_create_fork(fork_url): + fork = Fork.query.filter(Fork.github == fork_url).first() + if fork is None: + fork = Fork(fork_url) + g.db.add(fork) + try: + g.db.flush() + except IntegrityError: + g.db.rollback() + fork = Fork.query.filter(Fork.github == fork_url).first() + if fork is None: + return None, make_error_response( + 'internal_error', 'Failed to create or resolve fork.', http_status=500) + return fork, None + + +def _ci_artifact_exists(commit_sha, platform): + """Return True if a CI build artifact exists for this commit + platform. + + The worker runs prebuilt binaries downloaded from GitHub Actions rather + than building from source, so a run can only execute if a build artifact + keyed to ``commit_sha`` exists on the main repo (this is also true for + fork PR commits, whose artifacts are produced by the main repo's PR + workflow). Mirrors verify_artifacts_exist() in the webhook path. + + Fails open (returns True) if GitHub can't be reached, so run creation + never depends on a successful artifact lookup — the cron still guards + against genuinely missing artifacts. + """ + from run import config, log + try: + from github import Auth, Github + + from mod_ci.controllers import find_artifact_for_commit + gh = Github(auth=Auth.Token(config.get('GITHUB_TOKEN', ''))) + repo = gh.get_repo( + f"{config.get('GITHUB_OWNER', '')}/{config.get('GITHUB_REPOSITORY', '')}") + return find_artifact_for_commit(repo, commit_sha, platform, log) is not None + except Exception: + log.exception( + 'create_run: artifact pre-check failed; allowing run to proceed') + return True + + +@mod_api.route('/runs', methods=['POST']) +@require_scope(Scope.RUNS_WRITE) +@validate_body(RunCreateRequestSchema) +def create_run(validated_data=None): + """Trigger a new test run for a commit + platform combination. + + CI worker pickup: the cron (run_cron.py) picks up any Test row that has + no 'completed'/'canceled' TestProgress, then runs the prebuilt GitHub + Actions artifact for that commit. We therefore reject up front any + commit+platform with no build artifact (see _ci_artifact_exists), so + the run isn't accepted only to fail asynchronously in the worker. + """ + commit_sha = validated_data['commit_sha'] + platform_str = validated_data['platform'] + branch = validated_data.get('branch', 'master') + repository = validated_data.get('repository') + pull_request = validated_data.get('pull_request') or 0 + regression_test_ids = validated_data.get('regression_test_ids') + + platform = TestPlatform.from_string(platform_str) + + # Main repo requires contributor+; forks allow any authenticated user. + from run import config + main_owner = config.get('GITHUB_OWNER', '') + main_repo = config.get('GITHUB_REPOSITORY', '') + main_repo_full = f'{main_owner}/{main_repo}' + # repository is a required field (RunCreateRequestSchema), so it is always + # present; a main-repo run passes the main repo's "owner/repo" explicitly. + target_repo = repository + + err = _validate_run_permissions(g.api_user, target_repo, main_repo_full) + if err: + return err + + # Reject commits with no CI build artifact — the worker runs prebuilt + # binaries, so such a run would be accepted but never execute. + if not _ci_artifact_exists(commit_sha, platform): + return make_error_response( + 'unprocessable', + f'No CI build artifact found for commit {commit_sha[:8]} on ' + f'{platform.value}. Ensure the build workflow has completed for ' + 'this commit before triggering a run.', + details={'commit_sha': commit_sha, 'platform': platform.value}, + http_status=422, + ) + + fork_url = f'https://github.com/{repository}.git' + + fork, err = _get_or_create_fork(fork_url) + if err: + return err + + # Validate regression test IDs against active tests only. + regression_test_ids, err = _validate_regression_test_ids( + regression_test_ids) + if err: + return err + + test_type = TestType.pull_request if pull_request else TestType.commit + + test = Test( + platform=platform, + test_type=test_type, + fork_id=fork.id, + branch=branch, + commit=commit_sha, + pr_nr=pull_request, + ) + g.db.add(test) + try: + g.db.flush() + except Exception: + g.db.rollback() + return make_error_response( + 'internal_error', + 'Failed to create run.', + http_status=500) + + for rt_id in regression_test_ids: + ct = CustomizedTest(test.id, rt_id) + g.db.add(ct) + try: + g.db.commit() + except Exception: + g.db.rollback() + return make_error_response( + 'internal_error', + 'Failed to finalize run.', + http_status=500) + + return single_response( + _serialize_run(test), + schema=RunSchema(), + http_status=202) + + +@mod_api.route('/runs/', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_path_id('run_id') +def get_run(run_id): + """Fetch a single run by ID.""" + 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) + + return single_response(_serialize_run(test), schema=RunSchema()) + + +def _run_regression_ids(test): + """Regression test IDs that belong to this run. + + Uses the customized selection when present; otherwise falls back to + every ACTIVE regression test, mirroring create_run's default. (The + model's get_customized_regressiontests() falls back to all tests + including inactive ones, which inflates total_samples/skipped_count + with tests the run could never execute.) + """ + if test.customized_tests: + return [ct.regression_id for ct in test.customized_tests] + return [rt.id for rt in + RegressionTest.query.filter_by(active=True).all()] + + +def _aggregate_run_statistics( + results, + files_by_result, + expected_outputs_by_rt): + pass_count = fail_count = skipped_count = missing_count = total_runtime = 0 + for result in results: + result_files = files_by_result.get(result.regression_test_id, []) + expected = expected_outputs_by_rt.get(result.regression_test_id) + status = derive_sample_status(result, result_files, expected) + + if status == 'pass': + pass_count += 1 + elif status == 'fail': + fail_count += 1 + elif status == 'missing_output': + missing_count += 1 + else: + skipped_count += 1 + + if result.runtime: + total_runtime += result.runtime + + return pass_count, fail_count, skipped_count, missing_count, total_runtime + + +@mod_api.route('/runs//summary', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_path_id('run_id') +def get_run_summary(run_id): + """ + Aggregate pass/fail/skip/missing/error counts from result rows. + + fail_count comes from TestResult rows, not from test.failed (which + only reflects cancellation status and is unreliable for this purpose). + """ + 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) + + results = TestResult.query.filter_by(test_id=run_id).all() + total_samples = len(_run_regression_ids(test)) + + # Preload TestResultFiles + + all_files = ( + TestResultFile.query.options( + joinedload(TestResultFile.regression_test_output) + .joinedload(RegressionTestOutput.multiple_files) + ) + .filter_by(test_id=run_id).all() if results else [] + ) + files_by_result = defaultdict(list) + for f in all_files: + files_by_result[f.regression_test_id].append(f) + + # Preload expected outputs + expected_outputs_by_rt = defaultdict(list) + if results: + all_expected = RegressionTestOutput.query.filter( + RegressionTestOutput.regression_id.in_([r.regression_test_id for r in results]) + ).all() + for rto in all_expected: + expected_outputs_by_rt[rto.regression_id].append(rto) + + pass_count, fail_count, skipped_count, missing_count, total_runtime = _aggregate_run_statistics( + results, files_by_result, expected_outputs_by_rt) + + # Reconcile skipped samples (those without any TestResult row) + if len(results) < total_samples: + skipped_count += (total_samples - len(results)) + + # Retrieve error_count from the error service + error_count = len( + derive_errors_for_run( + run_id, + expected_outputs_by_rt, + preloaded_results=results, + preloaded_files=all_files)) + + statuses, _ = batch_get_run_data([test]) + run_status = statuses.get(test.id, 'queued') + + return single_response({ + 'run_id': run_id, + 'status': run_status, + 'total_samples': total_samples, + 'pass_count': pass_count, + 'fail_count': fail_count, + 'skipped_count': skipped_count, + 'missing_output_count': missing_count, + 'error_count': error_count, + 'duration_ms': total_runtime if total_runtime > 0 else None, + }, schema=RunSummarySchema()) + + +@mod_api.route('/runs//progress', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_path_id('run_id') +@validate_offset_pagination() +def get_run_progress(run_id, limit=50, offset=0): + """ + Get the timeline of progress events for a run, paginated. + + Events come from TestProgress rows written by the CI worker. + """ + 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) + + query = TestProgress.query.filter_by(test_id=run_id) + + # Optional status filter. + status_filter = request.args.get('status') + if status_filter: + try: + status_enum = TestStatus.from_string(status_filter) + query = query.filter(TestProgress.status == status_enum) + except Exception: + return make_error_response( + 'validation_error', + f'Invalid status filter: {status_filter}.', + details={ + 'fields': { + 'status': 'Must be one of: queued, preparation, testing, completed, canceled, error.'}}, + http_status=400, + ) + + query = query.order_by(TestProgress.id.asc()) + total = query.count() + progress = query.offset(offset).limit(limit).all() + + events = [{ + 'timestamp': p.timestamp, + 'status': p.status.name, + 'message': p.message, + } for p in progress] + + schema = ProgressEventSchema() + return paginated_response(events, total, limit, offset, schema=schema) + + +@mod_api.route('/runs//config', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_path_id('run_id') +def get_run_config(run_id): + """Get the configuration that was used to launch this run.""" + 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_ids = _run_regression_ids(test) + + return single_response({ + 'run_id': run_id, + 'platform': test.platform.value, + 'branch': test.branch, + 'commit_sha': test.commit, + 'regression_test_ids': regression_ids, + }) + + +@mod_api.route('/runs//cancel', methods=['POST']) +@require_roles([Role.admin, Role.contributor, Role.tester]) +@require_scope(Scope.RUNS_WRITE) +@validate_path_id('run_id') +def cancel_run(run_id): + """Cancel a running or queued test. + + Idempotent — canceling something already finished returns 202 + with status=no_op. + + Note: In this shared CI environment, any user with 'runs:write' + (admin, contributor, tester) can cancel any run on the platform, + regardless of ownership. This is intentional. + """ + test = Test.query.with_for_update().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) + + status = derive_run_status(test) + if status in ('pass', 'fail', 'canceled', 'error'): + return single_response({ + 'run_id': run_id, + 'action': 'cancel', + 'status': 'no_op', + 'message': f'Run is already in terminal state: {status}', + }, http_status=202) + + user = g.api_user + reason = None + if request.is_json and request.get_json(silent=True): + reason = request.get_json(silent=True).get('reason') + if reason: + reason_str = str(reason).strip() + if len(reason_str) < 5: + return make_error_response( + 'validation_error', + 'Cancel reason must be at least 5 characters.', + details={'fields': {'reason': 'Minimum length is 5.'}}, + http_status=400, + ) + reason = reason_str[:255] + + cancel_msg = f'Canceled by {user.name} via API' if user else 'Canceled via API' + if reason: + cancel_msg = f'{cancel_msg}: {reason}' + + progress = TestProgress(run_id, TestStatus.canceled, cancel_msg) + g.db.add(progress) + g.db.commit() + + return single_response({ + 'run_id': run_id, + 'action': 'cancel', + 'status': 'accepted', + 'message': 'Run has been canceled.', + }, http_status=202) diff --git a/mod_api/routes/system.py b/mod_api/routes/system.py new file mode 100644 index 000000000..6a9cd7b92 --- /dev/null +++ b/mod_api/routes/system.py @@ -0,0 +1,204 @@ +""" +System, health, and queue routes. + +GET /system/health Health check (unauthenticated) +GET /system/queue Queue status — active + queued runs +""" + +import os +from datetime import datetime, timezone + +from flask import g, jsonify, request +from sqlalchemy import text + +from mod_api import mod_api +from mod_api.middleware.auth import require_scope +from mod_api.middleware.error_handler import make_error_response +from mod_api.middleware.validation import validate_offset_pagination +from mod_api.models.api_token import Scope +from mod_api.schemas.common import DATETIME_FORMAT +from mod_api.services.status import batch_get_run_data +from mod_api.utils import paginated_response +from mod_test.models import Test, TestPlatform, TestProgress, TestStatus + + +@mod_api.route('/system/health', methods=['GET']) +def system_health(): + """ + Public health check — no auth required. + + Returns 200 when things are ok or degraded, 503 when the system is down. + Monitoring services and load balancers can hit this freely. + """ + now = datetime.now(timezone.utc) + dependencies = [] + overall = 'ok' + + # Database connectivity. + try: + g.db.execute(text('SELECT 1')) + dependencies.append( + {'name': 'database', 'status': 'ok', 'message': None}) + except Exception: + dependencies.append({'name': 'database', + 'status': 'down', + 'message': 'Database connection failed.'}) + overall = 'down' + + # Local sample storage. + try: + from run import config + sample_repo = config.get('SAMPLE_REPOSITORY', '') + if os.path.isdir(sample_repo): + dependencies.append( + {'name': 'local_storage', 'status': 'ok', 'message': None}) + else: + dependencies.append({ + 'name': 'local_storage', + 'status': 'degraded', + 'message': 'Local storage check failed.', + }) + if overall == 'ok': + overall = 'degraded' + except Exception: + dependencies.append({'name': 'local_storage', 'status': 'down', + 'message': 'Local storage check failed.'}) + overall = 'down' + + # Google Cloud Storage. + try: + from run import storage_client_bucket + if storage_client_bucket: + dependencies.append( + {'name': 'gcs', 'status': 'ok', 'message': None}) + else: + dependencies.append({'name': 'gcs', + 'status': 'degraded', + 'message': 'GCS client not initialized.'}) + if overall == 'ok': + overall = 'degraded' + except Exception: + dependencies.append({'name': 'gcs', 'status': 'degraded', + 'message': 'GCS connectivity check failed.'}) + if overall == 'ok': + overall = 'degraded' + + http_status = 503 if overall == 'down' else 200 + response = jsonify({ + 'status': overall, + 'checked_at': now.strftime(DATETIME_FORMAT), + 'dependencies': dependencies, + }) + response.status_code = http_status + return response + + +def _apply_queue_filters( + base_query, + running_subq, + queue_depth, + running_count, + status_filter): + if status_filter == 'queued': + query = base_query.filter(~Test.id.in_( + g.db.query(running_subq.c.test_id))) + total = queue_depth + elif status_filter == 'running': + query = base_query.filter(Test.id.in_( + g.db.query(running_subq.c.test_id))) + total = running_count + elif status_filter: + return None, None, make_error_response( + 'validation_error', 'Invalid status. Must be queued or running.', http_status=400) + else: + query = base_query + total = queue_depth + running_count + return query, total, None + + +@mod_api.route('/system/queue', methods=['GET']) +@require_scope(Scope.SYSTEM_READ) +@validate_offset_pagination() +def get_queue(limit=50, offset=0): + """ + Get queue summary and list of runs. + + Note: The `position` field is only populated when `?status=queued` is + explicitly provided. Otherwise, it will be null for all items. + + Excludes anything that's already completed or canceled. Supports + ?platform and ?status filters. + """ + terminal_subq = g.db.query( + TestProgress.test_id + ).filter( + TestProgress.status.in_([TestStatus.completed, TestStatus.canceled]) + ).group_by(TestProgress.test_id).subquery() + + running_subq = g.db.query( + TestProgress.test_id + ).filter( + TestProgress.status.in_([TestStatus.preparation, TestStatus.testing]) + ).group_by(TestProgress.test_id).subquery() + + base_query = Test.query.filter( + ~Test.id.in_(g.db.query(terminal_subq.c.test_id)) + ) + + platform_filter = request.args.get('platform') + if platform_filter: + try: + plat = TestPlatform.from_string(platform_filter) + base_query = base_query.filter(Test.platform == plat) + except Exception: + return make_error_response( + 'validation_error', + 'Invalid platform.', + http_status=400) + + running_count = base_query.filter(Test.id.in_( + g.db.query(running_subq.c.test_id))).count() + queue_depth = base_query.filter(~Test.id.in_( + g.db.query(running_subq.c.test_id))).count() + + status_filter = request.args.get('status') + query, total, err = _apply_queue_filters( + base_query, running_subq, queue_depth, running_count, status_filter) + if err: + return err + + query = query.order_by(Test.id.asc()) + paged_tests = query.offset(offset).limit(limit).all() + + statuses, timestamps = batch_get_run_data(paged_tests) + + paged_jobs = [] + queued_index = offset + 1 if status_filter == 'queued' else None + + for test in paged_tests: + status = statuses.get(test.id, 'queued') + ts = timestamps.get(test.id, {}) + + pos = None + if status == 'queued' and queued_index is not None: + pos = queued_index + queued_index += 1 + + paged_jobs.append({ + 'run_id': test.id, + 'status': status, + 'platform': test.platform.value, + # Same 'Z' format the schemas use, so the API emits one + # datetime style everywhere (timestamps are UTC). + 'queued_at': ts.get('queued_at').strftime(DATETIME_FORMAT) if ts.get('queued_at') else None, + 'started_at': ts.get('started_at').strftime(DATETIME_FORMAT) if ts.get('started_at') else None, + 'position': pos, + }) + + return paginated_response( + paged_jobs, total, limit, offset, + extra_meta={ + 'queue_depth': queue_depth, + 'running_count': running_count, + } + ) diff --git a/mod_api/schemas/runs.py b/mod_api/schemas/runs.py new file mode 100644 index 000000000..a4ab5539c --- /dev/null +++ b/mod_api/schemas/runs.py @@ -0,0 +1,99 @@ +"""Schemas for runs, summaries, and progress events.""" + +from marshmallow import RAISE, Schema, fields, validate + +from mod_api.schemas.common import DATETIME_FORMAT + + +class ProgressEventSchema(Schema): + """A single progress event in a run's timeline.""" + + timestamp = fields.DateTime(required=True, format=DATETIME_FORMAT) + status = fields.String(required=True) + message = fields.String(required=True) + + +class RunSchema(Schema): + """Full run details.""" + + run_id = fields.Integer(required=True) + status = fields.String(required=True, validate=validate.OneOf([ + 'queued', 'running', 'pass', 'fail', 'canceled', 'incomplete', 'error' + ])) + platform = fields.String( + required=True, validate=validate.OneOf(['linux', 'windows'])) + test_type = fields.String(validate=validate.OneOf(['commit', 'pr'])) + repository = fields.String(required=True) + branch = fields.String(allow_none=True) + commit_sha = fields.String(required=True) + pr_number = fields.Integer(allow_none=True, load_default=None) + created_at = fields.DateTime(allow_none=True, format=DATETIME_FORMAT) + queued_at = fields.DateTime(allow_none=True, format=DATETIME_FORMAT) + started_at = fields.DateTime(allow_none=True, format=DATETIME_FORMAT) + completed_at = fields.DateTime(allow_none=True, format=DATETIME_FORMAT) + github_link = fields.String(allow_none=True) + + +class RunSummarySchema(Schema): + """Pass/fail/skip aggregate counts for a run.""" + + run_id = fields.Integer(required=True) + status = fields.String(required=True) + total_samples = fields.Integer(required=True) + pass_count = fields.Integer(required=True) + fail_count = fields.Integer(required=True) + skipped_count = fields.Integer(required=True) + missing_output_count = fields.Integer(required=True) + error_count = fields.Integer(load_default=0) + duration_ms = fields.Integer(allow_none=True) + + +class RunCreateRequestSchema(Schema): + """POST /runs request body.""" + + commit_sha = fields.String( + required=True, + validate=validate.Regexp( + r'^[a-fA-F0-9]{40}$', + error='commit_sha must be a 40-character hex string.', + ), + ) + platform = fields.String( + required=True, + validate=validate.OneOf(['linux', 'windows']), + ) + branch = fields.String( + load_default='master', + validate=[ + validate.Length(max=100), + validate.Regexp( + r'^[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*$', + error='branch must match ^[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*$', + ), + ], + ) + repository = fields.String( + required=True, + validate=[ + validate.Length(max=100), + validate.Regexp( + r'^[a-zA-Z0-9_.\-]+/[a-zA-Z0-9_.\-]+$', + error='repository must match owner/repo format.', + ), + ], + ) + pull_request = fields.Integer( + load_default=None, + allow_none=True, + validate=validate.Range(min=1, max=2147483647), + ) + regression_test_ids = fields.List( + fields.Integer(validate=validate.Range(min=1, max=2147483647)), + load_default=None, + validate=validate.Length(max=500), + ) + + class Meta: + """Reject unknown fields.""" + + unknown = RAISE diff --git a/mod_api/services/error_service.py b/mod_api/services/error_service.py new file mode 100644 index 000000000..f8bcd85b1 --- /dev/null +++ b/mod_api/services/error_service.py @@ -0,0 +1,307 @@ +""" +Error derivation from TestResult and TestResultFile rows. + +Walks result data and produces structured ErrorItem dicts. There's no +dedicated error table — errors are inferred from: + exit_code_mismatch → exit code != expected + diff_mismatch → got != null and not in multiple correct files + missing_output → dummy (-1,-1,-1,'','error') row present +""" + +import logging +from collections import defaultdict +from typing import Any, Dict, List + +from sqlalchemy.orm import joinedload + +from mod_api.services.status import is_dummy_row +from mod_regression.models import RegressionTestOutput +from mod_test.models import (TestProgress, TestResult, TestResultFile, + TestStatus) + +_SEVERITY_ORDER = ('info', 'warning', 'error', 'critical') + + +def _is_output_acceptable(rf: TestResultFile) -> bool: + if not rf.regression_test_output: + return False + for multi in rf.regression_test_output.multiple_files: + if multi.file_hashes == rf.got: + return True + return False + + +def _check_exit_code_errors(result, test_id, occurred_at): + if result.exit_code != result.expected_rc: + return [{ + 'error_id': f'err_{test_id}_{result.regression_test_id}_rc', + 'run_id': test_id, + 'sample_id': _get_sample_id(result), + 'regression_id': result.regression_test_id, + 'type': 'exit_code_mismatch', + 'severity': 'error', + 'message': ( + f'Exit code {result.exit_code} != expected {result.expected_rc} ' + f'for regression test {result.regression_test_id}' + ), + 'occurred_at': occurred_at, + }] + return [] + + +def _check_missing_output_errors(result, result_files, test_id, occurred_at, expected_outputs): + errors = [] + actual_output_ids = {rf.regression_test_output_id for rf in result_files} + if expected_outputs is not None: + for rto in expected_outputs: + if not rto.ignore and rto.id not in actual_output_ids: + errors.append({ + 'error_id': f'err_{test_id}_{result.regression_test_id}_missing_{rto.id}', + 'run_id': test_id, + 'sample_id': _get_sample_id(result), + 'regression_id': result.regression_test_id, + 'type': 'missing_output', + 'severity': 'error', + 'message': ( + f'Regression test {result.regression_test_id} ' + f'produced no output for expected file {rto.id}' + ), + 'occurred_at': occurred_at, + }) + else: + for rf in result_files: + if is_dummy_row(rf): + errors.append({ + 'error_id': f'err_{test_id}_{result.regression_test_id}_missing', + 'run_id': test_id, + 'sample_id': _get_sample_id(result), + 'regression_id': result.regression_test_id, + 'type': 'missing_output', + 'severity': 'error', + 'message': ( + f'Regression test {result.regression_test_id} ' + f'produced no output when output was expected' + ), + 'occurred_at': occurred_at, + }) + return errors + + +def _check_diff_mismatch_errors(result, result_files, test_id, occurred_at): + errors = [] + for rf in result_files: + if is_dummy_row(rf): + continue + if rf.got is not None and not _is_output_acceptable(rf): + errors.append({ + 'error_id': f'err_{test_id}_{result.regression_test_id}_{rf.regression_test_output_id}', + 'run_id': test_id, + 'sample_id': _get_sample_id(result), + 'regression_id': result.regression_test_id, + 'type': 'diff_mismatch', + 'severity': 'warning', + 'message': ( + f'Output differs from expected for regression test ' + f'{result.regression_test_id}, output {rf.regression_test_output_id}' + ), + 'occurred_at': occurred_at, + }) + return errors + + +def _evaluate_test_result( + result, + result_files, + test_id, + occurred_at, + expected_outputs=None): + errors = [] + errors.extend(_check_exit_code_errors(result, test_id, occurred_at)) + errors.extend(_check_missing_output_errors(result, result_files, test_id, occurred_at, expected_outputs)) + errors.extend(_check_diff_mismatch_errors(result, result_files, test_id, occurred_at)) + return errors + + +def _group_result_files(test_id, results, preloaded_files=None): + """Map regression_test_id -> [TestResultFile], loading if not preloaded.""" + if preloaded_files is not None: + all_files = preloaded_files + elif results: + all_files = TestResultFile.query.options( + joinedload(TestResultFile.regression_test_output) + .joinedload(RegressionTestOutput.multiple_files) + ).filter_by(test_id=test_id).all() + else: + all_files = [] + files_by_result = defaultdict(list) + for f in all_files: + files_by_result[f.regression_test_id].append(f) + return files_by_result + + +def _load_expected_outputs(results): + """Map regression_test_id -> [RegressionTestOutput] for the given results. + + Missing-output detection must use the same RegressionTestOutput + comparison as /runs/{id}/summary — /errors and /summary have to agree. + id > 0 excludes the -1 sentinel some fixtures create to satisfy the + dummy TestResultFile row's foreign key; it is not a real expectation. + """ + if not results: + return {} + rt_ids = {r.regression_test_id for r in results} + expected_by_rt = defaultdict(list) + for rto in RegressionTestOutput.query.filter( + RegressionTestOutput.regression_id.in_(rt_ids), + RegressionTestOutput.id > 0).all(): + expected_by_rt[rto.regression_id].append(rto) + return expected_by_rt + + +def derive_errors_for_run(test_id: int, + expected_outputs_by_rt: Dict[int, + List[Any]] = None, + preloaded_results=None, + preloaded_files=None) -> List[Dict[str, + Any]]: + """Walk result rows and emit one ErrorItem per detected failure.""" + progress = TestProgress.query.filter_by(test_id=test_id).order_by( + TestProgress.timestamp.desc()).first() + occurred_at = progress.timestamp.isoformat( + ) if progress and progress.timestamp else None + + if preloaded_results is not None: + results = preloaded_results + else: + results = TestResult.query.filter_by(test_id=test_id).all() + + files_by_result = _group_result_files(test_id, results, preloaded_files) + + if expected_outputs_by_rt is None: + expected_outputs_by_rt = _load_expected_outputs(results) + + errors = [] + for result in results: + result_files = files_by_result.get(result.regression_test_id, []) + expected_outputs = expected_outputs_by_rt.get( + result.regression_test_id) if expected_outputs_by_rt else None + errors.extend(_evaluate_test_result( + result, result_files, test_id, occurred_at, expected_outputs)) + + return errors + + +def _aggregate_error_into_bucket(err, bucket): + bucket['count'] += 1 + + # Escalate severity to the worst we've seen. + try: + curr_idx = _SEVERITY_ORDER.index(bucket['severity']) + new_idx = _SEVERITY_ORDER.index(err['severity']) + if new_idx > curr_idx: + bucket['severity'] = err['severity'] + except ValueError: + # Fallback if unknown severity + if err['severity'] == 'error': + bucket['severity'] = 'error' + + err_time = err.get('occurred_at') + if err_time: + if bucket['first_seen_at'] is None or err_time < bucket['first_seen_at']: + bucket['first_seen_at'] = err_time + if bucket['last_seen_at'] is None or err_time > bucket['last_seen_at']: + bucket['last_seen_at'] = err_time + + sid = err.get('sample_id') + if sid and sid not in bucket['sample_ids'] and len( + bucket['sample_ids']) < 1000: + bucket['sample_ids'].append(sid) + + +def derive_error_summary( + test_id: int, group_by: str = 'type') -> List[Dict[str, Any]]: + """Group errors by the given key and return bucket counts.""" + errors = derive_errors_for_run(test_id) + buckets: Dict[str, Dict[str, Any]] = {} + + for err in errors: + key = str(err.get(group_by, 'unknown')) + + if key not in buckets: + buckets[key] = { + 'key': key, + 'group_by': group_by, + 'count': 0, + 'severity': err['severity'], + 'sample_ids': [], + 'first_seen_at': None, + 'last_seen_at': None, + } + + _aggregate_error_into_bucket(err, buckets[key]) + + return list(buckets.values()) + + +def derive_infrastructure_errors(test_id: int) -> List[Dict[str, Any]]: + """ + Best-effort infra error extraction from TestProgress messages. + + There's no structured error protocol from the CI worker yet, so we + do keyword matching against progress messages to guess the failure type. + """ + errors = [] + progress_rows = TestProgress.query.filter_by( + test_id=test_id, + status=TestStatus.canceled, + ).all() + + for p in progress_rows: + message = p.message or '' + # User-initiated cancellations (cancel_run writes "... via API") are not + # infrastructure failures, so they must not be reported here. + if 'via API' in message: + continue + error_type = _classify_infra_error(message.lower()) + errors.append({ + 'error_id': f'infra_{test_id}_{p.id}', + 'run_id': test_id, + 'sample_id': None, + 'regression_id': None, + 'type': error_type, + 'severity': 'critical', + 'message': p.message or 'Unknown infrastructure error', + 'location': None, + 'occurred_at': p.timestamp.isoformat() if p.timestamp else None, + }) + + return errors + + +def _classify_infra_error(message_lower: str) -> str: + """Guess the infra error type from progress message keywords.""" + if any(w in message_lower for w in ['provisioning', 'vm ', 'instance']): + return 'vm_provisioning' + if any(w in message_lower for w in ['checkout', 'git clone', 'fetch']): + return 'checkout' + if any(w in message_lower for w in ['merge', 'conflict']): + return 'merge' + if any(w in message_lower for w in ['build', 'compile', 'make']): + return 'build' + if any(w in message_lower for w in ['worker', 'timeout', 'connection']): + return 'worker' + if any(w in message_lower for w in ['storage', 'disk', 'gcs']): + return 'storage' + return 'worker' + + +def _get_sample_id(result: TestResult): + """Pull sample_id through the RegressionTest relationship, if available.""" + try: + if result.regression_test and result.regression_test.sample_id: + return result.regression_test.sample_id + except Exception: + logging.getLogger(__name__).exception( + f"Failed to fetch sample_id for TestResult {result.test_id}_{result.regression_test_id}" + ) + return None diff --git a/mod_api/services/storage.py b/mod_api/services/storage.py new file mode 100644 index 000000000..f9d56533e --- /dev/null +++ b/mod_api/services/storage.py @@ -0,0 +1,77 @@ +""" +Storage helpers for resolving artifact locations. + +Artifacts can live in local SAMPLE_REPOSITORY, GCS, or both. When both +exist, GCS is preferred and a signed URL is returned. When only local +exists, storage_status is 'degraded'. When neither exists, it's 'missing'. +""" + +import logging +import os +from datetime import timedelta +from typing import Optional, Tuple + +logger = logging.getLogger(__name__) + + +def resolve_artifact(relative_path: str) -> Tuple[Optional[str], str]: + """ + Look for an artifact in local storage and GCS. + + Returns (download_url_or_None, storage_status). + """ + from run import config, storage_client_bucket + + sample_repo = config.get('SAMPLE_REPOSITORY', '') + local_path = os.path.join(sample_repo, relative_path) + # Prevent path traversal: resolved path must stay within sample_repo + real_base = os.path.realpath(sample_repo) + real_path = os.path.realpath(local_path) + if not (real_path.startswith(real_base + os.sep) or real_path == real_base): + return None, 'missing' + local_exists = os.path.isfile(local_path) + + gcs_url = None + if storage_client_bucket: + try: + blob = storage_client_bucket.blob(relative_path) + if blob.exists(): + gcs_url = blob.generate_signed_url( + version='v4', + # int() guards against a string value in config, which + # would otherwise raise inside timedelta and be swallowed + # by the except below as a silent 'degraded'. + expiration=timedelta(minutes=int(config.get( + 'GCS_SIGNED_URL_EXPIRY_LIMIT', 60))), + method='GET', + ) + except Exception as e: + logger.warning(f"Failed to generate GCS signed URL for {relative_path}: {e}") + gcs_url = None + + if local_exists and gcs_url: + return gcs_url, 'ok' + elif gcs_url: + return gcs_url, 'degraded' + elif local_exists: + return None, 'degraded' + else: + return None, 'missing' + + +def get_log_file_path(run_id: int) -> Optional[str]: + """Return the absolute path to a run's build log, or None if it doesn't exist.""" + from run import config + + sample_repo = config.get('SAMPLE_REPOSITORY', '') + log_path = os.path.join(sample_repo, 'LogFiles', f'{run_id}.txt') + + if os.path.isfile(log_path): + return log_path + return None + + +def get_test_results_base_path() -> str: + """Return the base directory where TestResults files are stored.""" + from run import config + return os.path.join(config.get('SAMPLE_REPOSITORY', ''), 'TestResults') diff --git a/mod_auth/controllers.py b/mod_auth/controllers.py index a476b9afc..a1f773774 100755 --- a/mod_auth/controllers.py +++ b/mod_auth/controllers.py @@ -165,26 +165,30 @@ def github_redirect(): return f'https://github.com/login/oauth/authorize?client_id={github_client_id}&scope=public_repo' -def fetch_username_from_token() -> Any: +def fetch_username_from_token(user=None) -> Any: """ Get username from the GitHub token. + :param user: Optional user model to prevent redundant queries :return: username :rtype: str """ - import json - user = User.query.filter(User.id == g.user.id).first() + if user is None: + user = User.query.filter(User.id == g.user.id).first() + if user.github_token is None: return None url = 'https://api.github.com/user' session = requests.Session() session.auth = (user.email, user.github_token) try: - response = session.get(url) + response = session.get(url, timeout=(3.05, 10)) data = response.json() - return data['login'] + return data.get('login') except Exception as e: - g.log.error('Failed to fetch the user token') + import logging + log = getattr(g, 'log', logging.getLogger(__name__)) + log.error('Failed to fetch the user token') return None @@ -211,6 +215,12 @@ def github_callback(): if 'access_token' in response: user = User.query.filter(User.id == g.user.id).first() user.github_token = response['access_token'] + + # Fetch and store github_login + github_login = fetch_username_from_token(user) + if github_login: + user.github_login = github_login + g.db.commit() else: g.log.error("GitHub didn't return an access token") diff --git a/tests/api/test_middleware_auth.py b/tests/api/test_middleware_auth.py new file mode 100644 index 000000000..983600ba1 --- /dev/null +++ b/tests/api/test_middleware_auth.py @@ -0,0 +1,167 @@ +from flask import g + +from mod_api.models.api_token import DEFAULT_SCOPES, ApiToken +from mod_auth.models import Role, User +from tests.api.base import ApiTestCase + + +class TestMiddlewareAuth(ApiTestCase): + def setUp(self): + super().setUp() + user = User('testuser1', Role.user, 'testuser1@local.com', + User.generate_hash('user123')) + admin = User('testadmin1', Role.admin, + 'testadmin1@local.com', User.generate_hash('admin123')) + g.db.add_all([user, admin]) + g.db.commit() + self.user = user + self.admin = admin + + def generate_db_token(self, user, scopes=None, expires_in_days=7): + plaintext = ApiToken.generate_token() + token = ApiToken( + user_id=user.id, + token_name='test_token_' + ApiTestCase.create_random_string(8), + token_hash=ApiToken.hash_token(plaintext), + token_prefix=ApiToken.extract_prefix(plaintext), + scopes=scopes or DEFAULT_SCOPES, + expires_in_days=expires_in_days + ) + g.db.add(token) + g.db.commit() + return plaintext, token + + def test_missing_auth_header(self): + res = self.client.get('/api/v1/system/queue') + self.assertEqual(res.status_code, 401) + self.assertEqual(res.json['code'], 'unauthorized') + + def test_invalid_auth_header_format(self): + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': 'InvalidFormat'}) + self.assertEqual(res.status_code, 401) + + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': 'Bearer '}) + self.assertEqual(res.status_code, 401) + + def test_invalid_token_prefix(self): + res = self.client.get( + '/api/v1/system/queue', headers={'Authorization': 'Bearer invalid_prefix_token'}) + self.assertEqual(res.status_code, 401) + + def test_token_not_found(self): + res = self.client.get( + '/api/v1/system/queue', headers={'Authorization': 'Bearer spci_faketoken1234567890'}) + self.assertEqual(res.status_code, 401) + + def test_wrong_hash(self): + plaintext, token = self.generate_db_token(self.user) + wrong_token = token.token_prefix + 'A' * \ + (len(plaintext) - len(token.token_prefix)) + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': f'Bearer {wrong_token}'}) + self.assertEqual(res.status_code, 401) + + def test_revoked_token(self): + plaintext, token = self.generate_db_token(self.user) + token.revoke() + g.db.commit() + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': f'Bearer {plaintext}'}) + self.assertEqual(res.status_code, 401) + + def test_expired_token(self): + plaintext, _ = self.generate_db_token(self.user, expires_in_days=-1) + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': f'Bearer {plaintext}'}) + self.assertEqual(res.status_code, 401) + + def test_valid_token_missing_scope(self): + # /api/v1/system/queue requires 'system:read' + plaintext, _ = self.generate_db_token(self.user, scopes=['runs:read']) + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': f'Bearer {plaintext}'}) + self.assertEqual(res.status_code, 403) + self.assertIn('code', res.json) + self.assertEqual(res.json['code'], 'forbidden') + self.assertIn('missing_scopes', res.json['details']) + + def test_valid_token_with_scope(self): + plaintext, _ = self.generate_db_token(self.user, scopes=['system:read']) + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': f'Bearer {plaintext}'}) + self.assertEqual(res.status_code, 200) + + def test_role_decorator_missing_role(self): + # GET /api/v1/auth/tokens requires 'tokens:manage' and roles ['admin', 'contributor', 'tester'] + plaintext, _ = self.generate_db_token( + self.user, scopes=['tokens:manage']) # role is user + res = self.client.get('/api/v1/auth/tokens', + headers={'Authorization': f'Bearer {plaintext}'}) + self.assertEqual(res.status_code, 403) + self.assertEqual(res.json['code'], 'forbidden') + + def test_role_decorator_with_role(self): + plaintext, _ = self.generate_db_token( + self.admin, scopes=['tokens:manage']) # role is admin + res = self.client.get('/api/v1/auth/tokens', + headers={'Authorization': f'Bearer {plaintext}'}) + self.assertEqual(res.status_code, 200) + + def test_scope_boundary_write_endpoints_fail_on_read_only_scopes(self): + plaintext, _ = self.generate_db_token( + self.user, scopes=['runs:read', 'results:read']) + + # 1. POST /runs + res = self.client.post( + '/api/v1/runs', headers={'Authorization': f'Bearer {plaintext}'}) + self.assertEqual(res.status_code, 403) + self.assertEqual(res.json['code'], 'forbidden') + + # 2. POST /runs/1/cancel + res = self.client.post('/api/v1/runs/1/cancel', + 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']) + + # Force same prefix, must start with spci_ and be 16 chars long for extract_prefix + prefix = 'spci_abc12345678' + token1.token_prefix = prefix + token2.token_prefix = prefix + g.db.commit() + + # Modify plaintexts to have the same prefix + submitted1 = prefix + plaintext1[len(prefix):] + submitted2 = prefix + plaintext2[len(prefix):] + + token1.token_hash = ApiToken.hash_token(submitted1) + token2.token_hash = ApiToken.hash_token(submitted2) + g.db.commit() + + # It should correctly match token2 and ignore token1 + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': f'Bearer {submitted2}'}) + self.assertEqual(res.status_code, 200) + + # Invalid token with same prefix + submitted3 = prefix + 'A' * 32 + res3 = self.client.get( + '/api/v1/system/queue', headers={'Authorization': f'Bearer {submitted3}'}) + self.assertEqual(res3.status_code, 401) + + def test_auth_sets_g_api_user_and_token(self): + plaintext, token = self.generate_db_token(self.user, scopes=['system:read']) + expected_user_id = self.user.id + expected_token_id = token.id + with self.app.test_request_context('/api/v1/system/queue', headers={'Authorization': f'Bearer {plaintext}'}): + # This triggers all before_request handlers, including authenticate_request + resp = self.app.preprocess_request() + # If rate limit isn't cleared, it might return 429, but it is cleared in setUp + self.assertIsNone(resp) + self.assertEqual(g.api_user.id, expected_user_id) + self.assertEqual(g.api_token.id, expected_token_id) diff --git a/tests/api/test_routes_runs.py b/tests/api/test_routes_runs.py new file mode 100644 index 000000000..78843c06a --- /dev/null +++ b/tests/api/test_routes_runs.py @@ -0,0 +1,407 @@ +import json +from unittest.mock import patch + +from flask import g + +from mod_test.models import (Fork, Test, TestPlatform, TestProgress, + TestResult, TestResultFile, TestStatus, TestType) +from tests.api.base import ApiTestCase + + +class TestRoutesRuns(ApiTestCase): + def setUp(self): + super().setUp() + self.setup_run_data('runs') + self.progress = TestProgress( + self.test_id, TestStatus.preparation, "Queued") + g.db.add(self.progress) + g.db.commit() + patcher = patch.dict( + 'mod_api.middleware.rate_limit._rate_limit_store', {}, clear=True) + patcher.start() + self.addCleanup(patcher.stop) + + # create_run checks for a CI build artifact via GitHub; stub it as + # present by default so tests don't make network calls. The + # no-artifact path is covered explicitly in + # test_create_run_no_artifact_rejected. + artifact_patcher = patch( + 'mod_api.routes.runs._ci_artifact_exists', return_value=True) + artifact_patcher.start() + self.addCleanup(artifact_patcher.stop) + + def test_list_runs(self): + token = self.get_token('runs_user@local.com', + 'userpass123', 't1', scopes=['runs:read']) + res = self.client.get( + '/api/v1/runs', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + # BaseTestCase.setUp creates 2 Test objects; this setUp creates 1 more = 3 total + self.assertEqual(len(res.json['data']), 3) + self.assertTrue( + any(r['run_id'] == self.test_id for r in res.json['data'])) + + def test_list_runs_filters(self): + token = self.get_token('runs_user@local.com', + 'userpass123', 't2', scopes=['runs:read']) + # Invalid platform + res = self.client.get('/api/v1/runs?platform=invalid', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 400) + + # Valid platform + res = self.client.get('/api/v1/runs?platform=linux', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 3) + + # Invalid repository + res = self.client.get('/api/v1/runs?repository=invalid_repo', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 400) + + def test_list_runs_status_filter(self): + # We already have a TestProgress 'preparation' from setUp. + # Add a 'testing' one to make the run have 'running' / 'testing' status? + # Wait, the frontend query asks for 'testing'. The API uses 'running' or 'testing' in some places. + # Let's insert a TestStatus.testing progress to make the + # derive_run_status be 'running' + prog2 = TestProgress(self.test_id, TestStatus.testing, "Testing") + g.db.add(prog2) + g.db.commit() + + token = self.get_token('runs_user@local.com', + 'userpass123', 't3', scopes=['runs:read']) + res = self.client.get('/api/v1/runs?status=running', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 1) + + def test_list_runs_status_queued(self): + # A run with no TestProgress rows is 'queued'. This guards the + # status=queued filter, which must emit SQL `IS NULL` + # (TestProgress.id.is_(None)) rather than a Python identity check. + queued_test = Test(TestPlatform.linux, TestType.commit, + self.fork.id, 'master', 'queued_commit') + g.db.add(queued_test) + g.db.commit() + queued_id = queued_test.id # capture before the request detaches it + + token = self.get_token('runs_user@local.com', + 'userpass123', 'tq', scopes=['runs:read']) + res = self.client.get('/api/v1/runs?status=queued', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + run_ids = [r['run_id'] for r in res.json['data']] + # The new run (no progress) is queued; the setUp run (has progress) is not. + self.assertIn(queued_id, run_ids) + self.assertNotIn(self.test_id, run_ids) + + @patch('mod_api.routes.runs._ci_artifact_exists', return_value=False) + @patch('run.config') + def test_create_run_no_artifact_rejected(self, mock_config, _mock_artifact): + # When no CI build artifact exists for the commit+platform, the run + # cannot execute, so create_run must reject it with 422 rather than + # accepting a run that would fail silently in the worker. + mock_config.get.side_effect = lambda k, d='': 'testowner' if k == 'GITHUB_OWNER' else 'testrepo' + + token = self.get_token('runs_admin@local.com', + 'adminpass123', 'tna', scopes=['runs:write']) + payload = { + 'commit_sha': 'a' * 40, + 'platform': 'linux', + 'repository': 'testowner/testrepo', + 'regression_test_ids': [], + } + res = self.client.post( + '/api/v1/runs', + data=json.dumps(payload), + content_type='application/json', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 422) + self.assertEqual(res.json['code'], 'unprocessable') + + @patch('run.config') + def test_create_run(self, mock_config): + mock_config.get.side_effect = lambda k, d='': 'testowner' if k == 'GITHUB_OWNER' else 'testrepo' + + token = self.get_token('runs_admin@local.com', + 'adminpass123', 't4', scopes=['runs:write']) + payload = { + 'commit_sha': 'a' * 40, + 'platform': 'windows', + 'repository': 'testowner/testrepo', + 'regression_test_ids': [] + } + res = self.client.post( + '/api/v1/runs', + data=json.dumps(payload), + content_type='application/json', + headers={ + 'Authorization': f'Bearer {token}'}) + # Empty regression_test_ids gives 400 validation error + self.assertEqual(res.status_code, 400) + + # Test omitting regression_test_ids completely (it fetches active) + payload.pop('regression_test_ids') + res = self.client.post( + '/api/v1/runs', + data=json.dumps(payload), + content_type='application/json', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 202) + self.assertIn('run_id', res.json) + + def test_get_run(self): + token = self.get_token('runs_user@local.com', + 'userpass123', 't5', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['run_id'], self.test_id) + + def test_get_run_summary(self): + token = self.get_token('runs_user@local.com', + 'userpass123', 't6', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/summary', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['run_id'], self.test_id) + self.assertIn('total_samples', res.json) + + def test_get_run_progress(self): + token = self.get_token('runs_user@local.com', + 'userpass123', 't7', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/progress', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 1) + self.assertEqual(res.json['data'][0]['status'], 'preparation') + + def test_get_run_config(self): + token = self.get_token('runs_user@local.com', + 'userpass123', 't8', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/config', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['platform'], 'linux') + + def test_cancel_run(self): + token = self.get_token('runs_admin@local.com', + 'adminpass123', 't9', scopes=['runs:write']) + res = self.client.post( + f'/api/v1/runs/{self.test_id}/cancel', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 202) + self.assertEqual(res.json['status'], 'accepted') + + # Verify db change + progs = TestProgress.query.filter_by(test_id=self.test_id).all() + self.assertEqual(progs[-1].status, TestStatus.canceled) + + def test_cancel_run_idempotency(self): + token = self.get_token('runs_admin@local.com', + 'adminpass123', 't10', scopes=['runs:write']) + # First cancel + res = self.client.post( + f'/api/v1/runs/{self.test_id}/cancel', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 202) + + # Second cancel should still be 202 + res2 = self.client.post( + f'/api/v1/runs/{self.test_id}/cancel', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res2.status_code, 202) + self.assertEqual(res2.json['status'], 'no_op') + + @patch('run.config') + def test_create_run_inactive_regression_test(self, mock_config): + mock_config.get.side_effect = lambda k, d='': 'testowner' if k == 'GITHUB_OWNER' else 'testrepo' + + # Make a regression test inactive + from mod_regression.models import (Category, InputType, OutputType, + RegressionTest) + cat = Category('testcat', 'desc') + g.db.add(cat) + g.db.commit() + reg_test = RegressionTest( + 1, 'command', InputType.file, OutputType.file, cat.id, 0) + reg_test.active = False + g.db.add(reg_test) + g.db.flush() + reg_test_id = reg_test.id + g.db.commit() + + token = self.get_token('runs_admin@local.com', + 'adminpass123', 't11', scopes=['runs:write']) + payload = { + 'commit_sha': 'a' * 40, + 'platform': 'windows', + 'repository': 'testowner/testrepo', + 'regression_test_ids': [reg_test_id] + } + res = self.client.post( + '/api/v1/runs', + data=json.dumps(payload), + content_type='application/json', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 422) + self.assertIn('inactive', res.json['message']) + + def test_create_run_fork_owner_can_trigger(self): + # Verify that a user who owns a fork can trigger a run on it + self.user.github_login = 'userfork' + g.db.add(self.user) + g.db.commit() + + # Trigger run on a fork repo using contributor user + token = self.get_token('runs_user@local.com', + 'userpass123', 't12', scopes=['runs:write']) + payload = { + 'commit_sha': 'b' * 40, + 'platform': 'windows', + 'repository': 'userfork/testrepo' + } + res = self.client.post( + '/api/v1/runs', + data=json.dumps(payload), + content_type='application/json', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 202) + + def test_run_summary_fail_count_ignores_test_failed_flag(self): + # Ignore expected outputs so missing-output doesn't trigger first + from mod_regression.models import RegressionTestOutput + outputs = RegressionTestOutput.query.filter_by(regression_id=1).all() + for o in outputs: + o.ignore = True + g.db.add(o) + + # set up test result with exit code mismatch (which counts as fail) + tr = TestResult(self.test_id, 1, 100, 1, 0) + g.db.add(tr) + g.db.commit() + + token = self.get_token('runs_user@local.com', + 'userpass123', 't13', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/summary', headers={'Authorization': f'Bearer {token}'}) + + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['fail_count'], 1) + self.assertEqual(res.json['pass_count'], 0) + + def test_missing_output_not_double_counted_in_fail(self): + # Insert a dummy RegressionTestOutput with id = -1 to satisfy foreign + # key constraints + from mod_regression.models import RegressionTestOutput + dummy_out = RegressionTestOutput(1, '', '', '') + dummy_out.id = -1 + g.db.add(dummy_out) + g.db.commit() + + # exit code mismatch (would be fail) + tr = TestResult(self.test_id, 1, 100, 1, 0) + # but dummy row takes priority -> missing_output + rf = TestResultFile(self.test_id, 1, -1, '', 'error') + g.db.add_all([tr, rf]) + g.db.commit() + + token = self.get_token('runs_user@local.com', + 'userpass123', 't14', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/summary', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['missing_output_count'], 1) + self.assertEqual(res.json['fail_count'], 0) + + def test_cancel_run_reason_too_short(self): + token = self.get_token('runs_admin@local.com', + 'adminpass123', 't15', scopes=['runs:write']) + res = self.client.post(f'/api/v1/runs/{self.test_id}/cancel', + data=json.dumps({'reason': 'no'}), + content_type='application/json', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 400) + self.assertEqual(res.json['code'], 'validation_error') + + def test_create_run_rejects_extra_fields(self): + token = self.get_token('runs_admin@local.com', + 'adminpass123', 't17', scopes=['runs:write']) + payload = { + 'commit_sha': 'a' * 40, + 'platform': 'linux', + 'repository': 'testowner/testrepo', + 'unexpected_field': 'evil_val' + } + res = self.client.post( + '/api/v1/runs', + data=json.dumps(payload), + content_type='application/json', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 400) + self.assertEqual(res.json['code'], 'validation_error') + + def test_create_run_invalid_commit_sha_rejected(self): + token = self.get_token('runs_admin@local.com', + 'adminpass123', 't18', scopes=['runs:write']) + payload = { + 'commit_sha': 'shortsha', + 'platform': 'linux', + 'repository': 'testowner/testrepo' + } + res = self.client.post( + '/api/v1/runs', + data=json.dumps(payload), + content_type='application/json', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 400) + self.assertEqual(res.json['code'], 'validation_error') + + def test_get_run_nonexistent_resource_404(self): + token = self.get_token('runs_user@local.com', + 'userpass123', 't19', scopes=['runs:read']) + res = self.client.get('/api/v1/runs/999999', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 404) + self.assertEqual(res.json['code'], 'not_found') + + def test_create_run_non_admin_forbidden(self): + token = self.get_token( + 'runs_user@local.com', + 'userpass123', + 't_non_admin', + scopes=['runs:write']) + payload = { + 'commit_sha': 'a' * 40, + 'platform': 'windows', + 'repository': 'testowner/testrepo' + } + res = self.client.post( + '/api/v1/runs', + data=json.dumps(payload), + content_type='application/json', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 403) + + def test_list_runs_pagination(self): + # BaseTestCase.setUp creates 2 Test objects; this setUp creates 1 more = 3 total + token = self.get_token('runs_user@local.com', + 'userpass123', 't_pag', scopes=['runs:read']) + # Fetch first page with limit=2 + res1 = self.client.get('/api/v1/runs?limit=2', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res1.status_code, 200) + self.assertEqual(len(res1.json['data']), 2) + + # Fetch second page with offset=2 + res2 = self.client.get( + '/api/v1/runs?limit=2&offset=2', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res2.status_code, 200) + self.assertEqual(len(res2.json['data']), 1) diff --git a/tests/api/test_routes_system.py b/tests/api/test_routes_system.py new file mode 100644 index 000000000..bd71310de --- /dev/null +++ b/tests/api/test_routes_system.py @@ -0,0 +1,101 @@ +import json +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_test.models import Fork, Test, TestPlatform, TestType +from tests.api.base import ApiTestCase + + +class TestRoutesSystem(ApiTestCase): + def setUp(self): + super().setUp() + + # Create users + admin2 = User('admin2', Role.admin, 'admin2@local.com', + User.generate_hash('adminpass123')) + user2 = User('user2', Role.user, 'user2@local.com', + User.generate_hash('userpass123')) + g.db.add_all([admin2, user2]) + g.db.commit() + + # Create a test run + fork = Fork('https://github.com/test/test.git') + g.db.add(fork) + g.db.commit() + + self.test_obj = Test(TestPlatform.linux, + TestType.commit, fork.id, 'master', 'commit_hash') + g.db.add(self.test_obj) + g.db.commit() + self.test_id = self.test_obj.id + + _rate_limit_store.clear() + + def generate_system_token(self, email, password, scopes=None): + payload = { + 'email': email, + 'password': password, + 'token_name': 'test_token_' + self.create_random_string(8) + } + if scopes: + payload['scopes'] = scopes + + res = self.client.post( + '/api/v1/auth/tokens', data=json.dumps(payload), content_type='application/json') + if res.status_code != 201: + raise RuntimeError( + f"Failed to get token: {res.status_code} - {res.json}") + return res.json['token'] + + def test_health_check_unauthenticated(self): + res = self.client.get('/api/v1/system/health') + self.assertEqual(res.status_code, 200) + self.assertIn(res.json['status'], ['ok', 'degraded']) + self.assertIn('dependencies', res.json) + + def test_system_queue_requires_scope(self): + token = self.generate_system_token('user2@local.com', 'userpass123', ['runs:read']) + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': f'Bearer {token}'}) + # Forbidden due to missing scope + self.assertEqual(res.status_code, 403) + + def test_system_queue_with_scope(self): + # A test with no progress is "queued" + token = self.generate_system_token( + 'user2@local.com', 'userpass123', ['system:read']) + res = self.client.get('/api/v1/system/queue', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertIn('data', res.json) + self.assertEqual(res.json['meta']['queue_depth'], 1) + self.assertEqual(res.json['meta']['running_count'], 0) + self.assertEqual(res.json['data'][0]['run_id'], self.test_id) + self.assertEqual(res.json['data'][0]['status'], 'queued') + + def test_system_queue_platform_filter(self): + token = self.generate_system_token( + 'user2@local.com', 'userpass123', ['system:read']) + res = self.client.get('/api/v1/system/queue?platform=windows', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['meta']['queue_depth'], 0) + + @patch('mod_api.routes.system.text') + def test_system_health_db_down(self, mock_text): + mock_text.side_effect = Exception('DB Down') + res = self.client.get('/api/v1/system/health') + self.assertEqual(res.status_code, 503) + self.assertEqual(res.json['status'], 'down') + db_dep = next(d for d in res.json['dependencies'] if d['name'] == 'database') + self.assertEqual(db_dep['status'], 'down') + + def test_safe_resolve_path_traversal(self): + from mod_api.utils import safe_resolve + base = '/safe/base/path' + # Should return None for path traversal attempts + self.assertIsNone(safe_resolve(base, '../../../etc/passwd')) + self.assertIsNone(safe_resolve(base, '/etc/passwd')) diff --git a/tests/api/test_services_error_service.py b/tests/api/test_services_error_service.py new file mode 100644 index 000000000..4bf68027c --- /dev/null +++ b/tests/api/test_services_error_service.py @@ -0,0 +1,208 @@ +import datetime +from unittest.mock import MagicMock, PropertyMock + +from flask import g + +from mod_api.services.error_service import (_classify_infra_error, + _get_sample_id, + derive_error_summary, + derive_errors_for_run, + derive_infrastructure_errors) +from mod_regression.models import (Category, InputType, OutputType, + RegressionTest, RegressionTestOutput) +from mod_test.models import (Fork, Test, TestPlatform, TestProgress, + TestResult, TestResultFile, TestStatus, TestType) +from tests.api.base import ApiTestCase + + +class TestServicesErrorService(ApiTestCase): + def setUp(self): + super().setUp() + fork = Fork('https://github.com/test/test.git') + g.db.add(fork) + g.db.commit() + self.test_obj = Test(TestPlatform.linux, + TestType.commit, fork.id, 'master', 'commit_hash') + g.db.add(self.test_obj) + g.db.commit() + + self.category = Category('Test Category', 'Description') + g.db.add(self.category) + g.db.commit() + + self.reg_test1 = RegressionTest( + 1, 'cmd1', InputType.file, OutputType.file, self.category.id, 0) + self.reg_test2 = RegressionTest( + 1, 'cmd2', InputType.file, OutputType.file, self.category.id, 0) + g.db.add_all([self.reg_test1, self.reg_test2]) + g.db.commit() + + self.reg_out1 = RegressionTestOutput( + self.reg_test1.id, 'sample1_out', '.txt', 'exp1') + self.reg_out2 = RegressionTestOutput( + self.reg_test2.id, 'sample2_out', '.txt', 'exp2') + g.db.add_all([self.reg_out1, self.reg_out2]) + + dummy_out = RegressionTestOutput( + self.reg_test1.id, 'dummy', '', 'dummy') + dummy_out.id = -1 + g.db.merge(dummy_out) + + g.db.commit() + + def test_derive_errors_for_run_rc_mismatch(self): + tr = TestResult(self.test_obj.id, self.reg_test1.id, + 100, 1, 0) # runtime, exit_code, expected_rc + # The expected output was produced and matched (got=None), so the + # only error left is the exit-code mismatch. + rf = TestResultFile(self.test_obj.id, self.reg_test1.id, + self.reg_out1.id, 'exp1') + g.db.add_all([tr, rf]) + g.db.commit() + + errors = derive_errors_for_run(self.test_obj.id) + self.assertEqual(len(errors), 1) + self.assertEqual(errors[0]['type'], 'exit_code_mismatch') + self.assertEqual(errors[0]['severity'], 'error') + + def test_derive_errors_for_run_rc_mismatch_without_files_adds_missing(self): + # No result files at all: the rc mismatch is reported AND the + # expected output counts as missing — matching /runs/{id}/summary. + tr = TestResult(self.test_obj.id, self.reg_test1.id, 100, 1, 0) + g.db.add(tr) + g.db.commit() + + errors = derive_errors_for_run(self.test_obj.id) + types = sorted(e['type'] for e in errors) + self.assertEqual(types, ['exit_code_mismatch', 'missing_output']) + + def test_derive_errors_for_run_missing_output(self): + tr = TestResult(self.test_obj.id, self.reg_test1.id, 100, 0, 0) + rf = TestResultFile( + self.test_obj.id, self.reg_test1.id, -1, '', 'error') + g.db.add_all([tr, rf]) + g.db.commit() + + errors = derive_errors_for_run(self.test_obj.id) + print("ERRORS:", errors) + self.assertEqual(len(errors), 1) + self.assertEqual(errors[0]['type'], 'missing_output') + + def test_derive_errors_for_run_diff_mismatch(self): + tr = TestResult(self.test_obj.id, self.reg_test1.id, 100, 0, 0) + rf = TestResultFile(self.test_obj.id, self.reg_test1.id, + self.reg_out1.id, 'expected_hash', 'got_hash') + g.db.add_all([tr, rf]) + g.db.commit() + + errors = derive_errors_for_run(self.test_obj.id) + self.assertEqual(len(errors), 1) + self.assertEqual(errors[0]['type'], 'diff_mismatch') + self.assertEqual(errors[0]['severity'], 'warning') + + def test_derive_error_summary(self): + tr1 = TestResult(self.test_obj.id, self.reg_test1.id, + 100, 1, 0) # rc mismatch + # Matched output for reg_test1 so tr1 contributes only the rc error. + rf1 = TestResultFile(self.test_obj.id, self.reg_test1.id, + self.reg_out1.id, 'exp1') + tr2 = TestResult(self.test_obj.id, self.reg_test2.id, 100, 0, 0) + rf2 = TestResultFile(self.test_obj.id, self.reg_test2.id, + self.reg_out2.id, 'exp', 'got') # diff mismatch + g.db.add_all([tr1, rf1, tr2, rf2]) + g.db.commit() + + summary = derive_error_summary(self.test_obj.id) + self.assertEqual(len(summary), 2) + + # summary is a list of buckets + summary_dict = {b['key']: b for b in summary} + + self.assertEqual(summary_dict['exit_code_mismatch']['count'], 1) + self.assertEqual( + summary_dict['exit_code_mismatch']['severity'], 'error') + + self.assertEqual(summary_dict['diff_mismatch']['count'], 1) + self.assertEqual(summary_dict['diff_mismatch']['severity'], 'warning') + + def test_aggregate_error_severity_escalation(self): + # Create an error with severity 'warning' and another with 'error' in the same bucket + from mod_api.services.error_service import _aggregate_error_into_bucket + bucket = { + 'count': 1, + 'severity': 'warning', + 'sample_ids': [], + 'first_seen_at': None, + 'last_seen_at': None + } + + # New error with higher severity + err_error = {'severity': 'error', 'sample_id': 1} + _aggregate_error_into_bucket(err_error, bucket) + self.assertEqual(bucket['severity'], 'error') + self.assertEqual(bucket['count'], 2) + + # New error with lower severity should not downgrade + err_info = {'severity': 'info', 'sample_id': 2} + _aggregate_error_into_bucket(err_info, bucket) + self.assertEqual(bucket['severity'], 'error') + self.assertEqual(bucket['count'], 3) + + def test_derive_infrastructure_errors(self): + tp1 = TestProgress( + self.test_obj.id, TestStatus.canceled, 'provisioning VM failed') + tp1.timestamp = datetime.datetime(2023, 1, 1, 10, 0, 0) + + tp2 = TestProgress( + self.test_obj.id, TestStatus.canceled, 'merge conflict') + tp2.timestamp = datetime.datetime(2023, 1, 1, 10, 5, 0) + + g.db.add(tp1) + g.db.add(tp2) + g.db.commit() + + errors = derive_infrastructure_errors(self.test_obj.id) + self.assertEqual(len(errors), 2) + self.assertEqual(errors[0]['type'], 'vm_provisioning') + self.assertEqual(errors[1]['type'], 'merge') + + def test_derive_infrastructure_errors_excludes_api_cancels(self): + # A user/API cancellation must not be reported as an infrastructure error. + tp_api = TestProgress( + self.test_obj.id, TestStatus.canceled, 'Canceled by alice via API') + tp_api.timestamp = datetime.datetime(2023, 1, 1, 10, 0, 0) + tp_infra = TestProgress( + self.test_obj.id, TestStatus.canceled, 'build failed') + tp_infra.timestamp = datetime.datetime(2023, 1, 1, 10, 5, 0) + g.db.add(tp_api) + g.db.add(tp_infra) + g.db.commit() + + errors = derive_infrastructure_errors(self.test_obj.id) + self.assertEqual(len(errors), 1) + self.assertEqual(errors[0]['type'], 'build') + + def test_classify_infra_error(self): + self.assertEqual(_classify_infra_error( + 'timeout connecting to worker'), 'worker') + self.assertEqual(_classify_infra_error('failed to build'), 'build') + self.assertEqual(_classify_infra_error('storage is full'), 'storage') + self.assertEqual(_classify_infra_error( + 'fetch remote repository'), 'checkout') + self.assertEqual(_classify_infra_error('merge conflict'), 'merge') + self.assertEqual(_classify_infra_error( + 'random error string'), 'worker') + + def test_get_sample_id(self): + tr = TestResult(self.test_obj.id, 1, 100, 0, 0) + self.assertIsNone(_get_sample_id(tr)) + + tr.regression_test = MagicMock() + tr.regression_test.sample_id = 42 + self.assertEqual(_get_sample_id(tr), 42) + + # Test exception catching + mock_reg = MagicMock() + type(mock_reg).sample_id = PropertyMock(side_effect=RuntimeError('Mock exception')) + tr.regression_test = mock_reg + self.assertIsNone(_get_sample_id(tr)) diff --git a/tests/api/test_services_storage.py b/tests/api/test_services_storage.py new file mode 100644 index 000000000..d249f50d3 --- /dev/null +++ b/tests/api/test_services_storage.py @@ -0,0 +1,131 @@ +import os +import tempfile +from unittest.mock import MagicMock, patch + +from mod_api.services.storage import (get_log_file_path, + get_test_results_base_path, + resolve_artifact) +from tests.api.base import ApiTestCase + + +class TestServicesStorage(ApiTestCase): + def setUp(self): + super().setUp() + self.test_dir = tempfile.TemporaryDirectory() + self.dir_path = self.test_dir.name + + def tearDown(self): + self.test_dir.cleanup() + super().tearDown() + + def create_file(self, relative_path): + full_path = os.path.join(self.dir_path, relative_path) + os.makedirs(os.path.dirname(full_path), exist_ok=True) + with open(full_path, 'w') as f: + f.write('dummy content') + return full_path + + def mock_config_get(self, key, default=None): + if key == 'SAMPLE_REPOSITORY': + return self.dir_path + if key == 'GCS_SIGNED_URL_EXPIRY_LIMIT': + return 60 + return default + + @patch('run.config') + @patch('run.storage_client_bucket') + def test_resolve_artifact_both_exist(self, mock_bucket, mock_config): + mock_config.get.side_effect = self.mock_config_get + self.create_file('test_artifact.txt') + + mock_blob = MagicMock() + mock_blob.exists.return_value = True + mock_blob.generate_signed_url.return_value = 'https://signed.url' + mock_bucket.blob.return_value = mock_blob + + url, status = resolve_artifact('test_artifact.txt') + self.assertEqual(url, 'https://signed.url') + self.assertEqual(status, 'ok') + mock_blob.generate_signed_url.assert_called_once() + + @patch('run.config') + @patch('run.storage_client_bucket') + def test_resolve_artifact_only_gcs(self, mock_bucket, mock_config): + mock_config.get.side_effect = self.mock_config_get + + mock_blob = MagicMock() + mock_blob.exists.return_value = True + mock_blob.generate_signed_url.return_value = 'https://signed.url' + mock_bucket.blob.return_value = mock_blob + + url, status = resolve_artifact('test_artifact.txt') + self.assertEqual(url, 'https://signed.url') + self.assertEqual(status, 'degraded') + + @patch('run.config') + @patch('run.storage_client_bucket') + def test_resolve_artifact_gcs_blob_no_exists_check(self, mock_bucket, mock_config): + mock_config.get.side_effect = self.mock_config_get + self.create_file('test_artifact.txt') + + mock_blob = MagicMock() + mock_blob.generate_signed_url.return_value = 'https://signed.url' + mock_bucket.blob.return_value = mock_blob + + mock_blob.exists.return_value = True + resolve_artifact('test_artifact.txt') + mock_blob.exists.assert_called_once() + + @patch('run.config') + @patch('run.storage_client_bucket', new=None) + def test_resolve_artifact_only_local(self, mock_config): + mock_config.get.side_effect = self.mock_config_get + self.create_file('test_artifact.txt') + + url, status = resolve_artifact('test_artifact.txt') + self.assertIsNone(url) + self.assertEqual(status, 'degraded') + + @patch('run.config') + @patch('run.storage_client_bucket', new=None) + def test_resolve_artifact_missing(self, mock_config): + mock_config.get.side_effect = self.mock_config_get + + url, status = resolve_artifact('test_artifact.txt') + self.assertIsNone(url) + self.assertEqual(status, 'missing') + + @patch('run.config') + @patch('run.storage_client_bucket') + def test_resolve_artifact_gcs_exception(self, mock_bucket, mock_config): + mock_config.get.side_effect = self.mock_config_get + self.create_file('test_artifact.txt') + + mock_bucket.blob.side_effect = Exception("GCS Error") + + url, status = resolve_artifact('test_artifact.txt') + self.assertIsNone(url) + self.assertEqual(status, 'degraded') + + @patch('run.config') + def test_get_log_file_path_exists(self, mock_config): + mock_config.get.side_effect = self.mock_config_get + path = self.create_file('LogFiles/123.txt') + + result = get_log_file_path(123) + self.assertEqual(os.path.normpath(result), os.path.normpath(path)) + + @patch('run.config') + def test_get_log_file_path_missing(self, mock_config): + mock_config.get.side_effect = self.mock_config_get + + result = get_log_file_path(123) + self.assertIsNone(result) + + @patch('run.config') + def test_get_test_results_base_path(self, mock_config): + mock_config.get.return_value = '/fake/repo' + + result = get_test_results_base_path() + expected = os.path.join('/fake/repo', 'TestResults') + self.assertEqual(result, expected)