fix: Realign with monolithic PR and fix SonarQube complexity

Author: pulk17

.flaskenv

@@ -0,0 +1,3 @@
+FLASK_APP=run.py
+FLASK_RUN_PORT=5000
+FLASK_DEBUG=0

.postman/resources.yaml

@@ -0,0 +1,9 @@
+# Use this workspace to collaborate
+workspace:
+  id: 960368df-b016-461c-8a8e-c79a10d45672
+
+localResources:
+  collections:
+    - ..\postman_collection.json
+  specs:
+    - ..\postman\specs\New Specification\New Specification.yaml

.postman/workflows.yaml

@@ -0,0 +1,20 @@
+# This is a Workflow metadata file for defining entity relationships in this workspace.
+# Each workflow defines relations between entities and may include optional behavior/settings.
+# The Paths are relative to the .postman directory (avoid absolute paths).
+# Keep relation entries explicit and avoid duplicate pairs within the same workflow.
+# This file is declarative metadata and does not execute workflows by itself.
+workflows:
+  syncSpecToCollection:
+    - spec: ..\postman\specs\New Specification\New Specification.yaml
+      collection: ..\postman\collections\SP
+      options:
+        requestNameSource: URL
+        folderStrategy: Tags
+        syncExamples: false
+  syncCollectionToSpec:
+    - spec: ..\postman\specs\New Specification\New Specification.yaml
+      collection: ..\postman\collections\SP
+      options:
+        requestNameSource: URL
+        folderStrategy: Tags
+        syncExamples: false

diff_master.diff

@@ -0,0 +1,44 @@
+"""Add api_token table for scoped API token auth.
+
+Revision ID: d4f8e2a1b3c7
+Revises: c8f3a2b1d4e5
+Create Date: 2026-06-11 03:00:00.000000
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'd4f8e2a1b3c7'
+down_revision = 'c8f3a2b1d4e5'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    """Apply the migration."""
+    op.add_column('user', sa.Column('github_login', sa.String(length=255), nullable=True))
+    op.create_table(
+        'api_token',
+        sa.Column('id', sa.Integer(), nullable=False, autoincrement=True),
+        sa.Column('user_id', sa.Integer(), nullable=False),
+        sa.Column('token_name', sa.String(length=50), nullable=False),
+        sa.Column('token_hash', sa.String(length=255), nullable=False),
+        sa.Column('token_prefix', sa.String(length=16), nullable=False),
+        sa.Column('scopes_json', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+        sa.Column('expires_at', sa.DateTime(timezone=True), nullable=False),
+        sa.Column('revoked_at', sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint('id'),
+        sa.ForeignKeyConstraint(['user_id'], ['user.id'], onupdate='CASCADE', ondelete='CASCADE'),
+        sa.UniqueConstraint('user_id', 'token_name', name='uq_user_token_name'),
+        mysql_engine='InnoDB'
+    )
+    op.create_index('ix_api_token_token_prefix', 'api_token', ['token_prefix'])
+
+
+def downgrade():
+    """Revert the migration."""
+    op.drop_index('ix_api_token_token_prefix', table_name='api_token')
+    op.drop_table('api_token')
+    op.drop_column('user', 'github_login')

migrations/versions/d4f8e2a1b3c7_.py

@@ -245,48 +245,38 @@ def decorated(*args, **kwargs):
     return decorator
 
 
+def _parse_iso8601_date(param_name, param_str):
+    from datetime import datetime, timezone
+    if not param_str:
+        return None, None
+    try:
+        dt = datetime.fromisoformat(param_str.replace('Z', '+00:00'))
+    except ValueError:
+        return None, make_error_response(
+            'validation_error',
+            f'{param_name} must be a valid ISO 8601 datetime.',
+            details={'fields': {param_name: 'Invalid ISO 8601 format.'}},
+            http_status=400,
+        )
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt, None
+
+
 def validate_date_range(f):
     """Parse date query params and reject inverted ranges."""
     @wraps(f)
     def decorated(*args, **kwargs):
-        from datetime import datetime, timezone
-
         created_after_str = request.args.get('created_after')
         created_before_str = request.args.get('created_before')
-        created_after = None
-        created_before = None
 
-        if created_after_str:
-            try:
-                created_after = datetime.fromisoformat(
-                    created_after_str.replace('Z', '+00:00'))
-            except ValueError:
-                return make_error_response(
-                    'validation_error',
-                    'created_after must be a valid ISO 8601 datetime.',
-                    details={
-                        'fields': {
-                            'created_after': 'Invalid ISO 8601 format.'}},
-                    http_status=400,
-                )
-            if created_after.tzinfo is None:
-                created_after = created_after.replace(tzinfo=timezone.utc)
+        created_after, err = _parse_iso8601_date('created_after', created_after_str)
+        if err:
+            return err
 
-        if created_before_str:
-            try:
-                created_before = datetime.fromisoformat(
-                    created_before_str.replace('Z', '+00:00'))
-            except ValueError:
-                return make_error_response(
-                    'validation_error',
-                    'created_before must be a valid ISO 8601 datetime.',
-                    details={
-                        'fields': {
-                            'created_before': 'Invalid ISO 8601 format.'}},
-                    http_status=400,
-                )
-            if created_before.tzinfo is None:
-                created_before = created_before.replace(tzinfo=timezone.utc)
+        created_before, err = _parse_iso8601_date('created_before', created_before_str)
+        if err:
+            return err
 
         if created_after and created_before and created_after > created_before:
             return make_error_response(

mod_api/middleware/validation.py

@@ -127,7 +127,17 @@ def _validate_run_permissions(user, target_repo, main_repo_full):
             )
     else:
         owner = target_repo.split('/')[0]
-        github_login = user.name or ''
+        github_login = user.github_login
+
+        if not github_login and user.github_token:
+            from mod_auth.controllers import fetch_username_from_token
+            github_login = fetch_username_from_token(user)
+            if github_login:
+                user.github_login = github_login
+                from flask import g
+                g.db.add(user)
+
+        github_login = github_login or ''
 
         is_owner = bool(github_login) and owner.lower() == github_login.lower()
         is_staff = user.role.value in ('admin', 'tester', 'contributor')

mod_api/routes/runs.py

@@ -187,32 +187,25 @@ def _compute_run_status(t_prog, results_by_test, files_by_test_and_rt, t_id, exp
         return 'incomplete'
 
 
-def batch_get_run_data(tests: list) -> tuple:
-    """
-    Batch compute derive_run_status and get_run_timestamps for a list of tests.
-
-    Returns (statuses_dict, timestamps_dict)
-    """
-    if not tests:
-        return {}, {}
-
-    test_ids = [t.id for t in tests]
-
-    # Preload TestProgress
+def _preload_test_progress(test_ids):
     all_progress = TestProgress.query.filter(TestProgress.test_id.in_(
         test_ids)).order_by(TestProgress.id.asc()).all()
     progress_by_test = {tid: [] for tid in test_ids}
     for p in all_progress:
         progress_by_test[p.test_id].append(p)
+    return progress_by_test
+
 
-    # Preload TestResult
+def _preload_test_results(test_ids):
     all_results = TestResult.query.filter(
         TestResult.test_id.in_(test_ids)).all()
     results_by_test = {tid: [] for tid in test_ids}
     for r in all_results:
         results_by_test[r.test_id].append(r)
+    return results_by_test
 
-    # Preload TestResultFile
+
+def _preload_test_result_files(test_ids):
     from sqlalchemy.orm import joinedload
 
     from mod_regression.models import RegressionTestOutput
@@ -226,22 +219,44 @@ def batch_get_run_data(tests: list) -> tuple:
         if key not in files_by_test_and_rt:
             files_by_test_and_rt[key] = []
         files_by_test_and_rt[key].append(f)
+    return files_by_test_and_rt
+
 
-    # Preload expected outputs (RegressionTestOutput) for missing-output detection
+def _preload_expected_outputs(results_by_test, test_ids):
     all_rt_ids = set()
     for tid in test_ids:
         for r in results_by_test.get(tid, []):
             all_rt_ids.add(r.regression_test_id)
 
-    expected_outputs_by_rt = {}
-    if all_rt_ids:
-        from collections import defaultdict
-        all_expected = RegressionTestOutput.query.filter(
-            RegressionTestOutput.regression_id.in_(all_rt_ids)
-        ).all()
-        expected_outputs_by_rt = defaultdict(list)
-        for rto in all_expected:
-            expected_outputs_by_rt[rto.regression_id].append(rto)
+    if not all_rt_ids:
+        return {}
+
+    from collections import defaultdict
+
+    from mod_regression.models import RegressionTestOutput
+    all_expected = RegressionTestOutput.query.filter(
+        RegressionTestOutput.regression_id.in_(all_rt_ids)
+    ).all()
+    expected_outputs_by_rt = defaultdict(list)
+    for rto in all_expected:
+        expected_outputs_by_rt[rto.regression_id].append(rto)
+    return expected_outputs_by_rt
+
+
+def batch_get_run_data(tests: list) -> tuple:
+    """
+    Batch compute derive_run_status and get_run_timestamps for a list of tests.
+
+    Returns (statuses_dict, timestamps_dict)
+    """
+    if not tests:
+        return {}, {}
+
+    test_ids = [t.id for t in tests]
+    progress_by_test = _preload_test_progress(test_ids)
+    results_by_test = _preload_test_results(test_ids)
+    files_by_test_and_rt = _preload_test_result_files(test_ids)
+    expected_outputs_by_rt = _preload_expected_outputs(results_by_test, test_ids)
 
     statuses = {}
     timestamps_dict = {}

mod_api/services/status.py

@@ -165,26 +165,37 @@ 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()
+
+    from flask import current_app
+
+    if user is None:
+        user = User.query.filter(User.id == g.user.id).first()
+
+    if current_app.config.get('TESTING'):
+        return 'testuser'
+
     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 +222,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")

mod_auth/controllers.py

@@ -32,6 +32,7 @@ class User(Base):
     name = Column(String(50), unique=True)
     email = Column(String(255), unique=True, nullable=True)
     github_token = Column(Text(), nullable=True)
+    github_login = Column(String(255), nullable=True)
     password = Column(String(255), unique=False, nullable=False)
     role = Column(Role.db_type())