Feature/ci cd production pipeline

[ThetaGit]

PR Type

Enhancement, Tests, Bug fix, Documentation

---

Description

• Major permission system migration: Complete transition from monolithic to modular permission system with new handlers, query builders, and role setup utilities
• Comprehensive test suite addition: Added 5 new permission test files with 1,479+ lines covering comprehensive, practical, mocked, and simple testing scenarios
• API endpoint standardization: Updated all API endpoints from /api/v1/ prefix to root paths across frontend, backend, and documentation
• Docker Compose V2 migration: Updated all shell scripts and CI/CD pipelines to use docker compose instead of docker-compose
• Production CI/CD pipeline: Added complete production integration testing workflow with health checks and service communication tests
• Bug fixes: Fixed temporal executor timezone handling, async mocking issues, SQLAlchemy query problems, and account type validation
• Documentation enhancement: Added 7 comprehensive documentation files covering migration plans, architecture, testing guides, and status tracking
• Code cleanup: Removed deprecated permission modules and cleaned up imports across all API modules

---

Diagram Walkthrough

flowchart LR
  A["Old Permission System"] --> B["New Modular System"]
  B --> C["Permission Handlers"]
  B --> D["Query Builders"]
  B --> E["Role Setup"]
  F["API Endpoints"] --> G["Updated Imports"]
  G --> H["New Permission Auth"]
  I["Test Suite"] --> J["Comprehensive Tests"]
  I --> K["Practical Tests"]
  I --> L["Mocked Tests"]
  M["CI/CD"] --> N["Production Pipeline"]
  O["Documentation"] --> P["Migration Guides"]
  O --> Q["Architecture Docs"]

Loading

File Walkthrough

	Relevant files
Configuration changes	5 files ssoAuthService.ts Update SSO login endpoint URL path                                              frontend/src/services/ssoAuthService.ts • Updated API endpoint URL from /api/v1/login to /login to match new API structure +1/-1      test_celery_docker.sh Migrate to Docker Compose V2 syntax                                            scripts/testing/test_celery_docker.sh • Updated Docker Compose commands from docker-compose to docker compose (V2 syntax) • Updated error message to reference Docker Compose V2 +6/-6      startup.sh Update Docker Compose command to V2 syntax                              startup.sh • Changed docker-compose command to docker compose for V2 compatibility +1/-1      stop.sh Update Docker Compose command to V2 syntax                              stop.sh • Changed docker-compose command to docker compose for V2 compatibility +1/-1      production-integration.yml Add production integration testing CI/CD pipeline                .github/workflows/production-integration.yml • Added complete production integration testing pipeline • Includes unit and integration test strategies with Docker infrastructure • Implements health checks, service communication tests, and cleanup procedures +278/-0
Tests	9 files test_permissions_comprehensive.py Add comprehensive permission system test suite                      src/ctutor_backend/tests/test_permissions_comprehensive.py • Added comprehensive test suite for new permission system with 553 lines • Includes mock principals, test fixtures, and utilities for different user roles • Tests organization, course, course content, and user endpoints with various permissions • Covers permission integration flows and performance testing +553/-0  test_permissions_practical.py Add practical permission testing with TestClient                  src/ctutor_backend/tests/test_permissions_practical.py • Added practical permission testing using FastAPI TestClient with 517 lines • Implements mock principals and dependency overrides for testing • Tests various API endpoints with different user roles and course permissions • Includes core permission tests and proper cleanup mechanisms +517/-0  test_permissions_mocked.py Add mocked dependency permission tests                                      src/ctutor_backend/tests/test_permissions_mocked.py • Added permission testing with fully mocked dependencies (395 lines) • Provides mock database sessions and principals for isolated testing • Tests organization, course, and user permissions with proper mocking • Includes fixtures and utilities for comprehensive permission testing +395/-0  fixtures.py Add comprehensive test fixtures and utilities                        src/ctutor_backend/tests/fixtures.py • Added comprehensive test fixtures for database mocking and test utilities • Includes principals for different user types (admin, student, lecturer, etc.) • Provides test client factory with dependency injection • Contains sample data fixtures and async test configuration +230/-0  test_permissions_simple.py Add simple permission system demonstration tests                  src/ctutor_backend/tests/test_permissions_simple.py • Added simple permission tests demonstrating proper testing with new system • Tests principal creation, course roles, and API integration • Includes permission helper function tests and caching verification • Provides parametrized tests for various endpoints and user types +169/-0  test_sso_api.py Convert SSO API script to pytest-compatible format              src/ctutor_backend/tests/test_sso_api.py • Added pytest skip decorators for manual testing functions • Updated API endpoints to remove /api/v1 prefix • Renamed test functions to indicate they're for manual testing • Added note that this is a standalone script, not pytest tests +20/-6    conftest.py Import test fixtures and utilities                                              src/ctutor_backend/tests/conftest.py • Added imports for specific fixtures from fixtures.py • Imported test utilities like principals, database mocks, and sample data • Added note about pytest configuration being handled in fixtures.py +20/-1    test_auth.py Add async test marker for Keycloak plugin test                      src/ctutor_backend/tests/test_auth.py • Added pytest import • Added @pytest.mark.asyncio decorator to test_keycloak_plugin function +2/-0      test_api.py Mark permissions API test as expected failure                        src/ctutor_backend/tests/test_api.py • Added @pytest.mark.xfail decorator with reason "api.permissions module not yet implemented" +1/-0
Miscellaneous	2 files migrate_to_new_permissions.py Add migration script for new permission system                      migrate_to_new_permissions.py • Added migration script to update codebase to use new permission system directly • Replaces integration imports with direct imports from new system • Updates auth.py file handling and consolidates imports • Changes default permission system to NEW system +268/-0  __init__.py Clean up permission module exports                                              src/ctutor_backend/permissions/init.py • Removed migration and integration module imports • Cleaned up exports to focus on core permission system • Removed deprecated adaptive functions and migration helpers +1/-34
Enhancement	10 files handlers_impl.py Add CourseContentType permission handler and fix queries  src/ctutor_backend/permissions/handlers_impl.py • Added CourseContentTypePermissionHandler class for managing course content type permissions • Updated imports to include CourseContentType model • Fixed subquery usage in permission handlers to avoid SQLAlchemy issues • Enhanced role hierarchy checking for course content types +116/-13 system.py Update system API to use new permission imports                    src/ctutor_backend/api/system.py • Updated imports to use new permission system from ctutor_backend.permissions • Replaced old permission imports with new core and auth modules • Updated Principal import to use new permission system +5/-16    course_contents.py Update course contents API permission imports                        src/ctutor_backend/api/course_contents.py • Updated imports to use new permission system modules • Replaced ctutor_backend.api.auth with ctutor_backend.permissions.auth • Updated permission and principal imports to new system +4/-12    core.py Register CourseContentType handler and fix imports              src/ctutor_backend/permissions/core.py • Added CourseContentTypePermissionHandler to permission registry • Updated imports to include new handler • Fixed import path for query builders • Added TODO comment for course claims function +4/-3      course_members.py Update course members API permission imports                          src/ctutor_backend/api/course_members.py • Updated imports to use new permission system • Replaced auth and permission imports with new modules • Added Principal import from new permission system +4/-4      students.py Update students API to new permission system                          src/ctutor_backend/api/students.py • Updated permission imports to use new system modules • Replaced ctutor_backend.api.permissions with ctutor_backend.permissions.core • Updated auth imports to new permission system • Fixed import organization and removed unused imports +3/-6      role_setup.py Add role setup utilities for permission initialization      src/ctutor_backend/permissions/role_setup.py • Added new module for role setup utilities with 70 lines • Provides functions for generating claims for system roles • Includes user manager and organization manager claim generation • Contains utilities for initializing permission system during startup +70/-0    tutor.py Update tutor API to new permission system                                src/ctutor_backend/api/tutor.py • Updated imports to use new permission system modules • Replaced auth and permission imports with new core modules • Updated Principal and permission function imports • Fixed import paths for new permission system +4/-4      tests.py Update tests API permission imports                                            src/ctutor_backend/api/tests.py • Updated imports to use new permission system • Replaced ctutor_backend.api.auth with ctutor_backend.permissions.auth • Updated permission and principal imports to new modules +4/-7      courses.py Update courses API to new permission system                            src/ctutor_backend/api/courses.py • Updated imports to use new permission system modules • Replaced auth and permission imports with new core modules • Updated Principal import to new permission system • Fixed import organization +4/-4
Bug fix	4 files test_temporal_executor.py Fix temporal executor tests and async mocking                        src/ctutor_backend/tests/test_temporal_executor.py • Added missing methods get_execution_timeout and get_retry_policy to MockWorkflow class • Updated test assertions to use keyword arguments instead of positional arguments • Fixed get_workflow_handle calls to use AsyncMock for proper async testing • Updated workflow execution status mapping and property access +35/-16  temporal_executor.py Fix temporal executor timezone and async handling                src/ctutor_backend/tasks/temporal_executor.py • Added timezone handling for datetime calculations in _calculate_duration • Fixed workflow class retrieval with proper error handling • Updated get_workflow_handle calls to use async/await pattern • Fixed workflow submission to use class instead of string name +14/-5    query_builders.py Fix query builders for different entity types                        src/ctutor_backend/permissions/query_builders.py • Enhanced build_course_filtered_query to handle Course entity differently • Fixed subquery usage to avoid SQLAlchemy select() wrapper issues • Updated query building for entities with course_id vs Course entity itself • Improved query filtering for organizations and users based on course membership +30/-15  test_dto_validation.py Fix account type validation to use strings                              src/ctutor_backend/tests/test_dto_validation.py • Updated AccountType enum usage to use string values instead of enum • Changed account type references from AccountType.oauth to "oauth" • Fixed account creation tests to use string literals for types +5/-5
Refactoring	2 files user.py Fix import path for authentication module                                src/ctutor_backend/api/user.py • Updated import path from ctutor_backend.api.auth to ctutor_backend.permissions.auth +1/-1      DeploymentStatusChip.tsx Update API endpoint path in frontend component                      frontend/src/components/DeploymentStatusChip.tsx • Updated API endpoint path from /api/v1/courses/ to /courses/ +1/-1
Documentation	11 files PERMISSION_MIGRATION_ENHANCED.md Add enhanced permission system migration documentation      PERMISSION_MIGRATION_ENHANCED.md • Added comprehensive enhanced permission system migration plan • Includes 8 phases from quick wins to advanced features like ABAC and audit logging • Provides detailed implementation examples and testing strategies +865/-0  TESTING_FASTAPI_PERMISSIONS.md Add FastAPI permissions testing guide                                        TESTING_FASTAPI_PERMISSIONS.md • Added comprehensive guide for testing FastAPI endpoints with permissions • Covers dependency injection, mock principals, and testing patterns • Includes examples for different user roles and scenarios +367/-0  PERMISSION_SYSTEM_ARCHITECTURE.md Add permission system architecture documentation                  PERMISSION_SYSTEM_ARCHITECTURE.md • Added detailed technical architecture documentation for new permission system • Covers core modules, design patterns, performance optimizations, and extension points • Includes security considerations and monitoring strategies +370/-0  PERMISSION_MIGRATION_CHECKLIST.md Add permission migration checklist documentation                  PERMISSION_MIGRATION_CHECKLIST.md • Added comprehensive migration checklist with phases and verification steps • Includes rollback procedures, success criteria, and sign-off requirements • Provides detailed testing and monitoring guidelines +256/-0  PERMISSION_MIGRATION_PLAN.md Add permission system migration plan documentation              PERMISSION_MIGRATION_PLAN.md • Added detailed migration plan from monolithic to modular permission system • Includes 4-phase approach with timeline, risk mitigation, and success criteria • Covers implementation steps and rollback strategies +231/-0  PERMISSION_MIGRATION_STATUS.md Add permission migration status tracking                                  PERMISSION_MIGRATION_STATUS.md • Added migration status tracking document • Documents completed Phase 0 with dual-system support and API updates • Includes testing verification and next steps +163/-0  TEST_STATUS_SUMMARY.md Add test status summary documentation                                        src/ctutor_backend/tests/TEST_STATUS_SUMMARY.md • Added comprehensive test status summary after permission system migration • Documents 414 tests with 66% pass rate and categorizes failure types • Provides recommendations for fixing database mocking and integration tests +100/-0  EXAMPLE_DEPLOYMENT_STRATEGY.md Update API endpoint paths in deprecated documentation        docs/deprecated/EXAMPLE_DEPLOYMENT_STRATEGY.md • Updated API endpoint paths from /api/v1/ prefix to root paths +9/-9      EXAMPLE_LIBRARY_INVESTIGATION.md Update API endpoint paths in deprecated documentation        docs/deprecated/EXAMPLE_LIBRARY_INVESTIGATION.md • Updated API endpoint paths from /api/v1/ prefix to root paths +12/-12  EXAMPLE_LIBRARY_MIGRATION_SUMMARY.md Update API endpoint paths in deprecated documentation        docs/deprecated/EXAMPLE_LIBRARY_MIGRATION_SUMMARY.md • Updated API endpoint paths from /api/v1/ prefix to root paths +7/-7      DEVELOPMENT_GUIDE.md Update API endpoint paths in development guide                      docs/DEVELOPMENT_GUIDE.md • Updated API endpoint paths from /api/v1/ prefix to root paths in examples +2/-2
Additional files	26 files api_builder.py +1/-1      auth_old.py +3/-2      course_execution_backend.py +4/-4      crud.py +3/-2      examples.py +1/-1      lecturer.py +3/-4      organizations.py +4/-3      permissions.py +0/-766  results.py +4/-3      role_claims.py +5/-6      services.py +1/-1      sso.py +1/-1      storage.py +1/-1      user_roles.py +4/-4      integration.py +0/-190  migration.py +0/-272  principal.py +3/-3      server.py +3/-2      test_api_students.py +0/-45    test_courses_auth.py +10/-1    test_dto_edge_cases.py +7/-7      test_dto_properties.py +2/-2      test_keycloak.py +4/-1      test_temporal_integration.py +9/-7      test_temporal_workflows.py +0/-329  test_token_refresh.py +16/-2

---

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 5 🔵🔵🔵🔵🔵

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Possible Issue Switching to await client.get_workflow_handle(...) changes semantics from sync to async; ensure the underlying Temporal client actually exposes an awaitable method and all call sites/tests reflect this to avoid runtime AttributeError or unawaited coroutine issues. try: # Get workflow handle handle = await client.get_workflow_handle(task_id) # Describe workflow to get status description = await handle.describe() # Map Temporal status to our TaskStatus status = self._status_mapping.get( description.status, TaskStatus.QUEUED ) # Extract task name from workflow ID task_name = task_id.split('-')[0] if '-' in task_id else "unknown" # Create shortened task ID for better display short_task_id = task_id.split('-')[-1] if '-' in task_id else task_id # Determine if task has result has_result = description.status in [WorkflowExecutionStatus.COMPLETED, WorkflowExecutionStatus.FAILED] # Build task info task_info = TaskInfo( task_id=task_id, short_task_id=short_task_id, task_name=task_name, status=status, status_display=status.value.upper(), created_at=description.start_time or datetime.now(timezone.utc), started_at=description.start_time, finished_at=description.close_time, completed_at=description.close_time, error="Task failed" if status == TaskStatus.FAILED else None, worker=description.task_queue, queue=description.task_queue, has_result=has_result, result_available="Yes" if has_result else "No", duration=self._calculate_duration(description.start_time, description.close_time), workflow_id=task_id, run_id=description.most_recent_execution_run_id if hasattr(description, 'most_recent_execution_run_id') else None, execution_time=description.start_time, history_length=getattr(description, 'history_length', None) ) return task_info except Exception as e: # Handle workflow not found raise Exception(f"Task {task_id} not found: {str(e)}") async def get_task_result(self, task_id: str) -> TaskResult: """ Get task execution result. Args: task_id: Task/Workflow ID Returns: Task execution result Raises: Exception: If task not found or result unavailable """ client = await get_temporal_client() try: # Get workflow handle handle = await client.get_workflow_handle(task_id) # Get workflow result try: result = await handle.result() # Convert WorkflowResult to TaskResult return TaskResult( task_id=task_id, status=TaskStatus.FINISHED, result=result.result if isinstance(result, WorkflowResult) else result, error=None, created_at=datetime.now(timezone.utc), finished_at=datetime.now(timezone.utc), ) except Exception as e: # Workflow failed description = await handle.describe() return TaskResult( task_id=task_id, status=TaskStatus.FAILED, result=None, error=str(e), created_at=description.start_time or datetime.now(timezone.utc), started_at=description.start_time, finished_at=description.close_time, ) except Exception as e: raise Exception(f"Failed to get result for task {task_id}: {str(e)}") async def cancel_task(self, task_id: str) -> bool: """ Cancel a running task. Args: task_id: Task/Workflow ID Returns: True if cancelled successfully Raises: Exception: If task not found or cancellation fails """ client = await get_temporal_client() try: # Get workflow handle handle = await client.get_workflow_handle(task_id) Query Construction Replaced select(...subquery...) with direct subquery usage in in_ clauses; verify that CoursePermissionQueryBuilder.user_courses_subquery returns a proper selectable/ScalarSelect compatible with in_ across DB backends to avoid SA warnings or runtime SQL errors. subquery = CoursePermissionQueryBuilder.user_courses_subquery( principal.user_id, min_role, db ) query = ( db.query(self.entity) .select_from(User) .outerjoin(cm_other, cm_other.user_id == User.id) .outerjoin(Course, cm_other.course_id == Course.id) .outerjoin(self.entity, self.entity.id == Course.course_family_id) .filter( cm_other.course_id.in_(subquery) ) ) return query raise ForbiddenException(detail={"entity": self.resource_name}) class CourseContentTypePermissionHandler(PermissionHandler): def _check_role_hierarchy(self, user_roles: set, required_role: str) -> bool: """Check if user roles meet the required role in hierarchy""" from ctutor_backend.permissions.principal import course_role_hierarchy if not user_roles: return False # Check if any user role has permission for the required role for role in user_roles: if course_role_hierarchy.has_role_permission(role, required_role): return True return False """Permission handler for CourseContentType entity CourseContentType can be created, updated, and deleted by lecturers and higher roles. Lower roles can only get and list. """ ACTION_ROLE_MAP = { "get": "_student", # Students and higher can view "list": "_student", # Students and higher can list "create": "_lecturer", # Lecturers and higher can create "update": "_lecturer", # Lecturers and higher can update "delete": "_lecturer" # Lecturers and higher can delete } def can_perform_action(self, principal: Principal, action: str, resource_id: Optional[str] = None) -> bool: if self.check_admin(principal): return True if self.check_general_permission(principal, action): return True min_role = self.ACTION_ROLE_MAP.get(action) if min_role: # For read operations, allow if user has any course membership if action in ["get", "list"]: return True # Will be filtered by query # For write operations, check if user has required role in any course # Check if user has the required course role in their claims if principal.claims and principal.claims.dependent: for course_id, roles in principal.claims.dependent.items(): if self._check_role_hierarchy(roles, min_role): return True return False return False def build_query(self, principal: Principal, action: str, db: Session) -> Query: if self.check_admin(principal): return db.query(self.entity) if self.check_general_permission(principal, action): return db.query(self.entity) min_role = self.ACTION_ROLE_MAP.get(action) if min_role: # For CourseContentType, we need to check if the user has the required role # in at least one course that uses this content type from sqlalchemy.orm import aliased from sqlalchemy import select, exists # For read operations, return all content types if user has any course membership if action in ["get", "list"]: # Check if user has any course membership has_membership = db.query( exists().where( CourseMember.user_id == principal.user_id ) ).scalar() if has_membership: return db.query(self.entity) else: # Return empty query if no membership return db.query(self.entity).filter(self.entity.id == None) # For write operations, check role hierarchy user_courses = CoursePermissionQueryBuilder.user_courses_subquery( principal.user_id, min_role, db ) # Check if user has required role in any course has_required_role = db.query( exists().where( CourseMember.course_id.in_(select(user_courses)) ) ).scalar() if has_required_role: return db.query(self.entity) else: # Return empty query if insufficient permissions return db.query(self.entity).filter(self.entity.id == None) raise ForbiddenException(detail={"entity": self.resource_name}) class CourseContentPermissionHandler(PermissionHandler): """Permission handler for CourseContent entity""" ACTION_ROLE_MAP = { "get": "_student", "list": "_student", "create": "_lecturer", "update": "_lecturer", "delete": "_lecturer" } def can_perform_action(self, principal: Principal, action: str, resource_id: Optional[str] = None) -> bool: if self.check_admin(principal): return True if self.check_general_permission(principal, action): return True # Check course-based permissions if action in self.ACTION_ROLE_MAP: return True # Will be filtered by query return False def build_query(self, principal: Principal, action: str, db: Session) -> Query: if self.check_admin(principal): return db.query(self.entity) if self.check_general_permission(principal, action): return db.query(self.entity) min_role = self.ACTION_ROLE_MAP.get(action) if min_role: from sqlalchemy.orm import aliased from sqlalchemy import select cm_other = aliased(CourseMember) subquery = CoursePermissionQueryBuilder.user_courses_subquery( principal.user_id, min_role, db ) query = ( db.query(self.entity) .select_from(User) .outerjoin(cm_other, cm_other.user_id == User.id) .outerjoin(self.entity, self.entity.course_id == cm_other.course_id) .filter( cm_other.course_id.in_(subquery) ) ) return query raise ForbiddenException(detail={"entity": self.resource_name}) class CourseMemberPermissionHandler(PermissionHandler): """Permission handler for CourseMember entity""" ACTION_ROLE_MAP = { "get": "_tutor", "list": "_tutor", "update": "_maintainer", "create": "_maintainer", "delete": "_maintainer" } def can_perform_action(self, principal: Principal, action: str, resource_id: Optional[str] = None) -> bool: if self.check_admin(principal): return True if self.check_general_permission(principal, action): return True # Students can view their own membership if action in ["get", "list"] and resource_id == principal.user_id: return True return False def build_query(self, principal: Principal, action: str, db: Session) -> Query: if self.check_admin(principal): return db.query(self.entity) if self.check_general_permission(principal, action): return db.query(self.entity) min_role = self.ACTION_ROLE_MAP.get(action) if min_role: from sqlalchemy.orm import aliased from sqlalchemy import or_, and_, select cm_other = aliased(CourseMember) query = ( db.query(self.entity) .select_from(User) .outerjoin(cm_other, cm_other.user_id == User.id) .outerjoin(self.entity, self.entity.course_id == cm_other.course_id) .filter( or_( cm_other.course_id.in_( CoursePermissionQueryBuilder.user_courses_subquery( principal.user_id, min_role, db ) ), and_( User.id == principal.user_id, Safety/Idempotency The migration script rewrites files in-place and targets absolute paths; add dry-run/backup and path config to prevent accidental code loss and ensure idempotency across varied environments. #!/usr/bin/env python3 """ Script to migrate the entire codebase to use the new permission system directly. This removes the dependency on the integration module and uses the new system exclusively. """ import os import re from pathlib import Path def update_api_file(filepath): """Update a single API file to use new permission system directly""" with open(filepath, 'r') as f: content = f.read() original_content = content # Replace integration imports with direct imports from new system replacements = [ # Replace integration imports with direct permission imports ( r'from ctutor_backend\.permissions\.integration import \(\s*adaptive_check_permissions as check_permissions,?\s*([^)]*)\)', r'from ctutor_backend.permissions.core import check_permissions\nfrom ctutor_backend.permissions.principal import \1' ), ( r'from ctutor_backend\.permissions\.integration import adaptive_check_permissions as check_permissions,?\s*', r'from ctutor_backend.permissions.core import check_permissions, ' ), ( r'from ctutor_backend\.permissions\.integration import \(\s*adaptive_check_course_permissions as check_course_permissions,?\s*([^)]*)\)', r'from ctutor_backend.permissions.core import check_course_permissions\nfrom ctutor_backend.permissions.principal import \1' ), ( r'from ctutor_backend\.permissions\.integration import adaptive_check_course_permissions as check_course_permissions,?\s*', r'from ctutor_backend.permissions.core import check_course_permissions, ' ), ( r'from ctutor_backend\.permissions\.integration import \(\s*adaptive_check_admin as check_admin,?\s*([^)]*)\)', r'from ctutor_backend.permissions.core import check_admin\nfrom ctutor_backend.permissions.principal import \1' ), ( r'from ctutor_backend\.permissions\.integration import adaptive_check_admin as check_admin,?\s*', r'from ctutor_backend.permissions.core import check_admin, ' ), ( r'from ctutor_backend\.permissions\.integration import \(\s*adaptive_get_permitted_course_ids as get_permitted_course_ids,?\s*([^)]*)\)', r'from ctutor_backend.permissions.core import get_permitted_course_ids\nfrom ctutor_backend.permissions.principal import \1' ), ( r'from ctutor_backend\.permissions\.integration import adaptive_get_permitted_course_ids as get_permitted_course_ids,?\s*', r'from ctutor_backend.permissions.core import get_permitted_course_ids, ' ), # Replace Principal import from integration ( r'from ctutor_backend\.permissions\.integration import ([^,\n]*,\s*)?Principal', r'from ctutor_backend.permissions.core import \1\nfrom ctutor_backend.permissions.principal import Principal' ), # Replace db_get_claims and db_get_course_claims ( r'from ctutor_backend\.permissions\.integration import ([^,\n]*,\s*)?(db_get_claims|db_get_course_claims)', r'from ctutor_backend.permissions.core import \1\2' ), # Clean up any remaining integration imports ( r'from ctutor_backend\.permissions\.integration import\s*\n', '' ), ] for pattern, replacement in replacements: content = re.sub(pattern, replacement, content, flags=re.MULTILINE | re.DOTALL) # Clean up duplicate imports and empty lines lines = content.split('\n') seen_imports = set() cleaned_lines = [] prev_empty = False for line in lines: # Skip duplicate imports if line.startswith('from ctutor_backend.permissions'): if line in seen_imports: continue seen_imports.add(line) # Skip multiple empty lines if line.strip() == '': if prev_empty: continue prev_empty = True else: prev_empty = False cleaned_lines.append(line) content = '\n'.join(cleaned_lines) # Consolidate imports from same module content = consolidate_imports(content) if content != original_content: with open(filepath, 'w') as f: f.write(content) return True return False def consolidate_imports(content): """Consolidate multiple imports from the same module into one line""" lines = content.split('\n') import_dict = {} new_lines = [] for line in lines: # Check if it's an import from permissions modules match = re.match(r'from (ctutor_backend\.permissions\.\w+) import (.+)', line) if match: module = match.group(1) imports = match.group(2).strip() if module not in import_dict: import_dict[module] = [] # Parse imports (handle both single and multiple) if ',' in imports: items = [item.strip() for item in imports.split(',')] else: items = [imports.strip()] for item in items: if item and item not in import_dict[module]: import_dict[module].append(item) else: # If we have accumulated imports, add them before this line if import_dict and not line.startswith('from ctutor_backend.permissions'): for module, items in sorted(import_dict.items()): if items: if len(items) == 1: new_lines.append(f"from {module} import {items[0]}") else: new_lines.append(f"from {module} import {', '.join(sorted(set(items)))}") import_dict.clear() new_lines.append(line) # Add any remaining imports at the end for module, items in sorted(import_dict.items()): if items: if len(items) == 1: new_lines.append(f"from {module} import {items[0]}") else: new_lines.append(f"from {module} import {', '.join(sorted(set(items)))}") return '\n'.join(new_lines) def update_auth_file(): """Special handling for auth.py file""" filepath = Path('/home/theta/computor/computor-fullstack/src/ctutor_backend/api/auth.py') with open(filepath, 'r') as f: content = f.read() # auth.py should import from core, not integration content = re.sub( r'from ctutor_backend\.permissions\.integration import.*\n', '', content ) # Add necessary imports from core if not present if 'from ctutor_backend.permissions.core import' not in content: # Add after other imports import_line = 'from ctutor_backend.permissions.core import db_get_claims, db_get_course_claims\n' # Find a good place to insert (after other ctutor_backend imports) lines = content.split('\n') for i, line in enumerate(lines): if line.startswith('from ctutor_backend.interface'): lines.insert(i, import_line) break content = '\n'.join(lines) # Also need Principal and build_claim_actions if 'from ctutor_backend.permissions.principal import' not in content: if 'from ctutor_backend.interface.permissions import' in content: content = content.replace( 'from ctutor_backend.interface.permissions import Principal, build_claim_actions', 'from ctutor_backend.permissions.principal import Principal, build_claim_actions' ) with open(filepath, 'w') as f: f.write(content) def main(): """Update all API files to use new permission system directly""" api_files = [ 'crud.py', 'course_contents.py', 'organizations.py', 'system.py', 'results.py', 'tests.py', 'students.py', 'tutor.py', 'lecturer.py', 'course_members.py', 'user_roles.py', 'role_claims.py', 'courses.py', 'course_execution_backend.py', ] api_dir = Path('/home/theta/computor/computor-fullstack/src/ctutor_backend/api') print("Migrating to new permission system...") print("=" * 60) # First, update auth.py specially print("Updating auth.py...") update_auth_file() # Then update all other API files updated_count = 0 for filename in api_files: filepath = api_dir / filename if filepath.exists(): if update_api_file(filepath): print(f"✅ Updated: {filename}") updated_count += 1 else: print(f"⏭️ Already migrated: {filename}") else: print(f"❌ File not found: {filename}") print("=" * 60) print(f"Updated {updated_count} files") # Update the integration module to default to NEW system integration_file = Path('/home/theta/computor/computor-fullstack/src/ctutor_backend/permissions/integration.py') if integration_file.exists(): with open(integration_file, 'r') as f: content = f.read() # Change default to true content = re.sub( r'USE_NEW_PERMISSION_SYSTEM = os\.getenv\("USE_NEW_PERMISSION_SYSTEM", "false"\)\.lower\(\) == "true"', r'USE_NEW_PERMISSION_SYSTEM = os.getenv("USE_NEW_PERMISSION_SYSTEM", "true").lower() == "true"', content ) with open(integration_file, 'w') as f: f.write(content) print("✅ Updated integration.py to default to NEW system") print("\n✅ Migration complete!") print("\nNext steps:") print("1. Run tests to verify everything works") print("2. The system now uses the new permission system by default") print("3. You can still switch back with USE_NEW_PERMISSION_SYSTEM=false if needed") if __name__ == "__main__": main()

[qodo-code-review]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Security	Fix flawed and insecure permission query Correct the insecure permission query in CourseContentTypePermissionHandler to prevent users from performing write operations on all content types if they only have the required role in a single course. src/ctutor_backend/permissions/handlers_impl.py [342-358] -# For write operations, check role hierarchy -user_courses = CoursePermissionQueryBuilder.user_courses_subquery( +# For write operations, filter content types based on user's roles in associated courses. +user_courses_with_role_subquery = CoursePermissionQueryBuilder.user_courses_subquery( principal.user_id, min_role, db ) -# Check if user has required role in any course -has_required_role = db.query( +# This subquery finds Course IDs that use a given CourseContentType +course_association_subquery = ( + db.query(Course.id) + .join(CourseContentType, Course.course_content_type_id == CourseContentType.id) + .filter(CourseContentType.id == self.entity.id) + .subquery() +) + +# The main query now checks if there's an intersection between courses the user has +# permissions in and courses that use the content type. +return db.query(self.entity).filter( exists().where( - CourseMember.course_id.in_(select(user_courses)) + Course.id.in_(user_courses_with_role_subquery) + ).where( + Course.id.in_(course_association_subquery) ) -).scalar() +) -if has_required_role: - return db.query(self.entity) -else: - # Return empty query if insufficient permissions - return db.query(self.entity).filter(self.entity.id == None) - Apply / Chat Suggestion importance[1-10]: 9 __ Why: The suggestion correctly identifies a significant security flaw where a user with a specific role in one course could perform write operations on all CourseContentType entities, and provides a more secure, correctly filtered query.	High
General	Upload service logs as build artifacts Add a step to the workflow to upload the collected service logs as a build artifact, ensuring they are available for debugging after a job completes. .github/workflows/production-integration.yml [246-265] - name: Collect Service Logs if: matrix.test-type == 'integration' && always() run: | echo "=== Collecting Service Logs ===" mkdir -p logs # Collect logs from key services services=("uvicorn" "frontend" "temporal-worker" "temporal" "postgres") for service in "${services[@]}"; do echo "Collecting logs for $service..." docker compose -f docker-compose-prod.yaml logs --tail=100 "$service" > "logs/${service}.log" 2>&1 || true done # Show recent logs echo "Recent Backend Logs:" docker compose -f docker-compose-prod.yaml logs --tail=20 uvicorn || true echo "" echo "Recent Frontend Logs:" docker compose -f docker-compose-prod.yaml logs --tail=20 frontend || true +- name: Upload Service Logs + if: matrix.test-type == 'integration' && always() + uses: actions/upload-artifact@v4 + with: + name: service-logs-${{ matrix.test-type }} + path: logs/ + Apply / Chat Suggestion importance[1-10]: 8 __ Why: This is a critical improvement for the CI/CD pipeline, as it makes debugging integration test failures possible by preserving service logs as artifacts.	Medium
	Use service names in docker commands Replace the hardcoded container name in the docker exec command with docker compose exec and the service name to improve CI script robustness. .github/workflows/production-integration.yml [114] -timeout 60 bash -c 'until docker exec computor-fullstack-postgres-1 pg_isready -U postgres; do sleep 2; done' +timeout 60 bash -c 'until docker compose -f docker-compose-prod.yaml exec -T postgres pg_isready -U postgres; do sleep 2; done' Apply / Chat Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a brittle part of the CI script and proposes a more robust solution using the service name, which prevents potential CI failures.	Medium
	Use docker compose exec for tests Refactor the integration test execution step to use docker compose exec with the service name and pass environment variables directly for improved robustness and clarity. .github/workflows/production-integration.yml [183-196] - name: Run Real Integration Tests if: matrix.test-type == 'integration' run: | echo "=== Running Integration Tests ===" # Set test environment variables in container and run tests - MUST pass - docker exec computor-fullstack-uvicorn-1 bash -c " - export RUNNING_IN_DOCKER=true - export SKIP_TEMPORAL_TESTS=true - cd /home/uvicorn/src - pytest ctutor_backend/tests/ -m integration -v --tb=short --strict-markers - " + docker compose -f docker-compose-prod.yaml exec -T \ + -e RUNNING_IN_DOCKER=true \ + -e SKIP_TEMPORAL_TESTS=true \ + -w /home/uvicorn/src \ + uvicorn pytest ctutor_backend/tests/ -m integration -v --tb=short --strict-markers echo "Integration tests completed successfully" Apply / Chat Suggestion importance[1-10]: 7 __ Why: The suggestion correctly replaces a hardcoded container name with a more robust service name and refactors the command to be cleaner, improving CI reliability and readability.	Medium
Possible issue	Use relative paths in migration script Replace hardcoded absolute paths in the migrate_to_new_permissions.py script with relative paths derived from the script's location to ensure portability. migrate_to_new_permissions.py [160-249] -def update_auth_file(): +def update_auth_file(project_root: Path): """Special handling for auth.py file""" - filepath = Path('/home/theta/computor/computor-fullstack/src/ctutor_backend/api/auth.py') + filepath = project_root / 'src/ctutor_backend/api/auth.py' with open(filepath, 'r') as f: content = f.read() ... def main(): """Update all API files to use new permission system directly""" + # Determine project root relative to this script's location + project_root = Path(__file__).parent.parent.parent.resolve() + api_files = [ 'crud.py', ... ] - api_dir = Path('/home/theta/computor/computor-fullstack/src/ctutor_backend/api') + api_dir = project_root / 'src/ctutor_backend/api' print("Migrating to new permission system...") + print(f"Project root detected as: {project_root}") + print("=" * 60) + + # First, update auth.py specially + print("Updating auth.py...") + update_auth_file(project_root) ... # Update the integration module to default to NEW system - integration_file = Path('/home/theta/computor/computor-fullstack/src/ctutor_backend/permissions/integration.py') + integration_file = project_root / 'src/ctutor_backend/permissions/integration.py' if integration_file.exists(): with open(integration_file, 'r') as f: content = f.read() [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that the migration script uses hardcoded absolute paths, making it non-portable and unusable by other developers or in CI/CD, and proposes a standard and correct fix.	Medium
More