diff --git a/mod_api/__init__.py b/mod_api/__init__.py index dc930a63..a54a9413 100644 --- a/mod_api/__init__.py +++ b/mod_api/__init__.py @@ -37,4 +37,5 @@ # 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 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/samples.py b/mod_api/routes/samples.py new file mode 100644 index 00000000..9ebe7fb0 --- /dev/null +++ b/mod_api/routes/samples.py @@ -0,0 +1,647 @@ +""" +Sample and regression test routes. + +GET /runs/{id}/samples Per-run regression test results +GET /runs/{id}/samples/{sid} Single result in a run +GET /samples Media sample catalog +GET /samples/{id} Single media sample +GET /samples/{id}/history Cross-run history for a sample +GET /regression-tests Regression test definitions +""" + +from collections import defaultdict + +from flask import g, request +from sqlalchemy import func +from sqlalchemy.orm import joinedload, selectinload + +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_date_range, + validate_offset_pagination, + validate_path_id) +from mod_api.models.api_token import Scope +from mod_api.schemas.samples import SampleHistoryEntrySchema +from mod_api.services.status import (batch_get_run_data, derive_output_status, + derive_sample_status, get_run_timestamps, + is_dummy_row) +from mod_api.utils import paginated_response, single_response +from mod_regression.models import (Category, RegressionTest, + RegressionTestOutput) +from mod_sample.models import Sample, Tag +from mod_test.models import (Test, TestPlatform, TestProgress, TestResult, + TestResultFile) + +# Valid per-sample status values accepted by the ?status filter. Limited to the +# statuses derive_sample_status can actually emit, so filtering can't silently +# return empty for a value that never occurs. +_VALID_SAMPLE_STATUSES = frozenset({ + 'pass', 'fail', 'missing_output', 'not_started', +}) + + +def _preload_expected_outputs(results): + """Map regression_test_id -> [RegressionTestOutput] for the given results. + + Lets per-sample status derivation use the same missing-output detection + as /runs/{id}/summary, so the two endpoints can't disagree. + """ + rt_ids = {r.regression_test_id for r in results} + expected_by_rt = defaultdict(list) + if rt_ids: + for rto in RegressionTestOutput.query.filter( + RegressionTestOutput.regression_id.in_(rt_ids)).all(): + expected_by_rt[rto.regression_id].append(rto) + return expected_by_rt + + +def _serialize_outputs(result_files): + outputs = [] + for rf in result_files: + if is_dummy_row(rf): + continue + outputs.append({ + 'output_id': rf.regression_test_output_id, + 'filename': ( + rf.regression_test_output.create_correct_filename(rf.expected) + if rf.regression_test_output else rf.expected + ), + 'status': derive_output_status(rf), + }) + return outputs + + +def _serialize_run_sample(result, result_files, expected_outputs=None): + """Build the per-regression-test result dict for a run.""" + status = derive_sample_status(result, result_files, expected_outputs) + outputs = _serialize_outputs(result_files) + + sample_name = None + sample_id = None + command = None + categories = [] + + if result.regression_test: + rt = result.regression_test + command = rt.command + if rt.sample: + sample_id = rt.sample_id + sample_name = rt.sample.original_name + if rt.categories: + categories = [c.name for c in rt.categories] + + return { + 'regression_test_id': result.regression_test_id, + 'sample_id': sample_id, + 'sample_name': sample_name, + 'status': status, + 'exit_code': result.exit_code, + 'expected_rc': result.expected_rc, + 'runtime_ms': result.runtime, + 'command': command, + 'categories': categories, + 'outputs': outputs, + } + + +def _filter_run_samples_by_tag(serialized, tag_filter): + tag_lower = tag_filter.lower() + tagged_sample_ids = set() + + valid_sample_ids = [s['sample_id'] + for s in serialized if s.get('sample_id')] + samples = Sample.query.options(joinedload(Sample.tags)).filter( + Sample.id.in_(valid_sample_ids)).all() if valid_sample_ids else [] + sample_map = {sample.id: sample for sample in samples} + + for s in serialized: + if s['sample_id']: + sample = sample_map.get(s['sample_id']) + if sample and any(tag_lower == t.name.lower() + for t in sample.tags): + tagged_sample_ids.add(s['sample_id']) + return [s for s in serialized if s.get('sample_id') in tagged_sample_ids] + + +def _apply_run_sample_filters(serialized, args): + status_filter = args.get('status') + if status_filter: + serialized = [s for s in serialized if s['status'] == status_filter] + + name_filter = args.get('name') + if name_filter: + name_lower = name_filter.lower() + serialized = [s for s in serialized if s.get( + 'sample_name') and name_lower in s['sample_name'].lower()] + + tag_filter = args.get('tag') + if tag_filter: + serialized = _filter_run_samples_by_tag(serialized, tag_filter) + + category_filter = args.get('category') + if category_filter: + cat_lower = category_filter.lower() + serialized = [ + s for s in serialized + if s.get('categories') and cat_lower in [ + c.lower() for c in s['categories'] + ] + ] + return serialized + + +@mod_api.route('/runs//samples', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_path_id('run_id') +@validate_offset_pagination() +def list_run_samples(run_id, limit=50, offset=0): + """ + List per-sample results for a run, with optional filters. + + Supports ?status, ?name, ?tag, ?category query params. + """ + # Validate the status filter up front, before any DB work. + status_filter = request.args.get('status') + if status_filter and status_filter not in _VALID_SAMPLE_STATUSES: + return make_error_response( + 'validation_error', + f"Invalid status: {status_filter}", + http_status=400 + ) + + 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.options( + joinedload(TestResult.regression_test) + .joinedload(RegressionTest.sample), + joinedload(TestResult.regression_test) + .selectinload(RegressionTest.categories), + ).filter_by(test_id=run_id).all() + + # Preload TestResultFiles together with the expected-output rows that + # derive_sample_status compares against. + 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 so per-sample status matches /summary. + expected_by_rt = _preload_expected_outputs(results) + + # Serialize list to filter by derived status and joined fields + serialized = [] + for result in results: + result_files = files_by_result.get(result.regression_test_id, []) + serialized.append(_serialize_run_sample( + result, result_files, + expected_by_rt.get(result.regression_test_id))) + + # Apply query param filters. + serialized = _apply_run_sample_filters(serialized, request.args) + + total = len(serialized) + paged = serialized[offset:offset + limit] + return paginated_response(paged, total, limit, offset) + + +@mod_api.route('/runs//samples/', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_path_id('run_id') +@validate_path_id('regression_test_id') +def get_run_sample(run_id, regression_test_id): + """Get a single regression test result within a 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) + + result = TestResult.query.options( + joinedload(TestResult.regression_test) + .joinedload(RegressionTest.sample), + joinedload(TestResult.regression_test) + .selectinload(RegressionTest.categories), + ).filter_by( + test_id=run_id, + regression_test_id=regression_test_id, + ).first() + if result is None: + return make_error_response( + 'not_found', + f'Regression test {regression_test_id} not found in run {run_id}.', + http_status=404, + ) + + result_files = TestResultFile.query.options( + joinedload(TestResultFile.regression_test_output) + .joinedload(RegressionTestOutput.multiple_files) + ).filter_by( + test_id=run_id, + regression_test_id=regression_test_id, + ).all() + + expected_by_rt = _preload_expected_outputs([result]) + return single_response(_serialize_run_sample( + result, result_files, expected_by_rt.get(result.regression_test_id))) + + +@mod_api.route('/samples', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_offset_pagination() +def list_samples(limit=50, offset=0): + """ + List media samples from the catalog. + + Supports ?name, ?extension, ?tag, ?sha256, + ?status (active/inactive) filters. + """ + query = Sample.query.options(joinedload(Sample.tags)) + + name = request.args.get('name') + if name: + # Escape LIKE wildcards to prevent unintended pattern matching. + # The explicit escape char makes the backslash escaping portable + # rather than relying on the backend's default. + safe_name = name.replace('\\', '\\\\').replace( + '%', '\\%').replace('_', '\\_') + query = query.filter( + Sample.original_name.ilike(f'%{safe_name}%', escape='\\')) + + extension = request.args.get('extension') + if extension: + query = query.filter(Sample.extension == extension) + + sha256_filter = request.args.get('sha256') + if sha256_filter: + query = query.filter(Sample.sha == sha256_filter) + + tag_filter = request.args.get('tag') + if tag_filter: + + query = query.filter(Sample.tags.any( + func.lower(Tag.name) == tag_filter.lower())) + + status_filter = request.args.get('status') + if status_filter: + if status_filter.lower() not in ('active', 'inactive'): + return make_error_response( + 'validation_error', + 'Invalid status: {status_filter}. ' + 'Must be active or inactive.'.format( + status_filter=status_filter), + http_status=400) + want_active = status_filter.lower() == 'active' + if want_active: + query = query.filter( + Sample.tests.any(RegressionTest.active == True) # noqa: E712 + ) # tests refers to RegressionTest + else: + query = query.filter( + ~Sample.tests.any(RegressionTest.active == True) # noqa: E712 + ) # tests refers to RegressionTest + + # Paginate at DB level without Python-side filters + total = query.count() + samples = query.offset(offset).limit(limit).all() + + # Batch load active regression test counts + sample_ids = [s.id for s in samples] + counts_list = g.db.query( + RegressionTest.sample_id, + func.count(RegressionTest.id) + ).filter( + RegressionTest.sample_id.in_(sample_ids), + RegressionTest.active == True # noqa: E712 + ).group_by(RegressionTest.sample_id).all() if sample_ids else [] + counts = dict(counts_list) + + serialized = [] + for s in samples: + active_count = counts.get(s.id, 0) + serialized.append({ + 'sample_id': s.id, + 'sha': s.sha, + 'extension': s.extension, + 'original_name': s.original_name, + 'filename': s.filename, + 'tags': [t.name for t in s.tags], + 'regression_test_count': active_count, + 'active': active_count > 0, + }) + + return paginated_response(serialized, total, limit, offset) + + +@mod_api.route('/samples/', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_path_id('sample_id') +def get_sample(sample_id): + """Get a single media sample by its ID.""" + sample = Sample.query.options(joinedload(Sample.tags)).filter( + Sample.id == sample_id).first() + if sample is None: + return make_error_response( + 'not_found', + f'Sample {sample_id} not found.', + http_status=404) + + active_count = RegressionTest.query.filter_by( + sample_id=sample.id, active=True + ).count() + + return single_response({ + 'sample_id': sample.id, + 'sha': sample.sha, + 'extension': sample.extension, + 'original_name': sample.original_name, + 'filename': sample.filename, + 'tags': [t.name for t in sample.tags], + 'regression_test_count': active_count, + 'active': active_count > 0, + }) + + +def _get_history_failure_signature(result, result_files, status): + if status == 'fail': + for rf in result_files: + if rf.got is not None and not is_dummy_row(rf): + return f'diff_mismatch:output:{rf.regression_test_output_id}' + if result.exit_code != result.expected_rc: + return f'exit_code_mismatch:rc:{result.exit_code}' + elif status == 'missing_output': + return 'missing_output' + return None + + +def _process_history_entries( + results, + files_by_result, + status_filter, + timestamps_map=None, + test_map=None, + expected_by_rt=None): + entries = [] + for result in results: + test = test_map.get(result.test_id) if test_map else result.test + if test is None: + continue + + result_files = files_by_result.get( + (result.test_id, result.regression_test_id), []) + expected = expected_by_rt.get( + result.regression_test_id) if expected_by_rt else None + status = derive_sample_status(result, result_files, expected) + + if status_filter and status != status_filter: + continue + + failure_sig = _get_history_failure_signature( + result, result_files, status) + if timestamps_map is not None and test.id in timestamps_map: + timestamps = timestamps_map[test.id] + else: + timestamps = get_run_timestamps(test) + + entries.append({ + 'run_id': test.id, + 'regression_test_id': result.regression_test_id, + 'status': status, + 'platform': test.platform.value, + 'branch': test.branch, + 'commit_sha': test.commit, + 'tested_at': timestamps.get('completed_at') or timestamps.get('started_at'), + 'failure_signature': failure_sig, + }) + return entries + + +def _apply_history_filters( + query, + branch, + platform, + created_after, + created_before): + if branch: + query = query.filter(Test.branch == branch) + + 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', 'Invalid platform: {platform}. ' + 'Must be one of: {valid_platforms}.'.format( + platform=platform, valid_platforms=valid_platforms + ), + http_status=400, + ) + + if created_after or created_before: + + first_progress = ( + g.db.query(TestProgress.test_id, func.min( + TestProgress.timestamp).label('min_ts')) + .group_by(TestProgress.test_id) + .subquery() + ) + query = query.join(first_progress, Test.id == first_progress.c.test_id) + if created_after: + query = query.filter(first_progress.c.min_ts >= created_after) + if created_before: + query = query.filter(first_progress.c.min_ts <= created_before) + + return query, None + + +@mod_api.route('/samples//history', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_path_id('sample_id') +@validate_offset_pagination() +@validate_date_range +def get_sample_history( + sample_id, + limit=50, + offset=0, + created_after=None, + created_before=None): + """ + Show how a sample performed across different runs. + + Use failure_signature to tell apart genuine regressions from infra flakes. + """ + sample = Sample.query.options(joinedload(Sample.tags)).filter( + Sample.id == sample_id).first() + if sample is None: + return make_error_response( + 'not_found', + f'Sample {sample_id} not found.', + http_status=404) + + regression_tests = RegressionTest.query.filter_by( + sample_id=sample_id).all() + rt_ids = [rt.id for rt in regression_tests] + + if not rt_ids: + return paginated_response([], 0, limit, offset) + + # Validate the status filter up front, before any heavy query. + status_filter = request.args.get('status') + if status_filter and status_filter not in _VALID_SAMPLE_STATUSES: + return make_error_response( + 'validation_error', + f"Invalid status: {status_filter}", + http_status=400 + ) + + query = TestResult.query.filter( + TestResult.regression_test_id.in_(rt_ids) + ).join(Test, Test.id == TestResult.test_id) + + branch = request.args.get('branch') + platform = request.args.get('platform') + + query, err = _apply_history_filters( + query, branch, platform, created_after, created_before) + if err: + return err + + results = query.order_by(Test.id.desc()).all() + + # Preload TestResultFiles + test_ids = list({r.test_id for r in results}) + all_files = TestResultFile.query.options( + joinedload(TestResultFile.regression_test_output) + .joinedload(RegressionTestOutput.multiple_files) + ).filter( + TestResultFile.test_id.in_(test_ids)).all() if test_ids else [] + files_by_result = defaultdict(list) + for f in all_files: + files_by_result[(f.test_id, f.regression_test_id)].append(f) + + # Preload expected outputs so status matches /summary and /samples. + expected_by_rt = _preload_expected_outputs(results) + + # Batch load tests to avoid N+1 in _process_history_entries + unique_tests = Test.query.filter( + Test.id.in_(test_ids)).all() if test_ids else [] + test_map = {t.id: t for t in unique_tests} + + # Batch compute timestamps for all referenced tests + _, timestamps_map = batch_get_run_data(unique_tests) + + entries = _process_history_entries( + results, + files_by_result, + status_filter, + timestamps_map=timestamps_map, + test_map=test_map, + expected_by_rt=expected_by_rt) + + total = len(entries) + paged = entries[offset:offset + limit] + + return paginated_response( + paged, total, limit, offset, schema=SampleHistoryEntrySchema() + ) + + +def _serialize_rt(rt): + return { + 'regression_test_id': rt.id, + 'sample_id': rt.sample_id, + 'sample_name': rt.sample.original_name if rt.sample else None, + 'command': rt.command, + 'input_type': rt.input_type.value, + 'output_type': rt.output_type.value, + 'expected_rc': rt.expected_rc, + 'active': rt.active, + 'categories': [c.name for c in rt.categories], + 'description': rt.description, + } + + +@mod_api.route('/regression-tests', methods=['GET']) +@require_scope(Scope.RUNS_READ) +@validate_offset_pagination() +def list_regression_tests(limit=50, offset=0): + """ + List regression test definitions. + + Supports ?active, ?category, ?tag, ?sample_id filters. Note: when + ?active is omitted it defaults to true, so inactive regression tests + are hidden unless ?active=false is passed explicitly. + """ + query = RegressionTest.query.options( + joinedload(RegressionTest.sample), + selectinload(RegressionTest.categories), + ) + + active_filter = request.args.get('active') + if active_filter is not None: + value = active_filter.lower() + if value in ('true', '1', 'yes'): + is_active = True + elif value in ('false', '0', 'no'): + is_active = False + else: + # Reject garbage instead of silently treating it as false. + return make_error_response( + 'validation_error', + f'Invalid active filter: {active_filter}. ' + 'Must be true or false.', + details={'fields': {'active': 'Must be true or false.'}}, + http_status=400, + ) + else: + is_active = True + query = query.filter(RegressionTest.active == is_active) + + category = request.args.get('category') + if category: + query = query.join(RegressionTest.categories).filter( + Category.name == category) + + sample_id_filter = request.args.get('sample_id') + if sample_id_filter: + try: + sid = int(sample_id_filter) + if sid < 1 or sid > 2147483647: + raise ValueError("Out of bounds") + query = query.filter(RegressionTest.sample_id == sid) + except (ValueError, TypeError): + return make_error_response( + 'validation_error', + 'sample_id must be a positive integer ' + 'between 1 and 2147483647.', + details={ + 'fields': { + 'sample_id': 'Must be a positive integer ' + 'between 1 and 2147483647.'}}, + http_status=400, + ) + + tag_filter = request.args.get('tag') + if tag_filter: + query = query.filter( + RegressionTest.sample.has( + Sample.tags.any(func.lower(Tag.name) == tag_filter.lower()) + ) + ) + + # Paginate at DB level + total = query.count() + tests = query.offset(offset).limit(limit).all() + serialized = [_serialize_rt(rt) for rt in tests] + return paginated_response(serialized, total, limit, offset) diff --git a/mod_api/schemas/samples.py b/mod_api/schemas/samples.py new file mode 100644 index 00000000..0b43b012 --- /dev/null +++ b/mod_api/schemas/samples.py @@ -0,0 +1,18 @@ +"""Schemas for sample endpoints.""" + +from marshmallow import Schema, fields + +from mod_api.schemas.common import DATETIME_FORMAT + + +class SampleHistoryEntrySchema(Schema): + """One row in a sample's cross-run history.""" + + run_id = fields.Integer(required=True) + regression_test_id = fields.Integer(required=True) + status = fields.String(required=True) + platform = fields.String(required=True) + branch = fields.String(required=True) + commit_sha = fields.String(required=True) + tested_at = fields.DateTime(allow_none=True, format=DATETIME_FORMAT) + failure_signature = fields.String(allow_none=True) diff --git a/tests/api/test_routes_samples.py b/tests/api/test_routes_samples.py new file mode 100644 index 00000000..f38727af --- /dev/null +++ b/tests/api/test_routes_samples.py @@ -0,0 +1,284 @@ +from flask import g +from sqlalchemy import event + +from mod_api.middleware.rate_limit import _rate_limit_store +from mod_regression.models import (Category, InputType, OutputType, + RegressionTest, RegressionTestOutput) +from mod_sample.models import Sample +from mod_test.models import TestResult, TestResultFile +from tests.api.base import ApiTestCase + + +class TestRoutesSamples(ApiTestCase): + def setUp(self): + super().setUp() + self.setup_run_data('samp') + self.sample = Sample('test_sha', 'txt', 'test_sample') + g.db.add(self.sample) + g.db.commit() + self.sample_id = self.sample.id + + self.category = Category('Test Category', 'Description') + g.db.add(self.category) + g.db.commit() + + self.reg_test = RegressionTest( + self.sample_id, + 'command', + InputType.file, + OutputType.file, + self.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') + 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', + None) + g.db.add(self.result_file) + g.db.commit() + + _rate_limit_store.clear() + + def test_list_run_samples(self): + token = self.get_token('samp_user@local.com', + 'userpass123', 't1', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples', + headers={'Authorization': f'Bearer {token}'} + ) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 1) + self.assertEqual(res.json['data'][0] + ['regression_test_id'], self.reg_test_id) + + def test_list_run_samples_missing_output_consistent(self): + # A regression test with a non-ignored expected output but no + # result file must report 'missing_output' (same derivation as + # /runs/{id}/summary), not 'pass'. Guards the expected-outputs + # threading in list_run_samples. + reg_test2 = RegressionTest( + self.sample_id, 'command2', InputType.file, OutputType.file, + self.category.id, 0) + g.db.add(reg_test2) + g.db.commit() + reg_test2_id = reg_test2.id + g.db.add(RegressionTestOutput(reg_test2_id, 'hash2', '.txt', 'exp2')) + # A result whose expected output has no matching TestResultFile. + g.db.add(TestResult(self.test_id, reg_test2_id, 0, 0, 0)) + g.db.commit() + + token = self.get_token('samp_user@local.com', + 'userpass123', 'tmo', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples', + headers={'Authorization': f'Bearer {token}'} + ) + self.assertEqual(res.status_code, 200) + entry = next(s for s in res.json['data'] + if s['regression_test_id'] == reg_test2_id) + self.assertEqual(entry['status'], 'missing_output') + + def _count_queries(self, url, token): + """Return the number of SQL statements one GET request executes.""" + statements = [] + + def counter(conn, cursor, statement, parameters, context, + executemany): + statements.append(statement) + + engine = g.db.get_bind() + event.listen(engine, 'before_cursor_execute', counter) + try: + res = self.client.get( + url, headers={'Authorization': f'Bearer {token}'}) + finally: + event.remove(engine, 'before_cursor_execute', counter) + self.assertEqual(res.status_code, 200) + return len(statements) + + def test_list_run_samples_query_count_is_flat(self): + # Guards against reintroducing per-regression-test lazy loads: + # the number of queries must not depend on how many regression + # tests the run has. + token = self.get_token('samp_user@local.com', + 'userpass123', 'tqc', scopes=['runs:read']) + # Plain ids only: the request below detaches ORM objects held by + # this test's session. + category_id = self.category.id + url = f'/api/v1/runs/{self.test_id}/samples' + baseline = self._count_queries(url, token) + + for i in range(8): + rt = RegressionTest(self.sample_id, f'command_qc{i}', + InputType.file, OutputType.file, + category_id, 0) + g.db.add(rt) + g.db.commit() + rto = RegressionTestOutput(rt.id, f'hash_qc{i}', '.txt', 'exp') + g.db.add(rto) + g.db.commit() + g.db.add(TestResult(self.test_id, rt.id, 0, 0, 0)) + g.db.add(TestResultFile(self.test_id, rt.id, rto.id, + f'hash_qc{i}', None)) + g.db.commit() + + self.assertEqual(self._count_queries(url, token), baseline) + + def test_get_run_sample(self): + token = self.get_token('samp_user@local.com', + 'userpass123', 't2', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/{self.reg_test_id}', + headers={'Authorization': f'Bearer {token}'} + ) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['regression_test_id'], self.reg_test_id) + + def test_list_samples(self): + token = self.get_token('samp_user@local.com', + 'userpass123', 't3', scopes=['runs:read']) + res = self.client.get( + '/api/v1/samples', headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 3) + self.assertTrue( + any(s['sample_id'] == self.sample_id for s in res.json['data'])) + + def test_get_sample(self): + token = self.get_token('samp_user@local.com', + 'userpass123', 't4', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/samples/{self.sample_id}', + headers={'Authorization': f'Bearer {token}'} + ) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.json['sample_id'], self.sample_id) + + def test_get_sample_history(self): + token = self.get_token('samp_user@local.com', + 'userpass123', 't5', scopes=['runs:read']) + res = self.client.get( + f'/api/v1/samples/{self.sample_id}/history', + headers={'Authorization': f'Bearer {token}'} + ) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 1) + self.assertTrue( + any(h['run_id'] == self.test_id for h in res.json['data'])) + + def test_list_regression_tests(self): + token = self.get_token('samp_user@local.com', + 'userpass123', 't6', scopes=['runs:read']) + res = self.client.get('/api/v1/regression-tests', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 3) + self.assertTrue(any(rt['regression_test_id'] == self.reg_test_id + for rt in res.json['data'])) + + def test_list_regression_tests_active_filter(self): + # Create an inactive regression test + rt_inactive = RegressionTest( + self.sample_id, + 'cmd_inactive', + InputType.file, + OutputType.file, + self.category.id, + 0) + rt_inactive.active = False + g.db.add(rt_inactive) + g.db.commit() + rt_inactive_id = rt_inactive.id + + token = self.get_token( + 'samp_user@local.com', + 'userpass123', + 't_active_filter', + scopes=['runs:read']) + + # Default active=true + res = self.client.get('/api/v1/regression-tests', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 200) + self.assertTrue(any(rt['regression_test_id'] == self.reg_test_id + for rt in res.json['data'])) + self.assertFalse(any(rt['regression_test_id'] == rt_inactive_id + for rt in res.json['data'])) + + res_false = self.client.get( + '/api/v1/regression-tests?active=false', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res_false.status_code, 200) + self.assertFalse(any(rt['regression_test_id'] == self.reg_test_id + for rt in res_false.json['data'])) + self.assertTrue(any(rt['regression_test_id'] == rt_inactive_id + for rt in res_false.json['data'])) + + def test_list_samples_invalid_status(self): + token = self.get_token( + 'samp_user@local.com', + 'userpass123', + scopes=['runs:read']) + res = self.client.get( + '/api/v1/samples?status=invalid', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 400) + + def test_get_sample_not_found(self): + token = self.get_token( + 'samp_user@local.com', + 'userpass123', + scopes=['runs:read']) + res = self.client.get('/api/v1/samples/99999', + headers={'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 404) + + def test_list_run_samples_invalid_status(self): + token = self.get_token( + 'samp_user@local.com', + 'userpass123', + scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples?status=typo', + headers={'Authorization': f'Bearer {token}'} + ) + self.assertEqual(res.status_code, 400) + + def test_get_run_sample_not_found(self): + token = self.get_token( + 'samp_user@local.com', + 'userpass123', + scopes=['runs:read']) + res = self.client.get( + f'/api/v1/runs/{self.test_id}/samples/999', + headers={'Authorization': f'Bearer {token}'} + ) + self.assertEqual(res.status_code, 404) + + def test_get_sample_history_invalid_status(self): + token = self.get_token( + 'samp_user@local.com', + 'userpass123', + scopes=['runs:read']) + res = self.client.get( + f'/api/v1/samples/{self.sample_id}/history?status=typo', + headers={ + 'Authorization': f'Bearer {token}'}) + self.assertEqual(res.status_code, 400)