From 4cc2e6fceeceea52fc03130441f9801f8a761274 Mon Sep 17 00:00:00 2001 From: Jerome Hardaway Date: Wed, 23 Oct 2024 21:25:43 -0400 Subject: [PATCH 1/2] Add feedback JSON file and update requirements for new dependencies --- feedback/feedback_20241023_212458.json | 6 + requirements.txt | 26 +- streamlit_app.py | 623 ++++++++++++++++--------- 3 files changed, 434 insertions(+), 221 deletions(-) create mode 100644 feedback/feedback_20241023_212458.json diff --git a/feedback/feedback_20241023_212458.json b/feedback/feedback_20241023_212458.json new file mode 100644 index 0000000..74336bf --- /dev/null +++ b/feedback/feedback_20241023_212458.json @@ -0,0 +1,6 @@ +{ + "timestamp": "2024-10-23T21:24:58.418136", + "session_id": "2953efaf8e8a16eca63a8c20d9ffd803", + "rating": 3, + "feedback": "A little slow" +} \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 51f63b5..9c7eac5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,18 @@ -streamlit -httpx -nest-asyncio -better-profanity -PyPDF2 -python-docx -python-dotenv -openai +streamlit>=1.28.0 +httpx>=0.25.0 +nest-asyncio>=1.5.8 +better-profanity>=0.7.0 +PyPDF2>=3.0.0 +python-docx>=1.0.0 +python-dotenv>=1.0.0 +openai>=1.3.0 +# New dependencies for enhanced features +tenacity>=8.2.3 +typing-extensions>=4.8.0 +python-json-logger>=2.0.7 +aiofiles>=23.2.1 +tiktoken>=0.5.1 +cachetools>=5.3.2 +dataclasses-json>=0.6.1 +asyncio>=3.4.3 +aiohttp>=3.9.1 \ No newline at end of file diff --git a/streamlit_app.py b/streamlit_app.py index 7686659..e1e43e0 100644 --- a/streamlit_app.py +++ b/streamlit_app.py @@ -1,235 +1,432 @@ import streamlit as st import os -import httpx -import nest_asyncio -from better_profanity import profanity -from PyPDF2 import PdfReader -from PyPDF2.errors import PdfReadError -import docx -from zipfile import BadZipfile +import logging +from typing import Dict, List +from datetime import datetime from dotenv import load_dotenv - -# Apply nest_asyncio to allow nested event loops -nest_asyncio.apply() - -# Initialize the better-profanity filter -profanity.load_censor_words() - -# Load the environment variables from .env file +import openai +import json +import time +import hashlib + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler('vetsai.log'), + logging.StreamHandler() + ] +) +logger = logging.getLogger(__name__) + +# Load environment variables load_dotenv() -# Get the API key from environment variables -OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") - -# Function to inject custom CSS for light mode -def inject_custom_css(): - st.markdown(""" - - """, unsafe_allow_html=True) - -inject_custom_css() - -# Function to read and extract text from PDFs -def extract_text_from_pdf(file): - try: - reader = PdfReader(file) - - # Raise error if PDF is password protected - if reader.is_encrypted: - raise ValueError("Encrypted PDF files are not supported.") - - text = "" - - for page in reader.pages: - text += page.extract_text() - # Forward errors from PdfReader with a user-friendly message - except (TypeError, PdfReadError) as e: - raise TypeError("Invalid PDF file.") - return text - -# Function to read and extract text from Word documents -def extract_text_from_word(file): - try: - doc = docx.Document(file) - return "\n".join([paragraph.text for paragraph in doc.paragraphs]) - except ValueError: - raise - except BadZipfile as e: - raise TypeError("Invalid docx file. File may be password protected or corrupted.") - -# Function to load military job codes from the directories (TXT format) -def load_military_job_codes(base_path): - # Your existing implementation - pass - -# Function to translate military job code to civilian job suggestions -def translate_job_code(job_code, job_codes): - # Your existing implementation - pass - -# Fetch response from OpenAI using the API key with increased timeout -def fetch_from_model(conversation): - """Send a request to OpenAI using the conversation history.""" - url = "https://api.openai.com/v1/chat/completions" - - headers = { - "Authorization": f"Bearer {OPENAI_API_KEY}", - "Content-Type": "application/json" +# Configure OpenAI +openai.api_key = os.getenv("OPENAI_API_KEY") +if not openai.api_key: + raise ValueError("OpenAI API key not found in .env file") + +def load_military_job_codes() -> dict: + """ + Load military job codes from data directories and map them to software development paths. + Directory structure: + data/ + employment_transitions/ + job_codes/ + army/ + air_force/ + coast_guard/ + navy/ + marine_corps/ + """ + base_path = "data/employment_transitions/job_codes" + job_codes = {} + + # Map of service branches to their file paths and code prefixes + branches = { + "army": {"path": "army", "prefix": "MOS"}, + "air_force": {"path": "air_force", "prefix": "AFSC"}, + "coast_guard": {"path": "coast_guard", "prefix": "RATE"}, + "navy": {"path": "navy", "prefix": "RATE"}, + "marine_corps": {"path": "marine_corps", "prefix": "MOS"} } + + for branch, info in branches.items(): + branch_path = os.path.join(base_path, info["path"]) + if os.path.exists(branch_path): + for file in os.listdir(branch_path): + if file.endswith('.json'): + with open(os.path.join(branch_path, file), 'r') as f: + try: + branch_codes = json.load(f) + # Add VWC specific development paths to each job code + for code, details in branch_codes.items(): + vwc_mapping = map_to_vwc_path(details.get('category', ''), + details.get('skills', [])) + details.update({ + 'vwc_path': vwc_mapping['path'], + 'tech_focus': vwc_mapping['tech_focus'], + 'branch': branch, + 'code_type': info['prefix'] + }) + job_codes[f"{info['prefix']}_{code}"] = details + except json.JSONDecodeError as e: + logger.error(f"Error loading {file}: {e}") + continue + + return job_codes - payload = { - "model": "gpt-4o", # Use 'gpt-3.5-turbo' or 'gpt-4' if available - "messages": conversation, - "temperature": 0.7, - "max_tokens": 5000 +def map_to_vwc_path(category: str, skills: List[str]) -> dict: + """Map military job categories and skills to VWC tech stack paths.""" + + # Default full stack path + default_path = { + "path": "Full Stack Development", + "tech_focus": [ + "JavaScript/TypeScript fundamentals", + "Next.js and Tailwind for frontend", + "Python with FastAPI/Django for backend" + ] } - - try: - # Set a custom timeout (e.g., 60 seconds) to give more time for OpenAI to respond - response = httpx.post(url, headers=headers, json=payload, timeout=60.0) - response.raise_for_status() - - json_response = response.json() - return json_response['choices'][0]['message']['content'] - - except httpx.TimeoutException: - st.error("The request to OpenAI timed out. Please try again later.") - return "Timeout occurred while waiting for OpenAI response." - - except httpx.RequestError as e: - st.error(f"An error occurred while making a request to OpenAI: {e}") - return "Error communicating with the OpenAI API." - except Exception as e: - st.error(f"An unexpected error occurred: {e}") - return "Unexpected error while fetching response." - -# Callback to process user input and clear it afterward -def process_input(job_codes): - user_input = st.session_state["temp_input"] + # Category-based mappings + tech_paths = { + "information_technology": { + "path": "Full Stack Development", + "tech_focus": [ + "JavaScript/TypeScript with focus on system architecture", + "Next.js for complex web applications", + "Python backend services with FastAPI" + ] + }, + "cyber": { + "path": "Security-Focused Development", + "tech_focus": [ + "TypeScript for type-safe applications", + "Secure API development with FastAPI/Django", + "AI/ML for security applications" + ] + }, + "communications": { + "path": "Frontend Development", + "tech_focus": [ + "JavaScript/TypeScript specialization", + "Advanced Next.js and Tailwind", + "API integration with Python backends" + ] + }, + "intelligence": { + "path": "AI/ML Development", + "tech_focus": [ + "Python for data processing", + "ML model deployment with FastAPI", + "Next.js for ML application frontends" + ] + }, + "maintenance": { + "path": "Backend Development", + "tech_focus": [ + "Python backend development", + "API design with FastAPI/Django", + "Basic frontend with Next.js" + ] + } + } - if user_input: - # Store user input into chat history - st.session_state.messages.append({"role": "user", "content": user_input}) - - # Build conversation history for OpenAI API call - conversation = [{"role": "system", "content": "You are a helpful assistant for veterans seeking employment."}] - - # Include document content in the system prompt if available - if "document_content" in st.session_state and st.session_state["document_content"]: - conversation[0]["content"] += f" The user has provided the following document content to assist you: {st.session_state['document_content']}" - - # Append previous messages, being mindful of token limits - for msg in st.session_state.messages[-10:]: # Adjust the number of messages as needed - conversation.append(msg) - - # Fetch assistant's response - response = fetch_from_model(conversation) - - # Store assistant's response - st.session_state.messages.append({"role": "assistant", "content": response}) - - # Clear the temporary input - st.session_state["temp_input"] = "" - -# Handle user input and job code translation along with resume upload -def handle_user_input(job_codes): - """Handle user input for translating military job codes to civilian jobs, uploading resumes, and chatting.""" + # Skill-based adjustments + skill_keywords = { + "programming": "software", + "database": "data", + "network": "communications", + "security": "cyber", + "analysis": "intelligence" + } - # Display chat messages first - display_chat_messages() - - # File uploader for document uploads - uploaded_file = st.file_uploader("Upload your employment-related document (PDF, DOCX)", type=["pdf", "docx"]) + # Determine best path based on category and skills + if category.lower() in tech_paths: + return tech_paths[category.lower()] + + # Check skills for keywords + for skill in skills: + skill_lower = skill.lower() + for keyword, category in skill_keywords.items(): + if keyword in skill_lower and category in tech_paths: + return tech_paths[category] + + return default_path - if uploaded_file is not None: - try: - supported_file_types = [ - "application/pdf", - "application/vnd.openxmlformats-officedocument.wordprocessingml.document" +def translate_military_code(code: str, job_codes: dict) -> dict: + """Translate military code to VWC development path.""" + # Clean and standardize input + code = code.upper().strip() + + # Remove common prefixes if provided + prefixes = ["MOS", "AFSC", "RATE"] + for prefix in prefixes: + if code.startswith(prefix): + code = code.replace(prefix, "").strip() + + # Try different prefix combinations + possible_codes = [ + f"MOS_{code}", + f"AFSC_{code}", + f"RATE_{code}" + ] + + for possible_code in possible_codes: + if possible_code in job_codes: + job_data = job_codes[possible_code] + return { + "found": True, + "data": { + "title": job_data.get('title', 'Military Professional'), + "branch": job_data.get('branch', 'Military'), + "dev_path": job_data.get('vwc_path', 'Full Stack Development'), + "tech_focus": job_data.get('tech_focus', []), + "skills": job_data.get('skills', []) + } + } + + # Default response for unknown codes + return { + "found": False, + "data": { + "title": "Military Professional", + "branch": "Military", + "dev_path": "Full Stack Development", + "tech_focus": [ + "Start with JavaScript/TypeScript fundamentals", + "Build projects with Next.js and Tailwind", + "Learn Python backend development with FastAPI" + ], + "skills": [ + "Leadership and team coordination", + "Problem-solving and adaptation", + "Project planning and execution" ] + } + } - if uploaded_file.type not in supported_file_types: - raise TypeError("Invalid file type.") - - # Limit file uploads to less than 20 MB - if uploaded_file.size > 20 * 1024 * 1024: - raise ValueError("File size is too large. Uploaded files must be less than 20 MB.") - - file_text = "" - - if uploaded_file.type == "application/pdf": - file_text = extract_text_from_pdf(uploaded_file) - elif uploaded_file.type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document": - file_text = extract_text_from_word(uploaded_file) - - # Store the extracted content in session state - st.session_state["document_content"] = file_text - - st.success("Document uploaded and processed successfully!") - except (TypeError, ValueError) as e: - st.error(e) +def get_chat_response(messages: List[Dict]) -> str: + """Get response from OpenAI chat completion.""" + try: + response = openai.chat.completions.create( + model="gpt-4", + messages=messages, + temperature=0.7, + ) + return response.choices[0].message.content + except openai.OpenAIError as e: + logger.error(f"OpenAI API error: {e}") + raise + except Exception as e: + logger.error(f"Unexpected error in chat completion: {e}") + raise - # Input field for user queries (job code or general chat) at the bottom - st.text_input("Enter your military job code (e.g., 11B, AFSC, MOS) or ask a question:", - key="temp_input", - on_change=process_input, - args=(job_codes,)) +def export_chat_history(chat_history: List[Dict]) -> str: + """Export chat history to JSON.""" + export_data = { + "timestamp": datetime.now().isoformat(), + "messages": chat_history + } + return json.dumps(export_data, indent=2) -# Display the app title and description -def display_title_and_description(): - """Display the app title and description.""" - st.title("πŸ‡ΊπŸ‡Έ VetsAI: Employment Assistance for Veterans") - st.write( - "Welcome to VetsAI, an AI-powered virtual assistant designed " - "to help veterans navigate employment transitions and find opportunities in civilian careers." +def save_feedback(feedback: Dict): + """Save user feedback to file.""" + feedback_dir = "feedback" + os.makedirs(feedback_dir, exist_ok=True) + + feedback_file = os.path.join( + feedback_dir, + f"feedback_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" ) - -# Initialize session state -def initialize_session_state(): - """Initialize session state variables for messages and chat history.""" - if "messages" not in st.session_state: - st.session_state.messages = [] - if "temp_input" not in st.session_state: - st.session_state.temp_input = "" - if "document_content" not in st.session_state: - st.session_state.document_content = "" - -# Introduce the assistant -def introduce_assistant(): - """Introduce the VetsAI Assistant.""" - if not st.session_state.messages: - intro_message = ( - "Hi, I'm VetsAI! I'm here to assist you in finding employment opportunities and transitioning into civilian careers. " - "Feel free to ask me anything related to job searching, resume tips, or industries that align with your skills." - ) - st.session_state.messages.append({"role": "assistant", "content": intro_message}) - -# Display chat history -def display_chat_messages(): - """Display existing chat messages stored in session state.""" - for message in st.session_state["messages"]: - if message["role"] == "user": - with st.chat_message("user"): - st.markdown(f"You: {message['content']}") + + with open(feedback_file, 'w') as f: + json.dump(feedback, f, indent=2) + +def handle_command(command: str) -> str: + """Handle special commands including MOS translation.""" + parts = command.lower().split() + if not parts: + return None + + cmd = parts[0] + if cmd in ['/mos', '/afsc', '/rate']: + if len(parts) < 2: + return "Please provide a military job code. Example: `/mos 25B`" + + code = parts[1] + translation = translate_military_code(code, st.session_state.job_codes) + if translation['found']: + return ( + f"πŸŽ–οΈ **{translation['data']['title']}** ({translation['data']['branch']})\n\n" + f"πŸ’» **VWC Development Path**: {translation['data']['dev_path']}\n\n" + "πŸ”§ **Military Skills**:\n" + + "\n".join(f"- {skill}" for skill in translation['data']['skills']) + + "\n\nπŸ“š **VWC Tech Focus**:\n" + + "\n".join(f"{i+1}. {focus}" for i, focus in enumerate(translation['data']['tech_focus'])) + ) else: - with st.chat_message("assistant"): - st.markdown(f"VetsAI: {message['content']}") + return ( + "I don't have that specific code in my database, but here's a recommended " + "VWC learning path based on general military experience:\n\n" + + "\n".join(f"{i+1}. {focus}" for i, focus in enumerate(translation['data']['tech_focus'])) + ) + + return None + +def initialize_chat(): + """Initialize the chat with a VWC-focused welcome message.""" + welcome_message = { + "role": "assistant", + "content": ( + "Welcome to VetsAI - Your Vets Who Code Assistant! πŸ‘¨β€πŸ’»\n\n" + "I'm here to help you with:\n\n" + "πŸ”Ή VWC Tech Stack:\n" + "- JavaScript/TypeScript\n" + "- Python (FastAPI, Flask, Django)\n" + "- Next.js & Tailwind CSS\n" + "- AI/ML Integration\n\n" + "πŸ”Ή Commands:\n" + "- `/mos [code]` - Translate your MOS to dev path\n" + "- `/afsc [code]` - Translate your AFSC to dev path\n" + "- `/rate [code]` - Translate your Rate to dev path\n" + "- `/frontend` - Help with JS/TS/Next.js\n" + "- `/backend` - Help with Python frameworks\n" + "- `/ai` - AI/ML guidance\n\n" + "Let's start by checking how your military experience " + "aligns with software development! Share your MOS/AFSC/Rate, " + "or ask about any part of our tech stack." + ) + } + return [welcome_message] -# Main function to run the VetsAI Assistant app def main(): - """Main function to run the VetsAI Assistant app.""" - display_title_and_description() - initialize_session_state() - - # Load the military job codes from the 'data/employment_transitions/job_codes' directory - job_codes = load_military_job_codes("./data/employment_transitions/job_codes") - - # Ensure the assistant introduces itself only once - introduce_assistant() - - # Handle user input and chat - handle_user_input(job_codes) + """Main application function.""" + st.title("πŸ‡ΊπŸ‡Έ VetsAI: Vets Who Code Assistant") + + # Initialize session + if 'session_id' not in st.session_state: + st.session_state.session_id = hashlib.md5( + str(time.time()).encode() + ).hexdigest() + + # Load military job codes + if 'job_codes' not in st.session_state: + try: + st.session_state.job_codes = load_military_job_codes() + except Exception as e: + logger.error(f"Error loading job codes: {e}") + st.session_state.job_codes = {} + + if 'messages' not in st.session_state: + st.session_state.messages = initialize_chat() + + # Add sidebar with VWC tech stack resources + with st.sidebar: + st.markdown(""" + ### VWC Tech Stack + + 🌐 **Frontend** + - JavaScript/TypeScript + - CSS & Tailwind + - Next.js + + βš™οΈ **Backend** + - Python + - FastAPI + - Flask + - Django + + πŸ€– **AI/ML Integration** + - Machine Learning + - AI Applications + + πŸŽ–οΈ **Military Translation** + `/mos [code]` - Army/Marines + `/afsc [code]` - Air Force + `/rate [code]` - Navy/Coast Guard + """) + + # Chat interface + for message in st.session_state.messages: + with st.chat_message(message["role"]): + st.markdown(message["content"]) + + if prompt := st.chat_input(): + # Add user message + st.session_state.messages.append({"role": "user", "content": prompt}) + with st.chat_message("user"): + st.markdown(prompt) + + # Check for commands first + if prompt.startswith('/'): + command_response = handle_command(prompt) + if command_response: + with st.chat_message("assistant"): + st.markdown(command_response) + st.session_state.messages.append({ + "role": "assistant", + "content": command_response + }) + return + + # Generate and display assistant response + with st.chat_message("assistant"): + try: + messages = st.session_state.messages.copy() + messages.insert(0, { + "role": "system", + "content": ( + "You are a specialized AI assistant for Vets Who Code troops. " + "Focus specifically on our tech stack: JavaScript, TypeScript, " + "Python, CSS, Tailwind, FastAPI, Flask, Next.js, Django, and AI/ML. " + "Always reference these specific technologies in your answers. " + "Remember all users are VWC troops learning our stack." + ) + }) + + response = get_chat_response(messages) + st.markdown(response) + + st.session_state.messages.append({ + "role": "assistant", + "content": response + }) + except Exception as e: + st.error(f"Error generating response: {str(e)}") + + # Export chat history + if st.button("Export Chat History"): + chat_export = export_chat_history(st.session_state.messages) + st.download_button( + "Download Chat History", + chat_export, + "vetsai_chat_history.json", + "application/json" + ) + + # Feedback mechanism + with st.expander("Provide Feedback"): + feedback_rating = st.slider( + "Rate your experience (1-5)", + min_value=1, + max_value=5, + value=5 + ) + feedback_text = st.text_area("Additional feedback") + + if st.button("Submit Feedback"): + feedback = { + "timestamp": datetime.now().isoformat(), + "session_id": st.session_state.session_id, + "rating": feedback_rating, + "feedback": feedback_text + } + save_feedback(feedback) + st.success("Thank you for your feedback!") if __name__ == "__main__": - main() + main() \ No newline at end of file From 14d6a37677a97ed7a60061b14b82bb9658582f1b Mon Sep 17 00:00:00 2001 From: Jerome Hardaway Date: Thu, 24 Oct 2024 13:23:08 -0400 Subject: [PATCH 2/2] adding tests and refactor --- streamlit_app.py => app.py | 111 ++++++++++++----- tests/__init__.py | 0 tests/conftest.py | 7 ++ tests/test_app.py | 249 +++++++++++++++++++++++++++++++++++++ 4 files changed, 336 insertions(+), 31 deletions(-) rename streamlit_app.py => app.py (81%) create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_app.py diff --git a/streamlit_app.py b/app.py similarity index 81% rename from streamlit_app.py rename to app.py index e1e43e0..cc9b0a3 100644 --- a/streamlit_app.py +++ b/app.py @@ -28,19 +28,67 @@ if not openai.api_key: raise ValueError("OpenAI API key not found in .env file") -def load_military_job_codes() -> dict: +def parse_mos_file(file_content: str) -> dict: """ - Load military job codes from data directories and map them to software development paths. - Directory structure: - data/ - employment_transitions/ - job_codes/ - army/ - air_force/ - coast_guard/ - navy/ - marine_corps/ + Parse military job code text file content into a structured dictionary. + + Args: + file_content: Raw text content of the MOS file + + Returns: + dict: Structured data including title, category, and skills """ + lines = file_content.strip().split('\n') + + job_code = "" + title = "" + description = [] + parsing_description = False + + for line in lines: + line = line.strip() + if not line: + continue + + if line.startswith("Job Code:"): + job_code = line.replace("Job Code:", "").strip() + elif line.startswith("Description:"): + parsing_description = True + elif parsing_description: + description.append(line) + + # Get the first non-empty description line as title + for line in description: + if line: + title = line + break + + # Combine all description text for category analysis + full_text = ' '.join(description).lower() + + # More comprehensive category detection + category = "general" + category_keywords = { + "information_technology": ["technology", "computer", "network", "data", "software", "hardware", "system", "database"], + "communications": ["communications", "signal", "radio", "transmission", "telecom"], + "intelligence": ["intelligence", "analysis", "surveillance", "reconnaissance"], + "maintenance": ["maintenance", "repair", "technical", "equipment"], + "cyber": ["cyber", "security", "information assurance", "cryptographic"] + } + + # Check for category keywords in the full text + for cat, keywords in category_keywords.items(): + if any(keyword in full_text for keyword in keywords): + category = cat + break + + return { + "title": title or "Military Professional", + "category": category, + "skills": [line for line in description if line and len(line) > 10] + } + +def load_military_job_codes() -> dict: base_path = "data/employment_transitions/job_codes" job_codes = {} @@ -57,27 +105,28 @@ def load_military_job_codes() -> dict: branch_path = os.path.join(base_path, info["path"]) if os.path.exists(branch_path): for file in os.listdir(branch_path): - if file.endswith('.json'): - with open(os.path.join(branch_path, file), 'r') as f: - try: - branch_codes = json.load(f) - # Add VWC specific development paths to each job code - for code, details in branch_codes.items(): - vwc_mapping = map_to_vwc_path(details.get('category', ''), - details.get('skills', [])) - details.update({ - 'vwc_path': vwc_mapping['path'], - 'tech_focus': vwc_mapping['tech_focus'], - 'branch': branch, - 'code_type': info['prefix'] - }) - job_codes[f"{info['prefix']}_{code}"] = details - except json.JSONDecodeError as e: - logger.error(f"Error loading {file}: {e}") - continue + if file.endswith('.txt'): # Changed from .json to .txt + try: + with open(os.path.join(branch_path, file), 'r') as f: + content = f.read() + code = file.replace('.txt', '') + details = parse_mos_file(content) + + # Add VWC specific development paths + vwc_mapping = map_to_vwc_path(details.get('category', ''), + details.get('skills', [])) + details.update({ + 'vwc_path': vwc_mapping['path'], + 'tech_focus': vwc_mapping['tech_focus'], + 'branch': branch, + 'code_type': info['prefix'] + }) + job_codes[f"{info['prefix']}_{code}"] = details + except Exception as e: + logger.error(f"Error loading {file}: {e}") + continue return job_codes - def map_to_vwc_path(category: str, skills: List[str]) -> dict: """Map military job categories and skills to VWC tech stack paths.""" @@ -213,7 +262,7 @@ def get_chat_response(messages: List[Dict]) -> str: """Get response from OpenAI chat completion.""" try: response = openai.chat.completions.create( - model="gpt-4", + model="gpt-4o", messages=messages, temperature=0.7, ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..6b22c19 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,7 @@ +import os +import sys +from pathlib import Path + +# Add the project root directory to Python path +ROOT_DIR = Path(__file__).parent.parent +sys.path.append(str(ROOT_DIR)) \ No newline at end of file diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..992cffe --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,249 @@ +import os +import sys +from pathlib import Path +import pytest +from unittest.mock import patch, mock_open, MagicMock, call +import json +import openai +from datetime import datetime + +# Get the absolute path to the project root directory +ROOT_DIR = Path(__file__).parent.parent +sys.path.append(str(ROOT_DIR)) + +from app import ( + load_military_job_codes, + map_to_vwc_path, + translate_military_code, + get_chat_response, + handle_command, + export_chat_history, + save_feedback, + parse_mos_file +) + +# Sample text content +SAMPLE_MOS_TEXT = """ +Job Code: 25B + +Description: +Manages or supervises a specific automated system or node in a data or communications network. + +Manages or supervises a specific automated system or node in a data or communications network supporting tactical, theater, strategic or base operations; provides detailed technical direction and advice to commanders, staffs and other Command, Control, and Communications (C3) users at all echelons on the installation, operation and maintenance of distributed operating and data base systems, teleprocessing systems, and data communications supporting Battlefield Automated Systems (BAS); requires the practical application of automation theory to the design, implementation and successful interoperation of hardware and software for automated telecommunications and teleprocessing systems. +""" + +@pytest.fixture +def mock_job_codes(): + return { + "MOS_25B": { + "title": "Information Technology Specialist", + "branch": "army", + "category": "information_technology", + "skills": ["Network administration", "System maintenance"], + "vwc_path": "Full Stack Development", + "tech_focus": [ + "JavaScript/TypeScript with focus on system architecture", + "Next.js for complex web applications", + "Python backend services with FastAPI" + ], + "code_type": "MOS" + } + } + +@patch("os.path.join", lambda *args: "/".join(args)) +@patch("builtins.open", new_callable=mock_open) +def test_load_military_job_codes(mock_file): + # Setup mock file content + mock_file.return_value.__enter__.return_value.read.return_value = SAMPLE_MOS_TEXT + + def mock_exists(path): + return True + + def mock_listdir(path): + if path.endswith("job_codes"): + return ["army", "air_force", "navy", "marine_corps", "coast_guard"] + else: + return ["25B.txt"] + + with patch("os.path.exists", side_effect=mock_exists), \ + patch("os.listdir", side_effect=mock_listdir): + + job_codes = load_military_job_codes() + + # Basic validations + assert isinstance(job_codes, dict) + assert len(job_codes) > 0 + + # Verify the structure + for key, value in job_codes.items(): + assert isinstance(value, dict) + assert "title" in value + assert "branch" in value + assert "skills" in value + assert isinstance(value["skills"], list) + + # Verify that mock_file was called + assert mock_file.call_count > 0 + +def test_parse_mos_file(): + """Test the MOS file parsing function""" + result = parse_mos_file(SAMPLE_MOS_TEXT) + + # Basic structure tests + assert isinstance(result, dict) + assert "title" in result + assert "category" in result + assert "skills" in result + assert isinstance(result["skills"], list) + assert len(result["skills"]) > 0 + + # Content tests + assert result["title"].startswith("Manages or supervises") + assert result["category"] == "information_technology" # Should match because of network/data/system keywords + + # Skills check + assert any("network" in skill.lower() for skill in result["skills"]) + +def test_parse_mos_file_edge_cases(): + """Test parse_mos_file with various edge cases""" + # Empty content + empty_result = parse_mos_file("") + assert empty_result["title"] == "Military Professional" + assert empty_result["category"] == "general" + assert isinstance(empty_result["skills"], list) + + # Content with only job code + job_code_only = "Job Code: 25B" + job_code_result = parse_mos_file(job_code_only) + assert job_code_result["title"] == "Military Professional" + assert isinstance(job_code_result["skills"], list) + + # Content with special characters + special_chars = """ + Job Code: 25B + + Description: + Network & Systems Administrator (IT/IS) + + Manages & maintains computer networks/systems. + """ + special_result = parse_mos_file(special_chars) + assert special_result["category"] == "information_technology" + +def test_map_to_vwc_path_it_category(): + result = map_to_vwc_path("information_technology", ["programming", "networking"]) + assert result["path"] == "Full Stack Development" + assert len(result["tech_focus"]) > 0 + assert any("TypeScript" in focus for focus in result["tech_focus"]) + +def test_map_to_vwc_path_default(): + result = map_to_vwc_path("unknown_category", []) + assert result["path"] == "Full Stack Development" + assert len(result["tech_focus"]) > 0 + +def test_translate_military_code_found(mock_job_codes): + result = translate_military_code("25B", mock_job_codes) + assert result["found"] == True + assert result["data"]["title"] == "Information Technology Specialist" + assert result["data"]["branch"] == "army" + +def test_translate_military_code_not_found(mock_job_codes): + result = translate_military_code("99Z", mock_job_codes) + assert result["found"] == False + assert "dev_path" in result["data"] + assert isinstance(result["data"]["tech_focus"], list) + +@patch("openai.chat.completions.create") +def test_get_chat_response(mock_create): + # Mock the OpenAI response + mock_response = MagicMock() + mock_response.choices = [MagicMock(message=MagicMock(content="Test response"))] + mock_create.return_value = mock_response + + messages = [{"role": "user", "content": "Hello"}] + response = get_chat_response(messages) + assert response == "Test response" + mock_create.assert_called_once() + +def test_handle_command_mos(mock_job_codes): + with patch("streamlit.session_state") as mock_session: + mock_session.job_codes = mock_job_codes + response = handle_command("/mos 25B") + assert response is not None + assert "Information Technology Specialist" in response + assert "VWC Development Path" in response + +def test_handle_command_invalid(): + response = handle_command("/invalid") + assert response is None + +def test_handle_command_missing_code(): + response = handle_command("/mos") + assert "Please provide a military job code" in response + +def test_export_chat_history(): + chat_history = [ + {"role": "user", "content": "Hello"}, + {"role": "assistant", "content": "Hi"} + ] + result = export_chat_history(chat_history) + assert isinstance(result, str) + + # Verify JSON structure + exported_data = json.loads(result) + assert "timestamp" in exported_data + assert "messages" in exported_data + assert len(exported_data["messages"]) == 2 + +@patch("builtins.open", new_callable=mock_open) +@patch("os.makedirs") +def test_save_feedback(mock_makedirs, mock_file): + feedback = { + "rating": 5, + "feedback": "Great service!", + "session_id": "test123" + } + + # Call the function + save_feedback(feedback) + + # Verify makedirs was called + mock_makedirs.assert_called_once() + + # Verify open was called with write mode + mock_file.assert_called_once() + + # Get the mock file handle + handle = mock_file() + + # Get what was written to the file + written_calls = handle.write.call_args_list + assert len(written_calls) > 0 + + # Combine all written data + written_data = ''.join(call[0][0] for call in written_calls) + + # Verify it's valid JSON + try: + parsed_data = json.loads(written_data) + assert parsed_data["rating"] == 5 + assert parsed_data["feedback"] == "Great service!" + assert parsed_data["session_id"] == "test123" + except json.JSONDecodeError as e: + pytest.fail(f"Invalid JSON written to file: {written_data}") + +@pytest.mark.parametrize("category,expected_path", [ + ("cyber", "Security-Focused Development"), + ("intelligence", "AI/ML Development"), + ("communications", "Frontend Development"), + ("maintenance", "Backend Development"), + ("unknown", "Full Stack Development"), +]) +def test_map_to_vwc_path_categories(category, expected_path): + result = map_to_vwc_path(category, []) + assert result["path"] == expected_path + assert isinstance(result["tech_focus"], list) + assert len(result["tech_focus"]) > 0 + +if __name__ == "__main__": + pytest.main(["-v"]) \ No newline at end of file