""" Beta Testing Framework Closed beta management, feedback collection, and launch preparation """ from fastapi import FastAPI, HTTPException from pydantic import BaseModel, EmailStr from typing import Dict, List, Optional, Any import json import datetime import hashlib import secrets from enum import Enum from dataclasses import dataclass app = FastAPI(title="Beta Testing Service", version="1.0.0") class BetaStatus(Enum): PENDING = "pending" APPROVED = "approved" ACTIVE = "active" SUSPENDED = "suspended" COMPLETED = "completed" class FeedbackType(Enum): BUG = "bug" FEATURE_REQUEST = "feature_request" IMPROVEMENT = "improvement" USABILITY = "usability" PERFORMANCE = "performance" SECURITY = "security" class TestPhase(Enum): CLOSED_BETA = "closed_beta" OPEN_BETA = "open_beta" EARLY_ACCESS = "early_access" GENERAL_AVAILABILITY = "general_availability" @dataclass class BetaTester: id: str email: str name: str company: Optional[str] role: str experience_level: str status: BetaStatus invited_at: datetime.datetime joined_at: Optional[datetime.datetime] last_active: Optional[datetime.datetime] test_metrics: Dict[str, Any] @dataclass class Feedback: id: str tester_id: str feedback_type: FeedbackType title: str description: str severity: str reproduction_steps: Optional[str] expected_behavior: Optional[str] actual_behavior: Optional[str] environment: Dict[str, Any] attachments: List[str] created_at: datetime.datetime status: str assigned_to: Optional[str] @dataclass class TestSession: id: str tester_id: str start_time: datetime.datetime end_time: Optional[datetime.datetime] duration: Optional[float] features_tested: List[str] bugs_found: int issues_reported: int satisfaction_score: Optional[int] notes: str class BetaTestingManager: def __init__(self): self.testers = {} self.feedback = {} self.test_sessions = {} self.waiting_list = [] self.current_phase = TestPhase.CLOSED_BETA self.beta_limits = { TestPhase.CLOSED_BETA: 50, TestPhase.OPEN_BETA: 500, TestPhase.EARLY_ACCESS: 2000 } self.invite_codes = {} self.analytics = { "total_testers": 0, "active_testers": 0, "feedback_submitted": 0, "bugs_reported": 0, "feature_requests": 0, "satisfaction_average": 0.0 } def generate_invite_code(self, email: str) -> str: code = secrets.token_urlsafe(8).upper() self.invite_codes[code] = { "email": email, "created_at": datetime.datetime.now(), "used": False, "expires_at": datetime.datetime.now() + datetime.timedelta(days=30) } return code def add_to_waiting_list(self, email: str, name: str, company: Optional[str] = None) -> str: waiting_id = hashlib.sha256(f"{email}{datetime.datetime.now()}".encode()).hexdigest()[:16] entry = { "id": waiting_id, "email": email, "name": name, "company": company, "added_at": datetime.datetime.now(), "status": "waiting" } self.waiting_list.append(entry) return waiting_id def invite_tester(self, email: str) -> Optional[str]: waiting_entry = next((w for w in self.waiting_list if w["email"] == email), None) if waiting_entry: invite_code = self.generate_invite_code(email) waiting_entry["invite_code"] = invite_code waiting_entry["status"] = "invited" waiting_entry["invited_at"] = datetime.datetime.now() return invite_code return None def register_tester(self, invite_code: str, name: str, company: Optional[str], role: str, experience: str) -> Optional[str]: if invite_code not in self.invite_codes: return None code_info = self.invite_codes[invite_code] if code_info["used"] or datetime.datetime.now() > code_info["expires_at"]: return None tester_id = hashlib.sha256(f"{code_info['email']}{datetime.datetime.now()}".encode()).hexdigest()[:16] tester = BetaTester( id=tester_id, email=code_info["email"], name=name, company=company, role=role, experience_level=experience, status=BetaStatus.APPROVED, invited_at=code_info["created_at"], joined_at=datetime.datetime.now(), last_active=datetime.datetime.now(), test_metrics={ "sessions_completed": 0, "bugs_found": 0, "feedback_submitted": 0, "features_tested": [], "hours_tested": 0.0 } ) self.testers[tester_id] = tester code_info["used"] = True self.analytics["total_testers"] += 1 self.analytics["active_testers"] += 1 return tester_id def submit_feedback(self, tester_id: str, feedback_data: Dict[str, Any]) -> str: if tester_id not in self.testers: raise ValueError("Invalid tester ID") feedback_id = hashlib.sha256(f"{tester_id}{datetime.datetime.now()}{feedback_data['title']}".encode()).hexdigest()[:16] feedback = Feedback( id=feedback_id, tester_id=tester_id, feedback_type=FeedbackType(feedback_data["type"]), title=feedback_data["title"], description=feedback_data["description"], severity=feedback_data.get("severity", "medium"), reproduction_steps=feedback_data.get("reproduction_steps"), expected_behavior=feedback_data.get("expected_behavior"), actual_behavior=feedback_data.get("actual_behavior"), environment=feedback_data.get("environment", {}), attachments=feedback_data.get("attachments", []), created_at=datetime.datetime.now(), status="open", assigned_to=None ) self.feedback[feedback_id] = feedback # Update tester metrics tester = self.testers[tester_id] tester.test_metrics["feedback_submitted"] += 1 tester.last_active = datetime.datetime.now() # Update analytics self.analytics["feedback_submitted"] += 1 if feedback.feedback_type == FeedbackType.BUG: self.analytics["bugs_reported"] += 1 elif feedback.feedback_type == FeedbackType.FEATURE_REQUEST: self.analytics["feature_requests"] += 1 return feedback_id def start_test_session(self, tester_id: str, features: List[str]) -> str: if tester_id not in self.testers: raise ValueError("Invalid tester ID") session_id = hashlib.sha256(f"{tester_id}{datetime.datetime.now()}".encode()).hexdigest()[:16] session = TestSession( id=session_id, tester_id=tester_id, start_time=datetime.datetime.now(), end_time=None, duration=None, features_tested=features, bugs_found=0, issues_reported=0, satisfaction_score=None, notes="" ) self.test_sessions[session_id] = session # Update tester metrics tester = self.testers[tester_id] tester.test_metrics["features_tested"].extend(features) tester.last_active = datetime.datetime.now() return session_id def end_test_session(self, session_id: str, satisfaction_score: int, notes: str) -> None: if session_id not in self.test_sessions: raise ValueError("Invalid session ID") session = self.test_sessions[session_id] session.end_time = datetime.datetime.now() session.duration = (session.end_time - session.start_time).total_seconds() / 3600 session.satisfaction_score = satisfaction_score session.notes = notes # Update tester metrics tester = self.testers[session.tester_id] tester.test_metrics["sessions_completed"] += 1 tester.test_metrics["hours_tested"] += session.duration # Update satisfaction average self._update_satisfaction_average() def _update_satisfaction_average(self): completed_sessions = [s for s in self.test_sessions.values() if s.satisfaction_score is not None] if completed_sessions: total_score = sum(s.satisfaction_score for s in completed_sessions) self.analytics["satisfaction_average"] = total_score / len(completed_sessions) def get_beta_analytics(self) -> Dict[str, Any]: active_testers = len([t for t in self.testers.values() if t.status == BetaStatus.ACTIVE]) recent_activity = len([t for t in self.testers.values() if t.last_active and (datetime.datetime.now() - t.last_active).days < 7]) return { **self.analytics, "active_testers": active_testers, "recently_active": recent_activity, "current_phase": self.current_phase.value, "phase_capacity": self.beta_limits.get(self.current_phase, 0), "waiting_list_size": len(self.waiting_list), "feedback_by_type": self._get_feedback_breakdown(), "tester_satisfaction": self._get_satisfaction_metrics(), "most_tested_features": self._get_most_tested_features() } def _get_feedback_breakdown(self) -> Dict[str, int]: breakdown = {ft.value: 0 for ft in FeedbackType} for feedback in self.feedback.values(): breakdown[feedback.feedback_type.value] += 1 return breakdown def _get_satisfaction_metrics(self) -> Dict[str, Any]: scores = [s.satisfaction_score for s in self.test_sessions.values() if s.satisfaction_score is not None] if not scores: return {"average": 0, "count": 0, "distribution": {}} return { "average": sum(scores) / len(scores), "count": len(scores), "distribution": { "1-2": len([s for s in scores if s <= 2]), "3-4": len([s for s in scores if 3 <= s <= 4]), "5": len([s for s in scores if s == 5]) } } def _get_most_tested_features(self) -> List[Dict[str, int]]: feature_counts = {} for session in self.test_sessions.values(): for feature in session.features_tested: feature_counts[feature] = feature_counts.get(feature, 0) + 1 return sorted([{"feature": k, "count": v} for k, v in feature_counts.items()], key=lambda x: x["count"], reverse=True)[:10] def advance_phase(self, new_phase: TestPhase) -> bool: if new_phase.value <= self.current_phase.value: return False self.current_phase = new_phase return True def get_tester_leaderboard(self) -> List[Dict[str, Any]]: leaderboard = [] for tester in self.testers.values(): score = ( tester.test_metrics["sessions_completed"] * 10 + tester.test_metrics["bugs_found"] * 25 + tester.test_metrics["feedback_submitted"] * 15 + len(tester.test_metrics["features_tested"]) * 5 ) leaderboard.append({ "name": tester.name, "company": tester.company, "score": score, "sessions": tester.test_metrics["sessions_completed"], "bugs": tester.test_metrics["bugs_found"], "feedback": tester.test_metrics["feedback_submitted"], "features": len(tester.test_metrics["features_tested"]) }) return sorted(leaderboard, key=lambda x: x["score"], reverse=True)[:20] beta_manager = BetaTestingManager() class WaitingListRequest(BaseModel): email: EmailStr name: str company: Optional[str] = None role: str experience: str class FeedbackRequest(BaseModel): type: str title: str description: str severity: Optional[str] = "medium" reproduction_steps: Optional[str] = None expected_behavior: Optional[str] = None actual_behavior: Optional[str] = None environment: Optional[Dict[str, Any]] = None attachments: Optional[List[str]] = None class SessionRequest(BaseModel): features: List[str] class SessionEndRequest(BaseModel): satisfaction_score: int notes: str @app.get("/health") async def health(): return {"status": "healthy", "service": "beta-testing"} @app.post("/waiting-list") async def join_waiting_list(request: WaitingListRequest): try: waiting_id = beta_manager.add_to_waiting_list( request.email, request.name, request.company ) return { "waiting_id": waiting_id, "status": "added", "message": "You've been added to the waiting list. We'll notify you when a spot becomes available.", "estimated_wait": "2-4 weeks" } except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to join waiting list: {str(e)}") @app.post("/register") async def register_beta_tester(invite_code: str, name: str, company: Optional[str] = None, role: str = "developer", experience: str = "intermediate"): try: tester_id = beta_manager.register_tester(invite_code, name, company, role, experience) if not tester_id: raise HTTPException(status_code=400, detail="Invalid or expired invite code") return { "tester_id": tester_id, "status": "registered", "message": "Welcome to the beta program! You can now start testing.", "next_steps": [ "Complete the onboarding tutorial", "Join our Discord community", "Start your first test session" ] } except Exception as e: raise HTTPException(status_code=500, detail=f"Registration failed: {str(e)}") @app.post("/feedback") async def submit_feedback(tester_id: str, request: FeedbackRequest): try: feedback_id = beta_manager.submit_feedback(tester_id, request.dict()) return { "feedback_id": feedback_id, "status": "submitted", "message": "Thank you for your feedback! We'll review it and get back to you soon." } except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: raise HTTPException(status_code=500, detail=f"Feedback submission failed: {str(e)}") @app.post("/session/start") async def start_test_session(tester_id: str, request: SessionRequest): try: session_id = beta_manager.start_test_session(tester_id, request.features) return { "session_id": session_id, "status": "started", "message": "Test session started. Good luck!", "features_to_test": request.features } except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to start session: {str(e)}") @app.post("/session/{session_id}/end") async def end_test_session(session_id: str, request: SessionEndRequest): try: beta_manager.end_test_session(session_id, request.satisfaction_score, request.notes) return { "status": "completed", "message": "Test session completed. Thank you for your contribution!", "next_session_available": datetime.datetime.now() + datetime.timedelta(hours=24) } except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to end session: {str(e)}") @app.get("/analytics") async def get_beta_analytics(): try: return beta_manager.get_beta_analytics() except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to get analytics: {str(e)}") @app.get("/leaderboard") async def get_tester_leaderboard(): try: return {"leaderboard": beta_manager.get_tester_leaderboard()} except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to get leaderboard: {str(e)}") @app.get("/tester/{tester_id}") async def get_tester_info(tester_id: str): try: if tester_id not in beta_manager.testers: raise HTTPException(status_code=404, detail="Tester not found") tester = beta_manager.testers[tester_id] return { "id": tester.id, "name": tester.name, "company": tester.company, "status": tester.status.value, "joined_at": tester.joined_at.isoformat() if tester.joined_at else None, "last_active": tester.last_active.isoformat() if tester.last_active else None, "metrics": tester.test_metrics } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to get tester info: {str(e)}") @app.post("/admin/invite") async def invite_tester(email: str): try: invite_code = beta_manager.invite_tester(email) if not invite_code: raise HTTPException(status_code=404, detail="Email not found in waiting list") return { "invite_code": invite_code, "email": email, "status": "invited" } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to send invite: {str(e)}") @app.post("/admin/phase") async def advance_beta_phase(new_phase: str): try: from enum import Enum phase_enum = TestPhase(new_phase) success = beta_manager.advance_phase(phase_enum) if not success: raise HTTPException(status_code=400, detail="Cannot advance to this phase") return { "previous_phase": beta_manager.current_phase.value, "new_phase": new_phase, "status": "advanced", "capacity": beta_manager.beta_limits.get(phase_enum, 0) } except ValueError: raise HTTPException(status_code=400, detail=f"Invalid phase: {new_phase}") except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to advance phase: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8007)