init app of test arena

asf-cloud-server/testarena/backend/app/auth.py

@@ -0,0 +1,59 @@
+from datetime import datetime, timedelta
+from typing import Optional
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from fastapi.security import OAuth2PasswordBearer
+from fastapi import Depends, HTTPException, status
+from sqlalchemy.orm import Session
+from . import crud, models, schemas
+from .dependencies import get_db
+
+SECRET_KEY = "YOUR_SECRET_KEY_HERE_PLEASE_CHANGE_IN_PROD"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
+
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = schemas.TokenData(username=username)
+    except JWTError:
+        raise credentials_exception
+    user = crud.get_user_by_username(db, username=token_data.username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+def get_current_active_user(current_user: models.User = Depends(get_current_user)):
+    return current_user
+
+def get_current_admin_user(current_user: models.User = Depends(get_current_user)):
+    if current_user.role != models.UserRole.admin:
+        raise HTTPException(status_code=400, detail="Not enough permissions")
+    return current_user

asf-cloud-server/testarena/backend/app/crud.py

@@ -0,0 +1,52 @@
+from sqlalchemy.orm import Session
+from . import models, schemas
+
+def get_user(db: Session, user_id: int):
+    return db.query(models.User).filter(models.User.id == user_id).first()
+
+def get_user_by_username(db: Session, username: str):
+    return db.query(models.User).filter(models.User.username == username).first()
+
+def get_users(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(models.User).offset(skip).limit(limit).all()
+
+def create_user(db: Session, user: schemas.UserCreate, hashed_password: str):
+    db_user = models.User(username=user.username, hashed_password=hashed_password, role=user.role)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+def delete_user(db: Session, user_id: int):
+    db_user = db.query(models.User).filter(models.User.id == user_id).first()
+    if db_user:
+        db.delete(db_user)
+        db.commit()
+    return db_user
+
+def get_jobs(db: Session, skip: int = 0, limit: int = 100, user_id: int = None):
+    if user_id:
+        return db.query(models.Job).filter(models.Job.user_id == user_id).order_by(models.Job.created_at.desc()).offset(skip).limit(limit).all()
+    return db.query(models.Job).order_by(models.Job.created_at.desc()).offset(skip).limit(limit).all()
+
+def get_job(db: Session, job_id: int):
+    return db.query(models.Job).filter(models.Job.id == job_id).first()
+
+def create_job(db: Session, job: schemas.JobCreate, user_id: int):
+    db_job = models.Job(**job.dict(), user_id=user_id)
+    db.add(db_job)
+    db.commit()
+    db.refresh(db_job)
+    return db_job
+
+def update_job_status(db: Session, job_id: int, status: str, result_path: str = None, duration: str = None):
+    job = db.query(models.Job).filter(models.Job.id == job_id).first()
+    if job:
+        job.status = status
+        if result_path:
+            job.result_path = result_path
+        if duration:
+            job.duration = duration
+        db.commit()
+        db.refresh(job)
+    return job

asf-cloud-server/testarena/backend/app/database.py

@@ -0,0 +1,11 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+import os
+
+SQLALCHEMY_DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://user:password@db/testarena")
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

asf-cloud-server/testarena/backend/app/dependencies.py

@@ -0,0 +1,8 @@
+from .database import SessionLocal
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

asf-cloud-server/testarena/backend/app/main.py

@@ -0,0 +1,165 @@
+from fastapi import FastAPI, Depends, HTTPException, status, WebSocket, WebSocketDisconnect, BackgroundTasks
+from fastapi.middleware.cors import CORSMiddleware
+from sqlalchemy.orm import Session
+from typing import List
+from . import models, schemas, crud, auth, database, tasks
+from .socket_manager import manager
+from .dependencies import get_db
+
+models.Base.metadata.create_all(bind=database.engine)
+
+app = FastAPI(title="ASF TestArena")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"], # In production, set to specific domain
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.post("/auth/login", response_model=schemas.Token)
+def login_for_access_token(form_data: schemas.UserCreate, db: Session = Depends(get_db)):
+    user = crud.get_user_by_username(db, form_data.username)
+    if not user or not auth.verify_password(form_data.password, user.hashed_password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = auth.timedelta(minutes=auth.ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = auth.create_access_token(
+        data={"sub": user.username}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+@app.post("/auth/reset-password")
+def reset_password(
+    username: str, 
+    new_password: str, 
+    current_user: models.User = Depends(auth.get_current_admin_user), 
+    db: Session = Depends(get_db)
+):
+    user = crud.get_user_by_username(db, username)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    user.hashed_password = auth.get_password_hash(new_password)
+    db.commit()
+    return {"message": "Password reset successfully"}
+
+@app.get("/admin/users", response_model=List[schemas.User])
+def read_users(
+    skip: int = 0, 
+    limit: int = 100, 
+    current_user: models.User = Depends(auth.get_current_admin_user), 
+    db: Session = Depends(get_db)
+):
+    users = crud.get_users(db, skip=skip, limit=limit)
+    return users
+
+@app.post("/admin/users", response_model=schemas.User)
+def create_user(
+    user: schemas.UserCreate, 
+    current_user: models.User = Depends(auth.get_current_admin_user), 
+    db: Session = Depends(get_db)
+):
+    db_user = crud.get_user_by_username(db, username=user.username)
+    if db_user:
+        raise HTTPException(status_code=400, detail="Username already registered")
+    hashed_password = auth.get_password_hash(user.password)
+    return crud.create_user(db=db, user=user, hashed_password=hashed_password)
+
+@app.delete("/admin/users/{user_id}")
+def delete_user(
+    user_id: int, 
+    current_user: models.User = Depends(auth.get_current_admin_user), 
+    db: Session = Depends(get_db)
+):
+    crud.delete_user(db, user_id)
+    return {"message": "User deleted"}
+
+@app.get("/jobs", response_model=List[schemas.Job])
+def read_jobs(
+    skip: int = 0, 
+    limit: int = 100, 
+    current_user: models.User = Depends(auth.get_current_active_user), 
+    db: Session = Depends(get_db)
+):
+    if current_user.role == models.UserRole.admin:
+        jobs = crud.get_jobs(db, skip=skip, limit=limit)
+    else:
+        jobs = crud.get_jobs(db, skip=skip, limit=limit, user_id=current_user.id)
+    return jobs
+
+@app.get("/jobs/{job_id}", response_model=schemas.Job)
+def read_job(
+    job_id: int, 
+    current_user: models.User = Depends(auth.get_current_active_user), 
+    db: Session = Depends(get_db)
+):
+    job = crud.get_job(db, job_id=job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="Job not found")
+    if current_user.role != models.UserRole.admin and job.user_id != current_user.id:
+        raise HTTPException(status_code=403, detail="Not authorized to view this job")
+    return job
+
+@app.post("/jobs/submit", response_model=schemas.Job)
+def submit_job(
+    job: schemas.JobCreate, 
+    background_tasks: BackgroundTasks,
+    current_user: models.User = Depends(auth.get_current_active_user), 
+    db: Session = Depends(get_db)
+):
+    db_job = crud.create_job(db=db, job=job, user_id=current_user.id)
+    background_tasks.add_task(tasks.run_job_task, db_job.id)
+    return db_job
+
+@app.post("/jobs/{job_id}/abort")
+def abort_job(
+    job_id: int, 
+    current_user: models.User = Depends(auth.get_current_active_user), 
+    db: Session = Depends(get_db)
+):
+    job = crud.get_job(db, job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    if current_user.role != models.UserRole.admin and job.user_id != current_user.id:
+        raise HTTPException(status_code=403, detail="Not authorized")
+    
+    crud.update_job_status(db, job_id, "aborted")
+    return {"message": "Job aborted"}
+
+@app.post("/jobs/scenarios")
+def get_scenarios_endpoint(
+    branch_name: str,
+    current_user: models.User = Depends(auth.get_current_active_user)
+):
+    scenarios = tasks.get_scenarios(branch_name)
+    return scenarios
+
+@app.websocket("/ws/jobs")
+async def websocket_endpoint(websocket: WebSocket):
+    await manager.connect(websocket)
+    try:
+        while True:
+            await websocket.receive_text()
+    except WebSocketDisconnect:
+        manager.disconnect(websocket)
+
+import asyncio
+
+@app.on_event("startup")
+async def startup_event():
+    # Start cleanup task
+    asyncio.create_task(tasks.cleanup_old_results())
+    
+    db = database.SessionLocal()
+    try:
+        user = crud.get_user_by_username(db, "admin")
+        if not user:
+            hashed_password = auth.get_password_hash("admin123")
+            user_in = schemas.UserCreate(username="admin", password="admin123", role=models.UserRole.admin)
+            crud.create_user(db, user_in, hashed_password)
+    finally:
+        db.close()

asf-cloud-server/testarena/backend/app/models.py

@@ -0,0 +1,41 @@
+from sqlalchemy import Column, Integer, String, Enum, DateTime, ForeignKey, JSON
+from sqlalchemy.orm import relationship
+from .database import Base
+import datetime
+import enum
+
+class UserRole(str, enum.Enum):
+    admin = "admin"
+    user = "user"
+
+class JobStatus(str, enum.Enum):
+    pending = "pending"
+    running = "running"
+    passed = "passed"
+    failed = "failed"
+    aborted = "aborted"
+
+class User(Base):
+    __tablename__ = "users"
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String, unique=True, index=True)
+    hashed_password = Column(String)
+    role = Column(Enum(UserRole), default=UserRole.user)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+    jobs = relationship("Job", back_populates="owner")
+
+class Job(Base):
+    __tablename__ = "jobs"
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(Integer, ForeignKey("users.id"))
+    branch_name = Column(String)
+    scenarios = Column(JSON)
+    environment = Column(String)
+    test_mode = Column(String)
+    status = Column(Enum(JobStatus), default=JobStatus.pending)
+    result_path = Column(String, nullable=True)
+    duration = Column(String, nullable=True)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+    updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
+    
+    owner = relationship("User", back_populates="jobs")

asf-cloud-server/testarena/backend/app/schemas.py

@@ -0,0 +1,45 @@
+from pydantic import BaseModel
+from typing import List, Optional
+from datetime import datetime
+from .models import UserRole, JobStatus
+
+class UserBase(BaseModel):
+    username: str
+
+class UserCreate(UserBase):
+    password: str
+    role: UserRole = UserRole.user
+
+class User(UserBase):
+    id: int
+    role: UserRole
+    created_at: datetime
+    class Config:
+        orm_mode = True
+
+class JobBase(BaseModel):
+    branch_name: str
+    scenarios: List[str]
+    environment: str
+    test_mode: str
+
+class JobCreate(JobBase):
+    pass
+
+class Job(JobBase):
+    id: int
+    user_id: int
+    status: JobStatus
+    result_path: Optional[str]
+    duration: Optional[str]
+    created_at: datetime
+    updated_at: datetime
+    class Config:
+        orm_mode = True
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TokenData(BaseModel):
+    username: Optional[str] = None

asf-cloud-server/testarena/backend/app/socket_manager.py

@@ -0,0 +1,22 @@
+from fastapi import WebSocket
+from typing import List
+
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: List[WebSocket] = []
+
+    async def connect(self, websocket: WebSocket):
+        await websocket.accept()
+        self.active_connections.append(websocket)
+
+    def disconnect(self, websocket: WebSocket):
+        self.active_connections.remove(websocket)
+
+    async def broadcast(self, message: str):
+        for connection in self.active_connections:
+            try:
+                await connection.send_text(message)
+            except:
+                self.disconnect(connection)
+
+manager = ConnectionManager()

asf-cloud-server/testarena/backend/app/tasks.py

@@ -0,0 +1,109 @@
+import subprocess
+import asyncio
+from sqlalchemy.orm import Session
+from . import crud, models, database
+from .socket_manager import manager
+import json
+import os
+import time
+import shutil
+
+# In Docker, scripts are in /app/scripts
+# But for local testing, we might need a relative path or env var.
+# We'll stick to the Docker path assumption or use relative.
+SCRIPTS_DIR = os.getenv("SCRIPTS_DIR", "scripts")
+
+def get_scenarios(branch_name: str):
+    try:
+        # In a real scenario, we might need to git checkout first.
+        # For now, just run the script.
+        # Ensure we are running from the root of the backend if using relative paths
+        script_path = os.path.join(os.getcwd(), SCRIPTS_DIR, "get_scenarios.sh")
+        if not os.path.exists(script_path):
+             # Fallback for docker absolute path
+             script_path = f"/app/scripts/get_scenarios.sh"
+
+        result = subprocess.run(
+            [script_path, branch_name],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        # Expecting JSON output from script
+        return json.loads(result.stdout)
+    except subprocess.CalledProcessError as e:
+        print(f"Error getting scenarios: {e.stderr}")
+        return []
+    except Exception as e:
+        print(f"Error: {e}")
+        return []
+
+async def run_job_task(job_id: int):
+    db = database.SessionLocal()
+    try:
+        job = crud.get_job(db, job_id)
+        if not job:
+            return
+
+        crud.update_job_status(db, job_id, "running")
+        await manager.broadcast(json.dumps({"type": "job_update", "job_id": job_id, "status": "running"}))
+
+        # Run the script
+        # run_tests.sh <branch> <scenarios_json> <env> <mode> <job_id>
+        scenarios_str = json.dumps(job.scenarios)
+        
+        script_path = os.path.join(os.getcwd(), SCRIPTS_DIR, "run_tests.sh")
+        if not os.path.exists(script_path):
+             script_path = f"/app/scripts/run_tests.sh"
+
+        process = await asyncio.create_subprocess_exec(
+            script_path,
+            job.branch_name,
+            scenarios_str,
+            job.environment,
+            job.test_mode,
+            str(job_id),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE
+        )
+
+        stdout, stderr = await process.communicate()
+
+        if process.returncode == 0:
+            # Assume script generates index.html at /results/{job_id}/index.html
+            # We need to map where the script writes.
+            # For now, let's assume the script handles the file writing to a shared volume.
+            # In Docker, we'll mount /app/results
+            result_path = f"/results/{job_id}/index.html"
+            crud.update_job_status(db, job_id, "passed", result_path=result_path, duration="1m 30s") # Mock duration
+            await manager.broadcast(json.dumps({"type": "job_update", "job_id": job_id, "status": "passed"}))
+        else:
+            print(f"Script failed: {stderr.decode()}")
+            crud.update_job_status(db, job_id, "failed")
+            await manager.broadcast(json.dumps({"type": "job_update", "job_id": job_id, "status": "failed"}))
+
+    except Exception as e:
+        print(f"Job failed: {e}")
+        crud.update_job_status(db, job_id, "failed")
+        await manager.broadcast(json.dumps({"type": "job_update", "job_id": job_id, "status": "failed"}))
+    finally:
+        db.close()
+
+async def cleanup_old_results():
+    while True:
+        try:
+            print("Running cleanup...")
+            results_dir = "/results"
+            if os.path.exists(results_dir):
+                now = time.time()
+                for job_id in os.listdir(results_dir):
+                    job_path = os.path.join(results_dir, job_id)
+                    if os.path.isdir(job_path):
+                        mtime = os.path.getmtime(job_path)
+                        if now - mtime > 7 * 86400: # 7 days
+                            print(f"Deleting old result: {job_path}")
+                            shutil.rmtree(job_path)
+        except Exception as e:
+            print(f"Cleanup error: {e}")
+        
+        await asyncio.sleep(86400) # 24 hours