From 9d0fd563594a0f6d799f26e6007ad618d3adb3d2 Mon Sep 17 00:00:00 2001 From: Jerome Hardaway Date: Thu, 24 Oct 2024 14:29:52 -0400 Subject: [PATCH] Jeromehardaway/update code clean (#8) * Add feedback JSON file and update requirements for new dependencies * adding tests and refactor * pushed fixes * update action --- .github/workflows/unit-test.yml | 40 +- app.py | 481 +++++++++++++++++++++++++ feedback/feedback_20241023_212458.json | 6 + requirements.txt | 27 +- tests/conftest.py | 21 +- tests/test_app.py | 249 +++++++++++++ 6 files changed, 790 insertions(+), 34 deletions(-) create mode 100644 app.py create mode 100644 feedback/feedback_20241023_212458.json create mode 100644 tests/test_app.py diff --git a/.github/workflows/unit-test.yml b/.github/workflows/unit-test.yml index 5be6f89..2bd1bb5 100644 --- a/.github/workflows/unit-test.yml +++ b/.github/workflows/unit-test.yml @@ -1,23 +1,27 @@ -name: Run Unit Tests with Pytest - -on: [ push, pull_request ] +name: Python Tests +on: + push: + branches: [ main, jeromehardaway/update-code-clean ] + pull_request: + branches: [ main ] jobs: test: runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - name: Set up Python - uses: actions/setup-python@v5 - with: - python-version: '3.12' - - - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install -r requirements.txt - - - name: Run tests with coverage report - run: | - pytest --cov --cov-report=term-missing + - uses: actions/checkout@v4 + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: '3.12' + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + pip install pytest pytest-cov + - name: Run tests with coverage report + env: + TESTING: "true" + OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} + run: | + pytest --cov --cov-report=term-missing \ No newline at end of file diff --git a/app.py b/app.py new file mode 100644 index 0000000..cc9b0a3 --- /dev/null +++ b/app.py @@ -0,0 +1,481 @@ +import streamlit as st +import os +import logging +from typing import Dict, List +from datetime import datetime +from dotenv import load_dotenv +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() + +# 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 parse_mos_file(file_content: str) -> dict: + """ + 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 = {} + + # 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('.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.""" + + # 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" + ] + } + + # 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" + ] + } + } + + # Skill-based adjustments + skill_keywords = { + "programming": "software", + "database": "data", + "network": "communications", + "security": "cyber", + "analysis": "intelligence" + } + + # 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 + +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" + ] + } + } + +def get_chat_response(messages: List[Dict]) -> str: + """Get response from OpenAI chat completion.""" + try: + response = openai.chat.completions.create( + model="gpt-4o", + 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 + +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) + +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" + ) + + 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: + 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] + +def main(): + """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() \ No newline at end of file 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 6bd7fda..795b613 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,10 +1,19 @@ -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 +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 pytest -pytest-cov \ No newline at end of file +pytest-cov diff --git a/tests/conftest.py b/tests/conftest.py index 055f92c..9ae7f9e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,25 +1,32 @@ import io -import pytest import os +import sys +from pathlib import Path +import pytest -TEST_RESOURCE_DIR = f"{os.path.dirname(__file__)}/resources" +# Set up root directory path +ROOT_DIR = Path(__file__).parent.parent +sys.path.append(str(ROOT_DIR)) +# Define test resources directory +TEST_RESOURCE_DIR = Path(__file__).parent / "resources" def load_resource_file(file_name): + """Load a resource file and return its contents as BytesIO object.""" with open(file_name, "rb") as file: data = io.BytesIO(file.read()) return data - @pytest.fixture(scope="module") def file_resources(): + """Fixture to load all resource files from the test resources directory.""" library = {} - for filename in os.listdir(TEST_RESOURCE_DIR): - library[filename.split(".")[0]] = load_resource_file(f"{TEST_RESOURCE_DIR}/{filename}") + for file_path in TEST_RESOURCE_DIR.iterdir(): + library[file_path.stem] = load_resource_file(str(file_path)) yield library - def pytest_configure(config): + """Configure pytest with custom markers.""" config.addinivalue_line( "markers", "slow: marks tests as slow (deselect with '-m \"not slow\"')" - ) + ) \ 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