Add Watchtower trigger to CI workflow

- Add step to trigger Watchtower update after successful build
- Bump version to 1.1.1

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

.gitea/workflows/build.yml

@@ -25,3 +25,9 @@ jobs:
           tags: |
             git.schweitz.internal/jpmschweitzer/core-api:latest
             git.schweitz.internal/jpmschweitzer/core-api:${{ github.ref_name }}
+
+      - name: Trigger Watchtower update
+        if: success()
+        run: |
+          curl -sf -H "Authorization: Bearer ${{ secrets.WATCHTOWER_TOKEN }}" \
+            http://watchtower:8070/v1/update

CHANGELOG.md

@@ -5,6 +5,29 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.1.1] - 2024-12-14
+
+### Added
+
+- Watchtower trigger step in CI workflow for automatic container updates
+
+## [1.1.0] - 2024-12-14
+
+### Removed
+
+- Uptime Kuma integration (kuma_client.py deleted)
+- All /infrastructure/monitors endpoints
+- Kuma monitor pause/resume from service start/stop operations
+- Uptime percentage display from service control widget
+- Kuma-related configuration and credentials
+- Stale memory tests (memory functionality moved to core-ai service)
+
+### Changed
+
+- Service control widget now uses Portainer container status exclusively
+- Simplified widget-data endpoint response (removed monitors field)
+- Updated service start/stop to only manage containers via Portainer
+
 ## [1.0.0] - 2024-12-14
 
 ### Added

dev-requirements.txt

@@ -0,0 +1,8 @@
+# Development and testing dependencies
+-r requirements.txt
+
+# Testing
+pytest>=9.0.0
+pytest-asyncio>=0.24.0
+pytest-cov>=6.0.0
+httpx>=0.28.0  # For TestClient

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "core-api"
-version = "1.0.0"
+version = "1.1.1"
 description = "Core Code API - Infrastructure management and tools API"
 readme = "README.md"
 requires-python = ">=3.12"
@@ -8,3 +8,9 @@ license = {text = "MIT"}
 
 [tool.setuptools]
 packages = ["src"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]
+asyncio_mode = "auto"
+addopts = "-v"

src/clients/kuma_client.py

@@ -1,561 +0,0 @@
-"""
-Uptime Kuma Socket.IO Client
-
-Provides interface to Uptime Kuma via Socket.IO for monitor management.
-Also provides metrics API access for real-time status data.
-"""
-import socketio
-import asyncio
-import httpx
-import re
-from typing import Optional, Dict, List, Any
-from src.logging_config import get_logger
-from src.config import get_settings
-
-logger = get_logger(__name__)
-settings = get_settings()
-
-
-class KumaClient:
-    """
-    Socket.IO client for Uptime Kuma
-
-    Uses Socket.IO for real-time communication with Uptime Kuma.
-    """
-
-    def __init__(
-        self,
-        base_url: Optional[str] = None,
-        username: Optional[str] = None,
-        password: Optional[str] = None,
-        timeout: int = 30
-    ):
-        """
-        Initialize Kuma client
-
-        Args:
-            base_url: Kuma base URL (default from settings)
-            username: Kuma username (default from settings)
-            password: Kuma password (default from settings)
-            timeout: Request timeout in seconds
-        """
-        self.base_url = (base_url or settings.kuma_url).rstrip("/")
-        self.username = username or settings.kuma_username
-        self.password = password or settings.kuma_password
-        self.timeout = timeout
-
-        self.sio = socketio.AsyncClient(
-            reconnection=True,
-            reconnection_attempts=3,
-            reconnection_delay=1,
-        )
-        self._connected = False
-        self._authenticated = False
-        self._monitors_cache: Dict[int, Dict[str, Any]] = {}
-
-        if not self.username or not self.password:
-            logger.warning("Uptime Kuma credentials not configured")
-
-    async def _ensure_connected(self):
-        """Ensure we have an active connection and authentication"""
-        if not self._connected:
-            await self.connect()
-        if not self._authenticated:
-            await self.login()
-
-    async def connect(self):
-        """Connect to Uptime Kuma Socket.IO server"""
-        if self._connected:
-            return
-
-        try:
-            await self.sio.connect(self.base_url, transports=['websocket'])
-            self._connected = True
-            logger.info(f"Connected to Uptime Kuma at {self.base_url}")
-        except Exception as e:
-            logger.error(f"Failed to connect to Uptime Kuma: {e}")
-            raise
-
-    async def disconnect(self):
-        """Disconnect from Uptime Kuma"""
-        if self._connected:
-            await self.sio.disconnect()
-            self._connected = False
-            self._authenticated = False
-            logger.info("Disconnected from Uptime Kuma")
-
-    async def login(self):
-        """Authenticate with Uptime Kuma"""
-        if not self._connected:
-            await self.connect()
-
-        try:
-            # Uptime Kuma login event
-            login_response = await self.sio.call(
-                'login',
-                {
-                    'username': self.username,
-                    'password': self.password,
-                    'token': None
-                },
-                timeout=self.timeout
-            )
-
-            if login_response and login_response.get('ok'):
-                self._authenticated = True
-                logger.info("Successfully authenticated with Uptime Kuma")
-            else:
-                error_msg = login_response.get('msg', 'Unknown error') if login_response else 'No response'
-                raise Exception(f"Login failed: {error_msg}")
-
-        except Exception as e:
-            logger.error(f"Failed to authenticate with Uptime Kuma: {e}")
-            raise
-
-    async def health_check(self) -> bool:
-        """
-        Check if Uptime Kuma is accessible
-
-        Returns:
-            True if accessible, False otherwise
-        """
-        try:
-            await self._ensure_connected()
-            return self._authenticated
-        except Exception as e:
-            logger.error(f"Uptime Kuma health check failed: {e}")
-            return False
-
-    async def get_monitors(self) -> List[Dict[str, Any]]:
-        """
-        List all monitors with uptime data
-
-        Returns:
-            List of monitor configurations with uptime_24h field
-        """
-        await self._ensure_connected()
-
-        try:
-            # Storage for monitor list and uptime data received via events
-            monitor_list_data = {}
-            uptime_list_data = {}
-            monitor_event_received = asyncio.Event()
-            uptime_event_received = asyncio.Event()
-
-            # Register event handler for monitorList
-            @self.sio.event
-            async def monitorList(data):
-                nonlocal monitor_list_data
-                monitor_list_data = data
-                monitor_event_received.set()
-
-            # Register event handler for uptimeList (24h uptime percentages)
-            @self.sio.event
-            async def uptimeList(monitor_id, uptime_data):
-                nonlocal uptime_list_data
-                # uptime_data is typically a dict with time periods: {"24": 99.5, "720": 98.2, ...}
-                uptime_list_data[str(monitor_id)] = uptime_data
-                # Don't set event here as we'll get multiple calls
-
-            # Request monitor list - this triggers the server to send monitorList event
-            response = await self.sio.call('getMonitorList', timeout=self.timeout)
-            logger.info(f"getMonitorList call response: {response}")
-
-            # Wait for the monitorList event (with timeout)
-            try:
-                await asyncio.wait_for(monitor_event_received.wait(), timeout=5.0)
-                logger.info(f"Received monitorList event with {len(monitor_list_data)} items")
-
-                # Give time for uptimeList events to arrive
-                await asyncio.sleep(0.5)
-                logger.info(f"Received uptime data for {len(uptime_list_data)} monitors")
-            except asyncio.TimeoutError:
-                logger.warning("Timeout waiting for monitorList event")
-
-            # Process the monitor list data
-            if monitor_list_data and isinstance(monitor_list_data, dict):
-                monitors = []
-                for monitor_id, monitor_data in monitor_list_data.items():
-                    if isinstance(monitor_data, dict):
-                        monitor_data['id'] = int(monitor_id)
-
-                        # Add uptime data if available
-                        uptime_info = uptime_list_data.get(str(monitor_id), {})
-                        if isinstance(uptime_info, dict):
-                            # Uptime Kuma provides 24h uptime as key "24"
-                            monitor_data['uptime_24h'] = float(uptime_info.get('24', 0))
-                        else:
-                            monitor_data['uptime_24h'] = 0.0
-
-                        monitors.append(monitor_data)
-                        self._monitors_cache[int(monitor_id)] = monitor_data
-
-                logger.info(f"Found {len(monitors)} monitors total")
-                return monitors
-
-            logger.warning(f"No valid monitor data received")
-            return []
-
-        except Exception as e:
-            logger.error(f"Failed to get monitors: {e}", exc_info=True)
-            raise
-
-    async def get_monitor(self, monitor_id: int) -> Dict[str, Any]:
-        """
-        Get details of a specific monitor
-
-        Args:
-            monitor_id: Monitor identifier
-
-        Returns:
-            Monitor configuration details
-        """
-        await self._ensure_connected()
-
-        try:
-            response = await self.sio.call('getMonitor', monitor_id, timeout=self.timeout)
-
-            if response:
-                self._monitors_cache[monitor_id] = response
-                return response
-
-            raise Exception(f"Monitor {monitor_id} not found")
-
-        except Exception as e:
-            logger.error(f"Failed to get monitor {monitor_id}: {e}")
-            raise
-
-    async def find_monitor_by_name(self, name: str) -> Optional[Dict[str, Any]]:
-        """
-        Find a monitor by its name (case-insensitive)
-
-        Args:
-            name: Monitor name to search for
-
-        Returns:
-            Monitor object if found, None otherwise
-        """
-        monitors = await self.get_monitors()
-        name_lower = name.lower()
-
-        for monitor in monitors:
-            if monitor.get("name", "").lower() == name_lower:
-                return monitor
-
-        return None
-
-    async def find_monitors_by_tag(self, tag: str) -> List[Dict[str, Any]]:
-        """
-        Find all monitors with a specific tag
-
-        Args:
-            tag: Tag name to search for
-
-        Returns:
-            List of monitors with the tag
-        """
-        monitors = await self.get_monitors()
-        tagged_monitors = []
-
-        for monitor in monitors:
-            monitor_tags = monitor.get("tags", [])
-            if any(t.get("name", "").lower() == tag.lower() for t in monitor_tags):
-                tagged_monitors.append(monitor)
-
-        return tagged_monitors
-
-    async def pause_monitor(self, monitor_id: int) -> bool:
-        """
-        Pause a monitor (disable monitoring)
-
-        Args:
-            monitor_id: Monitor identifier
-
-        Returns:
-            True if successful
-        """
-        await self._ensure_connected()
-
-        try:
-            # Uptime Kuma pause event
-            response = await self.sio.call('pauseMonitor', monitor_id, timeout=self.timeout)
-
-            if response and response.get('ok'):
-                logger.info(f"Paused monitor {monitor_id}")
-                return True
-
-            error_msg = response.get('msg', 'Unknown error') if response else 'No response'
-            raise Exception(f"Failed to pause monitor: {error_msg}")
-
-        except Exception as e:
-            logger.error(f"Failed to pause monitor {monitor_id}: {e}")
-            raise
-
-    async def resume_monitor(self, monitor_id: int) -> bool:
-        """
-        Resume a monitor (enable monitoring)
-
-        Args:
-            monitor_id: Monitor identifier
-
-        Returns:
-            True if successful
-        """
-        await self._ensure_connected()
-
-        try:
-            # Uptime Kuma resume event
-            response = await self.sio.call('resumeMonitor', monitor_id, timeout=self.timeout)
-
-            if response and response.get('ok'):
-                logger.info(f"Resumed monitor {monitor_id}")
-                return True
-
-            error_msg = response.get('msg', 'Unknown error') if response else 'No response'
-            raise Exception(f"Failed to resume monitor: {error_msg}")
-
-        except Exception as e:
-            logger.error(f"Failed to resume monitor {monitor_id}: {e}")
-            raise
-
-    async def pause_monitor_by_name(self, name: str) -> bool:
-        """
-        Pause a monitor by its name
-
-        Args:
-            name: Monitor name
-
-        Returns:
-            True if successful, False if monitor not found
-        """
-        monitor = await self.find_monitor_by_name(name)
-        if not monitor:
-            logger.warning(f"Monitor '{name}' not found")
-            return False
-
-        await self.pause_monitor(monitor["id"])
-        return True
-
-    async def resume_monitor_by_name(self, name: str) -> bool:
-        """
-        Resume a monitor by its name
-
-        Args:
-            name: Monitor name
-
-        Returns:
-            True if successful, False if monitor not found
-        """
-        monitor = await self.find_monitor_by_name(name)
-        if not monitor:
-            logger.warning(f"Monitor '{name}' not found")
-            return False
-
-        await self.resume_monitor(monitor["id"])
-        return True
-
-    async def add_monitor(self, monitor_config: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        Create a new monitor
-
-        Args:
-            monitor_config: Monitor configuration dict
-
-        Returns:
-            Created monitor details including ID
-        """
-        await self._ensure_connected()
-
-        try:
-            # Uptime Kuma add monitor event
-            response = await self.sio.call('add', monitor_config, timeout=self.timeout)
-
-            if response and response.get('ok'):
-                monitor_id = response.get('monitorID')
-                logger.info(f"Created monitor '{monitor_config.get('name')}' with ID {monitor_id}")
-
-                # Get full monitor details
-                monitor = await self.get_monitor(monitor_id)
-                return monitor
-
-            error_msg = response.get('msg', 'Unknown error') if response else 'No response'
-            raise Exception(f"Failed to create monitor: {error_msg}")
-
-        except Exception as e:
-            logger.error(f"Failed to create monitor '{monitor_config.get('name')}': {e}")
-            raise
-
-    async def update_monitor(self, monitor_id: int, monitor_config: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        Update an existing monitor
-
-        Args:
-            monitor_id: Monitor identifier
-            monitor_config: Updated monitor configuration
-
-        Returns:
-            Updated monitor details
-        """
-        await self._ensure_connected()
-
-        try:
-            # Ensure ID is in the config
-            monitor_config['id'] = monitor_id
-
-            # Uptime Kuma edit monitor event
-            response = await self.sio.call('editMonitor', monitor_config, timeout=self.timeout)
-
-            if response and response.get('ok'):
-                logger.info(f"Updated monitor {monitor_id}")
-
-                # Get updated monitor details
-                monitor = await self.get_monitor(monitor_id)
-                return monitor
-
-            error_msg = response.get('msg', 'Unknown error') if response else 'No response'
-            raise Exception(f"Failed to update monitor: {error_msg}")
-
-        except Exception as e:
-            logger.error(f"Failed to update monitor {monitor_id}: {e}")
-            raise
-
-    async def delete_monitor(self, monitor_id: int) -> bool:
-        """
-        Delete a monitor
-
-        Args:
-            monitor_id: Monitor identifier
-
-        Returns:
-            True if successful
-        """
-        await self._ensure_connected()
-
-        try:
-            # Uptime Kuma delete monitor event
-            response = await self.sio.call('deleteMonitor', monitor_id, timeout=self.timeout)
-
-            if response and response.get('ok'):
-                logger.info(f"Deleted monitor {monitor_id}")
-
-                # Remove from cache
-                self._monitors_cache.pop(monitor_id, None)
-                return True
-
-            error_msg = response.get('msg', 'Unknown error') if response else 'No response'
-            raise Exception(f"Failed to delete monitor: {error_msg}")
-
-        except Exception as e:
-            logger.error(f"Failed to delete monitor {monitor_id}: {e}")
-            raise
-
-    async def delete_monitor_by_name(self, name: str) -> bool:
-        """
-        Delete a monitor by its name
-
-        Args:
-            name: Monitor name
-
-        Returns:
-            True if successful, False if monitor not found
-        """
-        monitor = await self.find_monitor_by_name(name)
-        if not monitor:
-            logger.warning(f"Monitor '{name}' not found")
-            return False
-
-        await self.delete_monitor(monitor["id"])
-        return True
-
-    async def get_metrics_status(self) -> Dict[str, Dict[str, Any]]:
-        """
-        Get monitor status from Prometheus metrics endpoint
-
-        This is simpler and more reliable than Socket.IO for getting current status.
-        Returns real-time UP/DOWN status but not historical uptime percentages.
-
-        Returns:
-            Dict mapping monitor names to status info:
-            {
-                "Portainer": {
-                    "status": 1,  # 1=UP, 0=DOWN, 2=PENDING, 3=MAINTENANCE
-                    "response_time": 5,  # ms
-                    "monitor_type": "http",
-                    "url": "http://192.168.86.149:8001"
-                },
-                ...
-            }
-        """
-        try:
-            # Use API key authentication
-            api_key = settings.kuma_api_key
-            if not api_key:
-                logger.warning("Kuma API key not configured")
-                return {}
-
-            # Fetch metrics with HTTP Basic Auth (empty username, API key as password)
-            async with httpx.AsyncClient(timeout=10.0) as client:
-                response = await client.get(
-                    f"{self.base_url}/metrics",
-                    auth=("", api_key)
-                )
-                response.raise_for_status()
-                metrics_text = response.text
-
-            # Parse Prometheus format metrics
-            # Format: metric_name{label1="value1",label2="value2"} value
-            monitor_data = {}
-
-            # Parse monitor_status lines
-            status_pattern = r'monitor_status\{monitor_name="([^"]+)",.*?\} (\d+)'
-            for match in re.finditer(status_pattern, metrics_text):
-                monitor_name = match.group(1)
-                status = int(match.group(2))
-
-                if monitor_name not in monitor_data:
-                    monitor_data[monitor_name] = {}
-                monitor_data[monitor_name]['status'] = status
-
-            # Parse monitor_response_time lines
-            response_pattern = r'monitor_response_time\{monitor_name="([^"]+)",monitor_type="([^"]+)",monitor_url="([^"]+)",.*?\} ([\d.]+)'
-            for match in re.finditer(response_pattern, metrics_text):
-                monitor_name = match.group(1)
-                monitor_type = match.group(2)
-                monitor_url = match.group(3)
-                response_time = float(match.group(4))
-
-                if monitor_name not in monitor_data:
-                    monitor_data[monitor_name] = {}
-                monitor_data[monitor_name].update({
-                    'response_time': response_time,
-                    'monitor_type': monitor_type,
-                    'url': monitor_url
-                })
-
-            logger.info(f"Fetched metrics for {len(monitor_data)} monitors")
-            return monitor_data
-
-        except Exception as e:
-            logger.error(f"Failed to fetch metrics: {e}")
-            return {}
-
-    async def __aenter__(self):
-        """Async context manager entry"""
-        await self._ensure_connected()
-        return self
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        """Async context manager exit"""
-        await self.disconnect()
-
-
-# Singleton instance
-_kuma_client: Optional[KumaClient] = None
-
-
-def get_kuma_client() -> KumaClient:
-    """Get singleton Kuma client instance"""
-    global _kuma_client
-    if _kuma_client is None:
-        _kuma_client = KumaClient()
-    return _kuma_client

src/config.py

@@ -25,7 +25,6 @@ try:
     from src.credentials import (
         PORTAINER_URL, PORTAINER_API_KEY,
         NPM_URL, NPM_EMAIL, NPM_PASSWORD,
-        KUMA_URL, KUMA_USERNAME, KUMA_PASSWORD, KUMA_API_KEY,
         BRAVE_SEARCH_API_KEY,
         GOOGLE_SEARCH_API_KEY, GOOGLE_SEARCH_ENGINE_ID
     )
@@ -37,10 +36,6 @@ except ImportError:
     NPM_URL = "http://localhost:81"
     NPM_EMAIL = ""
     NPM_PASSWORD = ""
-    KUMA_URL = "http://localhost:3001"
-    KUMA_USERNAME = ""
-    KUMA_PASSWORD = ""
-    KUMA_API_KEY = ""
     BRAVE_SEARCH_API_KEY = ""
     GOOGLE_SEARCH_API_KEY = ""
     GOOGLE_SEARCH_ENGINE_ID = ""
@@ -127,11 +122,6 @@ class Settings(BaseSettings):
     npm_email: str = NPM_EMAIL
     npm_password: str = NPM_PASSWORD
 
-    kuma_url: str = KUMA_URL
-    kuma_username: str = KUMA_USERNAME
-    kuma_password: str = KUMA_PASSWORD
-    kuma_api_key: str = KUMA_API_KEY
-
     # Core-AI Service (AI performance metrics)
     core_ai_base_url: str = "http://core-ai:8086"
 

src/controllers/infrastructure_controller.py

@@ -11,7 +11,6 @@ from pydantic import BaseModel, field_validator
 from src.controllers.base import BaseController
 from src.clients.portainer_client import get_portainer_client
 from src.clients.npm_client import get_npm_client
-from src.clients.kuma_client import get_kuma_client
 from src.logging_config import get_logger
 from src import service_groups
 from src.auth.oidc import get_admin_user, get_forward_auth_admin
@@ -963,7 +962,7 @@ class InfrastructureController(BaseController):
             "/services/{name}/stop",
             response_model=OperationResult,
             summary="Stop a service or service group",
-            description="Stop a service or service group by pausing monitors and stopping containers. Requires admin authentication when accessed externally via api.schweitz.net."
+            description="Stop a service or service group by stopping containers. Requires admin authentication when accessed externally via api.schweitz.net."
         )
         async def stop_service(
             name: str,
@@ -974,8 +973,7 @@ class InfrastructureController(BaseController):
 
             This will:
             1. Validate service can be stopped (not always-on)
-            2. Pause Uptime Kuma monitors for all services in group
-            3. Stop the Portainer stack(s)
+            2. Stop the Portainer stack containers
 
             Args:
                 name: Service or group name
@@ -984,7 +982,6 @@ class InfrastructureController(BaseController):
                 Operation result with details
             """
             portainer = get_portainer_client()
-            kuma = get_kuma_client()
 
             try:
                 # Get all services in the group
@@ -997,24 +994,13 @@ class InfrastructureController(BaseController):
 
                 results = {
                     "stopped_services": [],
-                    "paused_monitors": [],
                     "errors": []
                 }
 
                 # Stop each service
                 for service_name in services:
                     try:
-                        # 1. Pause Uptime Kuma monitor
-                        try:
-                            monitor_paused = await kuma.pause_monitor_by_name(service_name)
-                            if monitor_paused:
-                                results["paused_monitors"].append(service_name)
-                                logger.info(f"Paused Kuma monitor for {service_name}")
-                        except Exception as e:
-                            logger.warning(f"Failed to pause Kuma monitor for {service_name}: {e}")
-                            results["errors"].append(f"Kuma pause failed for {service_name}: {str(e)}")
-
-                        # 2. Stop Portainer stack
+                        # Stop Portainer stack containers
                         stacks = await portainer.get_stacks()
                         stack = next(
                             (s for s in stacks if s.get("Name", "").lower() == service_name.lower()),
@@ -1025,9 +1011,6 @@ class InfrastructureController(BaseController):
                             stack_id = stack.get("Id")
                             endpoint_id = stack.get("EndpointId")
 
-                            # Stop stack by deleting it (Portainer doesn't have a "stop" operation)
-                            # Note: This is destructive. For a gentler approach, we'd need to use docker compose stop
-                            # Let's use docker API instead
                             logger.info(f"Stopping containers for stack: {service_name}")
 
                             # Get containers for this stack
@@ -1078,7 +1061,7 @@ class InfrastructureController(BaseController):
             "/services/{name}/start",
             response_model=OperationResult,
             summary="Start a service or service group",
-            description="Start a service or service group by starting containers and resuming monitors. Requires admin authentication when accessed externally via api.schweitz.net."
+            description="Start a service or service group by starting containers. Requires admin authentication when accessed externally via api.schweitz.net."
         )
         async def start_service(
             name: str,
@@ -1088,8 +1071,7 @@ class InfrastructureController(BaseController):
             Start a service or service group
 
             This will:
-            1. Start the Portainer stack(s)
-            2. Resume Uptime Kuma monitors for all services in group
+            1. Start the Portainer stack containers
 
             Args:
                 name: Service or group name
@@ -1098,7 +1080,6 @@ class InfrastructureController(BaseController):
                 Operation result with details
             """
             portainer = get_portainer_client()
-            kuma = get_kuma_client()
 
             try:
                 # Get all services in the group
@@ -1106,14 +1087,13 @@ class InfrastructureController(BaseController):
 
                 results = {
                     "started_services": [],
-                    "resumed_monitors": [],
                     "errors": []
                 }
 
                 # Start each service
                 for service_name in services:
                     try:
-                        # 1. Start Portainer stack (start containers)
+                        # Start Portainer stack containers
                         stacks = await portainer.get_stacks()
                         stack = next(
                             (s for s in stacks if s.get("Name", "").lower() == service_name.lower()),
@@ -1147,16 +1127,6 @@ class InfrastructureController(BaseController):
                             })
                             logger.info(f"Started service: {service_name}")
 
-                            # 2. Resume Uptime Kuma monitor
-                            try:
-                                monitor_resumed = await kuma.resume_monitor_by_name(service_name)
-                                if monitor_resumed:
-                                    results["resumed_monitors"].append(service_name)
-                                    logger.info(f"Resumed Kuma monitor for {service_name}")
-                            except Exception as e:
-                                logger.warning(f"Failed to resume Kuma monitor for {service_name}: {e}")
-                                results["errors"].append(f"Kuma resume failed for {service_name}: {str(e)}")
-
                         else:
                             results["errors"].append(f"Stack not found: {service_name}")
 
@@ -1181,8 +1151,6 @@ class InfrastructureController(BaseController):
                 logger.error(f"Failed to start service group '{name}': {e}")
                 raise HTTPException(status_code=500, detail=str(e))
 
-        # ===== Monitoring Endpoints =====
-
         @router.get(
             "/widget-data",
             summary="Get combined data for service control widget",
@@ -1190,11 +1158,10 @@ class InfrastructureController(BaseController):
         )
         async def get_widget_data():
             """
-            Get combined service and monitor data for the widget
+            Get combined service data for the widget
 
             Returns all data needed by service-control widget in a single call:
             - Service list with status and container counts
-            - Monitor list with uptime percentages
             - Service groups and always-on list
 
             This endpoint is designed for browser-based widgets to avoid
@@ -1202,7 +1169,6 @@ class InfrastructureController(BaseController):
             """
             try:
                 portainer = get_portainer_client()
-                kuma = get_kuma_client()
                 npm = get_npm_client()
 
                 # Fetch services (same logic as /services endpoint)
@@ -1252,38 +1218,9 @@ class InfrastructureController(BaseController):
                         "containers_total": containers_total
                     })
 
-                # Fetch monitors with real-time status from metrics endpoint
-                monitors_list = []
-                try:
-                    # Get real-time status from Prometheus metrics
-                    metrics_data = await kuma.get_metrics_status()
-
-                    for monitor_name, monitor_info in metrics_data.items():
-                        # Status: 1=UP, 0=DOWN, 2=PENDING, 3=MAINTENANCE
-                        status = monitor_info.get('status', 0)
-
-                        # Convert status to simple up/down for widget
-                        # Treat UP (1) as 100%, anything else as 0%
-                        status_percentage = 100.0 if status == 1 else 0.0
-
-                        monitors_list.append({
-                            "id": None,  # Not available from metrics
-                            "name": monitor_name,
-                            "uptime_24h": status_percentage,  # Current status as percentage
-                            "active": True,  # Assume active if in metrics
-                            "status": status,  # 1=UP, 0=DOWN, 2=PENDING, 3=MAINTENANCE
-                            "response_time": monitor_info.get('response_time', 0)
-                        })
-
-                    logger.info(f"Fetched status for {len(monitors_list)} monitors from metrics")
-                except Exception as e:
-                    logger.warning(f"Failed to fetch monitors: {e}")
-                    # Continue without monitor data rather than failing
-
                 return {
                     "success": True,
                     "services": services,
-                    "monitors": monitors_list,
                     "service_groups": {
                         "groups": service_groups.list_service_groups(),
                         "always_on": list(service_groups.ALWAYS_ON_SERVICES),
@@ -1295,169 +1232,6 @@ class InfrastructureController(BaseController):
                 logger.error(f"Failed to fetch widget data: {e}")
                 raise HTTPException(status_code=500, detail=f"Failed to fetch widget data: {str(e)}")
 
-        @router.get(
-            "/monitors",
-            summary="List all monitors",
-            response_model=Dict[str, Any]
-        )
-        async def list_monitors():
-            """
-            List all Uptime Kuma monitors
-
-            Returns:
-                List of monitors with their configurations
-            """
-            try:
-                kuma = get_kuma_client()
-                monitors = await kuma.get_monitors()
-
-                return {
-                    "success": True,
-                    "monitors": monitors,
-                    "total": len(monitors)
-                }
-
-            except Exception as e:
-                logger.error(f"Failed to list monitors: {e}")
-                raise HTTPException(status_code=500, detail=f"Failed to list monitors: {str(e)}")
-
-        @router.post(
-            "/monitors",
-            summary="Create a new monitor",
-            description="Create a new Uptime Kuma monitor. Requires admin authentication.",
-            response_model=Dict[str, Any]
-        )
-        async def create_monitor(
-            monitor_config: Dict[str, Any],
-            user: Dict = Depends(get_admin_user)
-        ):
-            """
-            Create a new Uptime Kuma monitor
-
-            Args:
-                monitor_config: Monitor configuration (name, type, hostname, port, etc.)
-
-            Returns:
-                Created monitor details including ID
-            """
-            try:
-                kuma = get_kuma_client()
-                created_monitor = await kuma.add_monitor(monitor_config)
-
-                return {
-                    "success": True,
-                    "message": f"Monitor '{monitor_config.get('name')}' created successfully",
-                    "monitor": created_monitor
-                }
-
-            except Exception as e:
-                logger.error(f"Failed to create monitor: {e}")
-                raise HTTPException(status_code=500, detail=f"Failed to create monitor: {str(e)}")
-
-        @router.get(
-            "/monitors/{monitor_id}",
-            summary="Get monitor details",
-            response_model=Dict[str, Any]
-        )
-        async def get_monitor(monitor_id: int):
-            """
-            Get details of a specific monitor
-
-            Args:
-                monitor_id: Monitor identifier
-
-            Returns:
-                Monitor configuration and status
-            """
-            try:
-                kuma = get_kuma_client()
-                monitor = await kuma.get_monitor(monitor_id)
-
-                return {
-                    "success": True,
-                    "monitor": monitor
-                }
-
-            except Exception as e:
-                logger.error(f"Failed to get monitor {monitor_id}: {e}")
-                raise HTTPException(status_code=404, detail=f"Monitor {monitor_id} not found: {str(e)}")
-
-        @router.put(
-            "/monitors/{monitor_id}",
-            summary="Update a monitor",
-            description="Update an existing Uptime Kuma monitor. Requires admin authentication.",
-            response_model=Dict[str, Any]
-        )
-        async def update_monitor(
-            monitor_id: int,
-            updates: Dict[str, Any],
-            user: Dict = Depends(get_admin_user)
-        ):
-            """
-            Update an existing monitor
-
-            Args:
-                monitor_id: Monitor identifier
-                updates: Fields to update
-
-            Returns:
-                Updated monitor details
-            """
-            try:
-                kuma = get_kuma_client()
-
-                # Get existing monitor
-                existing = await kuma.get_monitor(monitor_id)
-
-                # Merge updates
-                monitor_config = existing.copy()
-                monitor_config.update(updates)
-
-                # Update monitor
-                updated_monitor = await kuma.update_monitor(monitor_id, monitor_config)
-
-                return {
-                    "success": True,
-                    "message": f"Monitor {monitor_id} updated successfully",
-                    "monitor": updated_monitor
-                }
-
-            except Exception as e:
-                logger.error(f"Failed to update monitor {monitor_id}: {e}")
-                raise HTTPException(status_code=500, detail=f"Failed to update monitor: {str(e)}")
-
-        @router.delete(
-            "/monitors/{monitor_id}",
-            summary="Delete a monitor",
-            description="Delete an Uptime Kuma monitor. Requires admin authentication.",
-            response_model=Dict[str, Any]
-        )
-        async def delete_monitor(
-            monitor_id: int,
-            user: Dict = Depends(get_admin_user)
-        ):
-            """
-            Delete a monitor
-
-            Args:
-                monitor_id: Monitor identifier
-
-            Returns:
-                Success confirmation
-            """
-            try:
-                kuma = get_kuma_client()
-                await kuma.delete_monitor(monitor_id)
-
-                return {
-                    "success": True,
-                    "message": f"Monitor {monitor_id} deleted successfully"
-                }
-
-            except Exception as e:
-                logger.error(f"Failed to delete monitor {monitor_id}: {e}")
-                raise HTTPException(status_code=500, detail=f"Failed to delete monitor: {str(e)}")
-
         # ========================================================================
         # Container Management Endpoints (for core-ai infrastructure tools)
         # ========================================================================

src/credentials.example.py

@@ -17,8 +17,3 @@ PORTAINER_API_KEY = "ptr_your_api_token_here"  # Create in Portainer UI: User me
 NPM_URL = "http://localhost:81"
 NPM_EMAIL = "admin@example.com"
 NPM_PASSWORD = "your_password_here"
-
-# Uptime Kuma Configuration
-KUMA_URL = "http://localhost:3001"
-KUMA_USERNAME = "admin"
-KUMA_PASSWORD = "your_password_here"

src/service_groups.py

@@ -10,7 +10,6 @@ ALWAYS_ON_SERVICES: Set[str] = {
     "portainer",
     "nginx-proxy-manager",
     "core-api",
-    "uptime-kuma",
     "organizr",
     "headscale",
     "watchtower",

static/widgets/service-control.html

@@ -71,14 +71,14 @@
             font-weight: 600;
             color: #fff;
             text-transform: capitalize;
-            min-width: 300px;
+            min-width: 200px;
         }
 
         .service-status {
             display: flex;
             align-items: center;
             gap: 8px;
-            min-width: 120px;
+            min-width: 180px;
         }
 
         .status-indicator {
@@ -103,54 +103,6 @@
             color: #a0a0a0;
         }
 
-        .uptime-status {
-            display: flex;
-            align-items: center;
-            gap: 8px;
-            min-width: 150px;
-            padding: 4px 10px;
-            background: rgba(0, 0, 0, 0.2);
-            border-radius: 4px;
-            cursor: pointer;
-            transition: background 0.2s;
-            text-decoration: none;
-            color: inherit;
-        }
-
-        .uptime-status:hover {
-            background: rgba(0, 0, 0, 0.4);
-        }
-
-        .uptime-percentage {
-            font-size: 13px;
-            font-weight: 600;
-        }
-
-        .uptime-percentage.excellent {
-            color: #48bb78;
-        }
-
-        .uptime-percentage.good {
-            color: #68d391;
-        }
-
-        .uptime-percentage.warning {
-            color: #ed8936;
-        }
-
-        .uptime-percentage.critical {
-            color: #f56565;
-        }
-
-        .uptime-percentage.unknown {
-            color: #718096;
-        }
-
-        .uptime-icon {
-            font-size: 11px;
-            color: #a0a0a0;
-        }
-
         .service-right {
             display: flex;
             align-items: center;
@@ -248,10 +200,6 @@
                 min-width: 100px;
             }
 
-            .uptime-status {
-                min-width: 100px;
-            }
-
             .service-right {
                 width: 100%;
                 justify-content: flex-end;
@@ -264,12 +212,12 @@
         <div id="error-container"></div>
 
         <div class="section">
-            <div class="section-header">🎛️ Stoppable Services</div>
+            <div class="section-header">Stoppable Services</div>
             <div id="stoppable-container" class="loading">Loading services...</div>
         </div>
 
         <div class="section">
-            <div class="section-header">🔒 Always-On Infrastructure</div>
+            <div class="section-header">Always-On Infrastructure</div>
             <div id="always-on-container" class="loading">Loading infrastructure...</div>
         </div>
     </div>
@@ -277,11 +225,9 @@
     <script>
         // Use relative URL to work in any context (iframe, direct access, etc.)
         const API_BASE = '';
-        const KUMA_BASE = window.location.protocol + '//' + window.location.hostname + ':3001';
 
         let services = [];
         let alwaysOnServices = [];
-        let monitors = {};
 
         async function fetchData() {
             try {
@@ -306,27 +252,13 @@
                     alwaysOnServices = data.service_groups.always_on;
                 }
 
-                // Build monitors map
-                const monitorsMap = {};
-                if (data.monitors) {
-                    data.monitors.forEach(monitor => {
-                        const name = monitor.name.toLowerCase().replace(/[^a-z0-9]/g, '-');
-                        monitorsMap[name] = {
-                            id: monitor.id,
-                            uptime_24h: monitor.uptime_24h || 0,
-                            active: monitor.active !== false
-                        };
-                    });
-                }
-                monitors = monitorsMap;
-
                 renderServices();
                 document.getElementById('error-container').innerHTML = '';
 
             } catch (error) {
                 console.error('Error fetching data:', error);
                 document.getElementById('error-container').innerHTML =
-                    `<div class="error">❌ Failed to connect to API: ${error.message}</div>`;
+                    `<div class="error">Failed to connect to API: ${error.message}</div>`;
             }
         }
 
@@ -334,43 +266,9 @@
             return alwaysOnServices.includes(serviceName.toLowerCase());
         }
 
-        function getUptimeInfo(serviceName) {
-            const monitorKey = serviceName.toLowerCase().replace(/[^a-z0-9]/g, '-');
-            const monitor = monitors[monitorKey];
-
-            if (!monitor) {
-                return {
-                    percentage: 0,
-                    class: 'unknown',
-                    text: 'No monitor',
-                    id: null
-                };
-            }
-
-            const uptime = monitor.uptime_24h;
-            let className = 'unknown';
-
-            if (uptime >= 99.5) className = 'excellent';
-            else if (uptime >= 95) className = 'good';
-            else if (uptime >= 90) className = 'warning';
-            else if (uptime > 0) className = 'critical';
-
-            return {
-                percentage: uptime,
-                class: className,
-                text: uptime > 0 ? `${uptime.toFixed(1)}% ↑` : 'Down',
-                id: monitor.id
-            };
-        }
-
         function renderServiceRow(service) {
             const isRunning = service.containers_running > 0;
             const alwaysOn = isAlwaysOn(service.name);
-            const uptime = getUptimeInfo(service.name);
-
-            const kumaLink = uptime.id ?
-                `${KUMA_BASE}/dashboard/${uptime.id}` :
-                KUMA_BASE;
 
             return `
                 <div class="service-row" data-service="${service.name}">
@@ -385,10 +283,6 @@
                                 ${isRunning ? `Running (${service.containers_running}/${service.containers_total})` : 'Stopped'}
                             </span>
                         </div>
-                        <a href="${kumaLink}" target="_blank" class="uptime-status" title="View in Uptime Kuma">
-                            <span class="uptime-icon">📊</span>
-                            <span class="uptime-percentage ${uptime.class}">${uptime.text}</span>
-                        </a>
                     </div>
                     <div class="service-right">
                         <button class="btn btn-start"

tests/__init__.py

@@ -0,0 +1 @@
+# Tests package

tests/test_config.py

@@ -0,0 +1,98 @@
+"""Tests for config module."""
+import pytest
+from src.config import (
+    __version__,
+    Settings,
+    get_settings,
+    _get_version_from_pyproject,
+)
+
+
+class TestVersion:
+    """Test version loading from pyproject.toml."""
+
+    def test_version_is_loaded(self):
+        """Version should be loaded from pyproject.toml."""
+        assert __version__ is not None
+        assert isinstance(__version__, str)
+
+    def test_version_format(self):
+        """Version should follow semver format."""
+        parts = __version__.split(".")
+        assert len(parts) >= 2, "Version should have at least major.minor"
+        assert all(p.isdigit() for p in parts), "Version parts should be numeric"
+
+    def test_version_matches_settings(self):
+        """Settings app_version should match module version."""
+        settings = get_settings()
+        assert settings.app_version == __version__
+
+
+class TestGetVersionFromPyproject:
+    """Test the version loading function."""
+
+    def test_returns_string(self):
+        """Should return a string version."""
+        version = _get_version_from_pyproject()
+        assert isinstance(version, str)
+
+    def test_returns_valid_version(self):
+        """Should return a valid version (not 0.0.0 if file exists)."""
+        version = _get_version_from_pyproject()
+        # Since pyproject.toml exists, version should not be fallback
+        assert version != "0.0.0"
+
+
+class TestSettings:
+    """Test Settings configuration class."""
+
+    def test_settings_has_app_name(self):
+        """Settings should have app_name."""
+        settings = get_settings()
+        assert settings.app_name == "Core Code API"
+
+    def test_settings_has_version(self):
+        """Settings should have app_version."""
+        settings = get_settings()
+        assert settings.app_version is not None
+
+    def test_settings_default_host(self):
+        """Settings should have default host."""
+        settings = get_settings()
+        assert settings.host == "0.0.0.0"
+
+    def test_settings_default_port(self):
+        """Settings should have default port."""
+        settings = get_settings()
+        assert settings.port == 8083
+
+    def test_no_kuma_settings(self):
+        """Settings should not have Kuma-related attributes."""
+        settings = get_settings()
+        assert not hasattr(settings, "kuma_url")
+        assert not hasattr(settings, "kuma_username")
+        assert not hasattr(settings, "kuma_password")
+        assert not hasattr(settings, "kuma_api_key")
+
+
+class TestGetSettings:
+    """Test get_settings function."""
+
+    def test_returns_settings_instance(self):
+        """Should return a Settings instance."""
+        settings = get_settings()
+        assert isinstance(settings, Settings)
+
+    def test_returns_cached_instance(self):
+        """Should return the same cached instance."""
+        settings1 = get_settings()
+        settings2 = get_settings()
+        assert settings1 is settings2
+
+    def test_model_aliases_property(self):
+        """Model aliases property should return dict."""
+        settings = get_settings()
+        aliases = settings.model_aliases
+        assert isinstance(aliases, dict)
+        assert "gpt-3.5-turbo" in aliases
+        assert "gpt-4" in aliases

tests/test_health.py

@@ -0,0 +1,123 @@
+"""Tests for health endpoints."""
+import pytest
+from fastapi.testclient import TestClient
+from unittest.mock import patch, AsyncMock
+
+from src.main import app
+
+
+@pytest.fixture
+def client():
+    """Create a test client."""
+    return TestClient(app)
+
+
+class TestRootEndpoint:
+    """Test root endpoint."""
+
+    def test_root_returns_200(self, client):
+        """Root endpoint should return 200."""
+        response = client.get("/")
+        assert response.status_code == 200
+
+    def test_root_returns_service_info(self, client):
+        """Root endpoint should return service information."""
+        response = client.get("/")
+        data = response.json()
+
+        assert "service" in data
+        assert "version" in data
+        assert "status" in data
+        assert data["service"] == "Core Code API"
+        assert data["status"] == "healthy"
+
+    def test_root_returns_documentation_links(self, client):
+        """Root endpoint should return documentation links."""
+        response = client.get("/")
+        data = response.json()
+
+        assert "documentation" in data
+        assert "swagger_ui" in data["documentation"]
+        assert "redoc" in data["documentation"]
+
+    def test_root_returns_endpoints(self, client):
+        """Root endpoint should return available endpoints."""
+        response = client.get("/")
+        data = response.json()
+
+        assert "endpoints" in data
+        assert "health" in data["endpoints"]
+
+
+class TestHealthEndpoint:
+    """Test /health endpoint."""
+
+    @patch("src.controllers.health_controller.get_ollama_client")
+    def test_health_returns_200_when_ollama_healthy(self, mock_get_ollama, client):
+        """Health endpoint should return 200 when Ollama is healthy."""
+        mock_client = AsyncMock()
+        mock_client.health_check.return_value = True
+        mock_get_ollama.return_value = mock_client
+
+        response = client.get("/health")
+        assert response.status_code == 200
+
+    @patch("src.controllers.health_controller.get_ollama_client")
+    def test_health_returns_status(self, mock_get_ollama, client):
+        """Health endpoint should return status information."""
+        mock_client = AsyncMock()
+        mock_client.health_check.return_value = True
+        mock_get_ollama.return_value = mock_client
+
+        response = client.get("/health")
+        data = response.json()
+
+        assert "status" in data
+        assert "ollama_connected" in data
+
+    @patch("src.controllers.health_controller.get_ollama_client")
+    def test_health_returns_ollama_connected_true(self, mock_get_ollama, client):
+        """Health should report Ollama connected when healthy."""
+        mock_client = AsyncMock()
+        mock_client.health_check.return_value = True
+        mock_get_ollama.return_value = mock_client
+
+        response = client.get("/health")
+        data = response.json()
+
+        assert data["ollama_connected"] is True
+
+    @patch("src.controllers.health_controller.get_ollama_client")
+    def test_health_returns_ollama_connected_false(self, mock_get_ollama, client):
+        """Health should report Ollama disconnected when unhealthy."""
+        mock_client = AsyncMock()
+        mock_client.health_check.return_value = False
+        mock_get_ollama.return_value = mock_client
+
+        response = client.get("/health")
+        data = response.json()
+
+        assert data["ollama_connected"] is False
+
+
+class TestOpenAPIEndpoint:
+    """Test OpenAPI documentation endpoints."""
+
+    def test_openapi_spec_available(self, client):
+        """OpenAPI spec should be available."""
+        response = client.get("/openapi.json")
+        assert response.status_code == 200
+
+        data = response.json()
+        assert "openapi" in data
+        assert "info" in data
+
+    def test_swagger_ui_available(self, client):
+        """Swagger UI should be available."""
+        response = client.get("/docs")
+        assert response.status_code == 200
+
+    def test_redoc_available(self, client):
+        """ReDoc should be available."""
+        response = client.get("/redoc")
+        assert response.status_code == 200

tests/test_memory_integration.py

@@ -1,269 +0,0 @@
-"""
-Integration tests for Phase 2 Memory System
-
-Tests the complete memory stack:
-- Tier 1: ConversationBufferMemory
-- Tier 2/3: QdrantConversationMemory
-- Embedding Client
-"""
-import asyncio
-import pytest
-from datetime import datetime
-from src.memory import (
-    ConversationBufferMemory,
-    QdrantConversationMemory,
-    ConversationTurn,
-    MessageRole,
-    TokenUsage,
-    get_buffer_memory,
-    get_qdrant_memory
-)
-from src.models.embeddings import get_embedding_client
-
-
-class TestEmbeddingClient:
-    """Test embedding generation"""
-
-    def test_embedding_client_init(self):
-        """Test embedding client initialization"""
-        client = get_embedding_client()
-        assert client is not None
-        assert client.dimension == 384
-        print(f"✓ Embedding client initialized: {client.model_name}")
-
-    def test_single_embedding(self):
-        """Test single text embedding"""
-        client = get_embedding_client()
-        text = "Hello, this is a test message for embedding generation"
-
-        embedding = client.embed_text(text)
-
-        assert isinstance(embedding, list)
-        assert len(embedding) == 384
-        assert all(isinstance(x, float) for x in embedding)
-        print(f"✓ Single embedding generated: {len(embedding)} dimensions")
-
-    def test_batch_embedding(self):
-        """Test batch text embedding"""
-        client = get_embedding_client()
-        texts = [
-            "First message about Python programming",
-            "Second message about machine learning",
-            "Third message about data science"
-        ]
-
-        embeddings = client.embed_batch(texts)
-
-        assert len(embeddings) == 3
-        assert all(len(emb) == 384 for emb in embeddings)
-        print(f"✓ Batch embeddings generated: {len(embeddings)} texts")
-
-
-class TestQdrantMemory:
-    """Test Qdrant memory storage and retrieval"""
-
-    @pytest.fixture
-    def qdrant_memory(self):
-        """Get Qdrant memory instance"""
-        return get_qdrant_memory()
-
-    @pytest.fixture
-    def test_conversation_id(self):
-        """Generate unique test conversation ID"""
-        return f"test_conv_{int(datetime.utcnow().timestamp())}"
-
-    @pytest.mark.asyncio
-    async def test_qdrant_connection(self, qdrant_memory):
-        """Test Qdrant connection and collection"""
-        assert qdrant_memory.client is not None
-        assert qdrant_memory.collection_name == "core_api_conversations"
-        print(f"✓ Connected to Qdrant: {qdrant_memory.host}:{qdrant_memory.port}")
-
-    @pytest.mark.asyncio
-    async def test_add_turn(self, qdrant_memory, test_conversation_id):
-        """Test adding a turn to Qdrant"""
-        turn = ConversationTurn(
-            role=MessageRole.USER,
-            content="What is Python?",
-            turn_number=1,
-            tokens=TokenUsage(prompt=10, completion=0, total=10)
-        )
-
-        await qdrant_memory.add_turn(test_conversation_id, turn)
-
-        # Verify it was stored
-        exists = await qdrant_memory.conversation_exists(test_conversation_id)
-        assert exists is True
-        print(f"✓ Turn stored in Qdrant: {test_conversation_id}")
-
-    @pytest.mark.asyncio
-    async def test_chronological_retrieval(self, qdrant_memory, test_conversation_id):
-        """Test Tier 2 mode: chronological retrieval"""
-        # Add multiple turns
-        turns = [
-            ConversationTurn(role=MessageRole.USER, content="What is Python?", turn_number=1),
-            ConversationTurn(role=MessageRole.ASSISTANT, content="Python is a programming language", turn_number=2),
-            ConversationTurn(role=MessageRole.USER, content="How do I learn it?", turn_number=3),
-        ]
-
-        for turn in turns:
-            await qdrant_memory.add_turn(test_conversation_id, turn)
-
-        # Retrieve turns chronologically
-        retrieved = await qdrant_memory.get_turns(test_conversation_id)
-
-        assert len(retrieved) == 3
-        assert retrieved[0].turn_number == 1
-        assert retrieved[1].turn_number == 2
-        assert retrieved[2].turn_number == 3
-        assert retrieved[0].content == "What is Python?"
-        print(f"✓ Chronological retrieval works: {len(retrieved)} turns")
-
-    @pytest.mark.asyncio
-    async def test_semantic_search(self, qdrant_memory, test_conversation_id):
-        """Test Tier 3 mode: semantic search"""
-        # Add turns with distinct topics
-        turns = [
-            ConversationTurn(role=MessageRole.USER, content="I love machine learning and neural networks", turn_number=10),
-            ConversationTurn(role=MessageRole.USER, content="Pizza is my favorite food", turn_number=11),
-            ConversationTurn(role=MessageRole.USER, content="Deep learning models are fascinating", turn_number=12),
-        ]
-
-        for turn in turns:
-            await qdrant_memory.add_turn(test_conversation_id, turn)
-
-        # Search for AI-related content
-        results = await qdrant_memory.similarity_search(
-            query="artificial intelligence and AI",
-            conversation_id=test_conversation_id,
-            limit=3
-        )
-
-        assert len(results) > 0
-        # Top results should be about ML/AI, not pizza
-        top_result = results[0]
-        assert "machine learning" in top_result["content"] or "Deep learning" in top_result["content"]
-        assert top_result["score"] > 0.5  # Reasonable similarity score
-        print(f"✓ Semantic search works: {len(results)} matches, top score: {results[0]['score']:.3f}")
-
-    @pytest.mark.asyncio
-    async def test_conversation_stats(self, qdrant_memory, test_conversation_id):
-        """Test conversation statistics"""
-        stats = await qdrant_memory.get_conversation_stats(test_conversation_id)
-
-        assert stats["conversation_id"] == test_conversation_id
-        assert stats["total_turns"] >= 0
-        assert "total_tokens" in stats
-        print(f"✓ Stats retrieved: {stats['total_turns']} turns, {stats['total_tokens']} tokens")
-
-    @pytest.mark.asyncio
-    async def test_clear_conversation(self, qdrant_memory, test_conversation_id):
-        """Test clearing a conversation"""
-        # Add a turn
-        turn = ConversationTurn(role=MessageRole.USER, content="Test message", turn_number=99)
-        await qdrant_memory.add_turn(test_conversation_id, turn)
-
-        # Clear it
-        await qdrant_memory.clear_conversation(test_conversation_id)
-
-        # Verify it's gone
-        exists = await qdrant_memory.conversation_exists(test_conversation_id)
-        assert exists is False
-        print(f"✓ Conversation cleared: {test_conversation_id}")
-
-
-class TestIntegration:
-    """Test full integration: Tier 1 + Qdrant + Embeddings"""
-
-    @pytest.mark.asyncio
-    async def test_full_memory_flow(self):
-        """Test complete memory flow: Buffer → Qdrant"""
-        conversation_id = f"integration_test_{int(datetime.utcnow().timestamp())}"
-
-        # Initialize both tiers
-        buffer_memory = get_buffer_memory()
-        qdrant_memory = get_qdrant_memory()
-
-        # 1. Add turns to buffer (Tier 1)
-        turns = [
-            ConversationTurn(role=MessageRole.USER, content="Hello!", turn_number=1),
-            ConversationTurn(role=MessageRole.ASSISTANT, content="Hi there!", turn_number=2),
-            ConversationTurn(role=MessageRole.USER, content="How are you?", turn_number=3),
-        ]
-
-        for turn in turns:
-            await buffer_memory.add_turn(conversation_id, turn)
-
-        # Verify buffer has them
-        buffer = await buffer_memory.get_buffer(conversation_id)
-        assert len(buffer.turns) == 3
-        print(f"✓ Tier 1 buffer: {len(buffer.turns)} turns")
-
-        # 2. Move to Qdrant (Tier 2/3)
-        for turn in buffer.turns:
-            await qdrant_memory.add_turn(conversation_id, turn)
-
-        # Verify Qdrant has them
-        qdrant_turns = await qdrant_memory.get_turns(conversation_id)
-        assert len(qdrant_turns) == 3
-        print(f"✓ Tier 2/3 Qdrant: {len(qdrant_turns)} turns")
-
-        # 3. Test semantic search across both
-        search_results = await qdrant_memory.similarity_search(
-            query="greeting",
-            conversation_id=conversation_id,
-            limit=2
-        )
-        assert len(search_results) > 0
-        print(f"✓ Semantic search: {len(search_results)} matches")
-
-        # Cleanup
-        await qdrant_memory.clear_conversation(conversation_id)
-        await buffer_memory.clear_conversation(conversation_id)
-        print(f"✓ Full memory flow complete!")
-
-
-def run_tests():
-    """Run all tests"""
-    print("\n" + "="*60)
-    print("Phase 2 Memory System Integration Tests")
-    print("="*60 + "\n")
-
-    # Test 1: Embedding Client
-    print("Test 1: Embedding Client")
-    print("-" * 40)
-    test_embed = TestEmbeddingClient()
-    test_embed.test_embedding_client_init()
-    test_embed.test_single_embedding()
-    test_embed.test_batch_embedding()
-    print()
-
-    # Test 2: Qdrant Memory
-    print("Test 2: Qdrant Memory Storage")
-    print("-" * 40)
-    test_qdrant = TestQdrantMemory()
-    qdrant_memory = get_qdrant_memory()
-    test_conv_id = f"test_conv_{int(datetime.utcnow().timestamp())}"
-
-    asyncio.run(test_qdrant.test_qdrant_connection(qdrant_memory))
-    asyncio.run(test_qdrant.test_add_turn(qdrant_memory, test_conv_id))
-    asyncio.run(test_qdrant.test_chronological_retrieval(qdrant_memory, test_conv_id))
-    asyncio.run(test_qdrant.test_semantic_search(qdrant_memory, test_conv_id))
-    asyncio.run(test_qdrant.test_conversation_stats(qdrant_memory, test_conv_id))
-    asyncio.run(test_qdrant.test_clear_conversation(qdrant_memory, test_conv_id))
-    print()
-
-    # Test 3: Full Integration
-    print("Test 3: Full Integration (Tier 1 + Tier 2/3)")
-    print("-" * 40)
-    test_integration = TestIntegration()
-    asyncio.run(test_integration.test_full_memory_flow())
-    print()
-
-    print("="*60)
-    print("✅ All Memory System Tests Passed!")
-    print("="*60)
-
-
-if __name__ == "__main__":
-    run_tests()

tests/test_memory_manager.py

@@ -1,150 +0,0 @@
-#!/usr/bin/env python3
-"""
-Test MemoryManager orchestration
-
-Verifies unified memory interface works correctly.
-"""
-import asyncio
-import sys
-from datetime import datetime
-
-sys.path.insert(0, '/app')
-
-from src.memory import MemoryManager, get_memory_manager, MessageRole, TokenUsage
-
-
-async def test_memory_manager():
-    """Test MemoryManager orchestration"""
-    print("\n" + "="*60)
-    print("MEMORY MANAGER TEST")
-    print("="*60)
-
-    test_conv_id = f"manager_test_{int(datetime.utcnow().timestamp())}"
-
-    try:
-        # Initialize manager
-        manager = get_memory_manager()
-        print(f"✓ MemoryManager initialized")
-
-        # Test 1: Add turns through manager
-        print("\n1. Adding turns via MemoryManager...")
-        turn1 = await manager.add_turn(
-            conversation_id=test_conv_id,
-            role=MessageRole.USER,
-            content="Hello, how are you?",
-            tokens=TokenUsage(prompt=5, completion=0, total=5)
-        )
-        assert turn1.turn_number == 1
-        print(f"  ✓ Turn 1 added: {turn1.content[:30]}...")
-
-        turn2 = await manager.add_turn(
-            conversation_id=test_conv_id,
-            role=MessageRole.ASSISTANT,
-            content="I'm doing great! How can I help you today?",
-            tokens=TokenUsage(prompt=5, completion=10, total=15)
-        )
-        assert turn2.turn_number == 2
-        print(f"  ✓ Turn 2 added: {turn2.content[:30]}...")
-
-        # Test 2: Get recent turns (from buffer)
-        print("\n2. Getting recent turns from buffer...")
-        recent = await manager.get_recent_turns(test_conv_id, limit=10)
-        assert len(recent) == 2
-        assert recent[0].turn_number == 1
-        assert recent[1].turn_number == 2
-        print(f"  ✓ Retrieved {len(recent)} recent turns from buffer")
-
-        # Test 3: Add more turns to trigger consolidation (threshold = 10)
-        print("\n3. Adding turns to trigger auto-consolidation...")
-        for i in range(3, 11):  # Add turns 3-10
-            await manager.add_turn(
-                conversation_id=test_conv_id,
-                role=MessageRole.USER if i % 2 == 1 else MessageRole.ASSISTANT,
-                content=f"Test message number {i}",
-                tokens=TokenUsage(prompt=5, completion=5, total=10)
-            )
-        print(f"  ✓ Added 8 more turns (total: 10)")
-
-        # Check if consolidation happened (turn 10 should trigger it)
-        print("\n4. Verifying auto-consolidation...")
-        stats = await manager.get_conversation_stats(test_conv_id)
-        print(f"  Buffer turns: {stats['buffer_turns']}")
-        print(f"  Qdrant turns: {stats['qdrant_turns']}")
-        print(f"  Exists in buffer: {stats['exists_in_buffer']}")
-        print(f"  Exists in Qdrant: {stats['exists_in_qdrant']}")
-
-        if stats['qdrant_turns'] > 0:
-            print(f"  ✓ Auto-consolidation triggered! {stats['qdrant_turns']} turns in Qdrant")
-        else:
-            print(f"  ⚠ No auto-consolidation yet (threshold may not be reached)")
-
-        # Test 4: Manual consolidation
-        print("\n5. Testing manual consolidation...")
-        consolidated = await manager.consolidate(test_conv_id)
-        print(f"  ✓ Manually consolidated {consolidated} turns")
-
-        # Test 5: Get full history (buffer + Qdrant)
-        print("\n6. Getting full conversation history...")
-        full_history = await manager.get_full_history(test_conv_id)
-        print(f"  ✓ Retrieved {len(full_history)} total turns")
-        assert len(full_history) == 10, f"Expected 10 turns, got {len(full_history)}"
-        print(f"  ✓ Full history verified (10 turns)")
-
-        # Test 6: Semantic search
-        print("\n7. Testing semantic search...")
-        search_results = await manager.search_conversations(
-            query="greeting hello",
-            conversation_id=test_conv_id,
-            limit=3
-        )
-        if len(search_results) > 0:
-            print(f"  ✓ Semantic search found {len(search_results)} matches")
-            print(f"    Top: '{search_results[0]['content'][:40]}...' (score: {search_results[0]['score']:.3f})")
-        else:
-            print(f"  ⚠ No semantic search results (may need more data)")
-
-        # Test 7: Clear conversation
-        print("\n8. Clearing conversation...")
-        await manager.clear_conversation(test_conv_id)
-        stats_after = await manager.get_conversation_stats(test_conv_id)
-        assert stats_after['buffer_turns'] == 0
-        assert stats_after['qdrant_turns'] == 0
-        print(f"  ✓ Conversation cleared from all tiers")
-
-        print("\n" + "="*60)
-        print("✅ MEMORY MANAGER TEST: PASSED")
-        print("="*60)
-        print("\nMemoryManager verified:")
-        print("  ✓ Add turns with auto turn numbering")
-        print("  ✓ Get recent turns from buffer")
-        print("  ✓ Auto-consolidation (when threshold reached)")
-        print("  ✓ Manual consolidation")
-        print("  ✓ Get full history (buffer + Qdrant)")
-        print("  ✓ Semantic search")
-        print("  ✓ Clear conversation")
-        print("  ✓ Conversation stats")
-        return True
-
-    except Exception as e:
-        print(f"\n❌ MEMORY MANAGER TEST: FAILED")
-        print(f"Error: {e}")
-        import traceback
-        traceback.print_exc()
-
-        # Cleanup on error
-        try:
-            await manager.clear_conversation(test_conv_id)
-        except:
-            pass
-
-        return False
-
-
-def main():
-    """Run the test"""
-    success = asyncio.run(test_memory_manager())
-    return 0 if success else 1
-
-
-if __name__ == "__main__":
-    exit(main())

tests/test_memory_simple.py

@@ -1,330 +0,0 @@
-#!/usr/bin/env python3
-"""
-Simple integration tests for Phase 2 Memory System
-No external dependencies beyond the memory system itself
-"""
-import asyncio
-import sys
-from datetime import datetime
-
-# Add src to path
-sys.path.insert(0, '/app')
-
-from src.memory import (
-    ConversationBufferMemory,
-    QdrantConversationMemory,
-    ConversationTurn,
-    MessageRole,
-    TokenUsage,
-    get_buffer_memory,
-    get_qdrant_memory
-)
-from src.models.embeddings import get_embedding_client
-
-
-def test_embedding_client():
-    """Test 1: Embedding Client"""
-    print("\n" + "="*60)
-    print("Test 1: Embedding Client")
-    print("="*60)
-
-    try:
-        # Initialize
-        client = get_embedding_client()
-        assert client is not None
-        assert client.dimension == 384
-        print(f"✓ Embedding client initialized: {client.model_name}")
-        print(f"✓ Embedding dimension: {client.dimension}")
-
-        # Single embedding
-        text = "Hello, this is a test message for embedding generation"
-        embedding = client.embed_text(text)
-        assert isinstance(embedding, list)
-        assert len(embedding) == 384
-        assert all(isinstance(x, float) for x in embedding)
-        print(f"✓ Single embedding generated: {len(embedding)} dimensions")
-        print(f"  Sample values: [{embedding[0]:.4f}, {embedding[1]:.4f}, {embedding[2]:.4f}, ...]")
-
-        # Batch embedding
-        texts = [
-            "First message about Python programming",
-            "Second message about machine learning",
-            "Third message about data science"
-        ]
-        embeddings = client.embed_batch(texts)
-        assert len(embeddings) == 3
-        assert all(len(emb) == 384 for emb in embeddings)
-        print(f"✓ Batch embeddings generated: {len(embeddings)} texts")
-
-        print("\n✅ Embedding Client Tests: PASSED")
-        return True
-
-    except Exception as e:
-        print(f"\n❌ Embedding Client Tests: FAILED")
-        print(f"Error: {e}")
-        import traceback
-        traceback.print_exc()
-        return False
-
-
-async def test_qdrant_memory():
-    """Test 2: Qdrant Memory Storage"""
-    print("\n" + "="*60)
-    print("Test 2: Qdrant Memory Storage")
-    print("="*60)
-
-    test_conv_id = f"test_conv_{int(datetime.utcnow().timestamp())}"
-
-    try:
-        # Initialize
-        qdrant_memory = get_qdrant_memory()
-        assert qdrant_memory.client is not None
-        assert qdrant_memory.collection_name == "core_api_conversations"
-        print(f"✓ Connected to Qdrant: {qdrant_memory.host}:{qdrant_memory.port}")
-        print(f"✓ Collection: {qdrant_memory.collection_name}")
-
-        # Add single turn
-        turn1 = ConversationTurn(
-            role=MessageRole.USER,
-            content="What is Python?",
-            turn_number=1,
-            tokens=TokenUsage(prompt=10, completion=0, total=10)
-        )
-        await qdrant_memory.add_turn(test_conv_id, turn1)
-        print(f"✓ Turn 1 stored in Qdrant")
-
-        # Verify it exists
-        exists = await qdrant_memory.conversation_exists(test_conv_id)
-        assert exists is True
-        print(f"✓ Conversation exists: {test_conv_id}")
-
-        # Add more turns for chronological test
-        turn2 = ConversationTurn(
-            role=MessageRole.ASSISTANT,
-            content="Python is a high-level programming language known for simplicity and readability",
-            turn_number=2
-        )
-        turn3 = ConversationTurn(
-            role=MessageRole.USER,
-            content="How do I learn Python programming?",
-            turn_number=3
-        )
-
-        await qdrant_memory.add_turn(test_conv_id, turn2)
-        await qdrant_memory.add_turn(test_conv_id, turn3)
-        print(f"✓ Turns 2-3 stored in Qdrant")
-
-        # Test chronological retrieval (Tier 2 mode)
-        retrieved = await qdrant_memory.get_turns(test_conv_id)
-        assert len(retrieved) == 3
-        assert retrieved[0].turn_number == 1
-        assert retrieved[1].turn_number == 2
-        assert retrieved[2].turn_number == 3
-        assert retrieved[0].content == "What is Python?"
-        print(f"✓ Chronological retrieval works: {len(retrieved)} turns")
-        for i, turn in enumerate(retrieved, 1):
-            print(f"  Turn {turn.turn_number}: {turn.role.value} - {turn.content[:50]}...")
-
-        # Add turns with distinct topics for semantic search
-        turn10 = ConversationTurn(
-            role=MessageRole.USER,
-            content="I love machine learning and neural networks and artificial intelligence",
-            turn_number=10
-        )
-        turn11 = ConversationTurn(
-            role=MessageRole.USER,
-            content="Pizza is my favorite food and I enjoy eating pasta",
-            turn_number=11
-        )
-        turn12 = ConversationTurn(
-            role=MessageRole.USER,
-            content="Deep learning models and transformers are fascinating AI technologies",
-            turn_number=12
-        )
-
-        await qdrant_memory.add_turn(test_conv_id, turn10)
-        await qdrant_memory.add_turn(test_conv_id, turn11)
-        await qdrant_memory.add_turn(test_conv_id, turn12)
-        print(f"✓ Added 3 more turns for semantic search test")
-
-        # Test semantic search (Tier 3 mode)
-        search_results = await qdrant_memory.similarity_search(
-            query="artificial intelligence and deep learning",
-            conversation_id=test_conv_id,
-            limit=3
-        )
-        assert len(search_results) > 0
-        print(f"✓ Semantic search works: {len(search_results)} matches")
-
-        # Top result should be about AI/ML, not food
-        top_result = search_results[0]
-        print(f"  Top match (score: {top_result['score']:.3f}): {top_result['content'][:60]}...")
-        assert top_result["score"] > 0.5, "Semantic similarity score too low"
-
-        # Verify top matches are AI-related
-        ai_keywords = ["machine learning", "neural networks", "Deep learning", "AI", "artificial intelligence"]
-        top_content = search_results[0]["content"]
-        assert any(keyword in top_content for keyword in ai_keywords), "Top result not AI-related"
-        print(f"✓ Semantic relevance verified (AI-related content ranked higher)")
-
-        # Test conversation stats
-        stats = await qdrant_memory.get_conversation_stats(test_conv_id)
-        assert stats["conversation_id"] == test_conv_id
-        assert stats["total_turns"] == 6
-        print(f"✓ Stats retrieved: {stats['total_turns']} turns, {stats['total_tokens']} tokens")
-
-        # Cleanup
-        await qdrant_memory.clear_conversation(test_conv_id)
-        exists_after = await qdrant_memory.conversation_exists(test_conv_id)
-        assert exists_after is False
-        print(f"✓ Conversation cleared successfully")
-
-        print("\n✅ Qdrant Memory Tests: PASSED")
-        return True
-
-    except Exception as e:
-        print(f"\n❌ Qdrant Memory Tests: FAILED")
-        print(f"Error: {e}")
-        import traceback
-        traceback.print_exc()
-
-        # Cleanup on error
-        try:
-            await qdrant_memory.clear_conversation(test_conv_id)
-        except:
-            pass
-
-        return False
-
-
-async def test_full_integration():
-    """Test 3: Full Integration (Tier 1 + Tier 2/3)"""
-    print("\n" + "="*60)
-    print("Test 3: Full Integration (Tier 1 + Tier 2/3)")
-    print("="*60)
-
-    test_conv_id = f"integration_test_{int(datetime.utcnow().timestamp())}"
-
-    try:
-        # Initialize both tiers
-        buffer_memory = get_buffer_memory()
-        qdrant_memory = get_qdrant_memory()
-        print(f"✓ Initialized Tier 1 (Buffer) and Tier 2/3 (Qdrant)")
-
-        # 1. Add turns to buffer (Tier 1)
-        turns = [
-            ConversationTurn(role=MessageRole.USER, content="Hello!", turn_number=1),
-            ConversationTurn(role=MessageRole.ASSISTANT, content="Hi there! How can I help?", turn_number=2),
-            ConversationTurn(role=MessageRole.USER, content="How are you?", turn_number=3),
-            ConversationTurn(role=MessageRole.ASSISTANT, content="I'm doing great, thanks!", turn_number=4),
-        ]
-
-        for turn in turns:
-            await buffer_memory.add_turn(test_conv_id, turn)
-
-        # Verify buffer has them
-        buffer = await buffer_memory.get_buffer(test_conv_id)
-        assert len(buffer.turns) == 4
-        print(f"✓ Tier 1 buffer: {len(buffer.turns)} turns stored")
-
-        # 2. Move to Qdrant (Tier 2/3) - simulating consolidation
-        for turn in buffer.turns:
-            await qdrant_memory.add_turn(test_conv_id, turn)
-
-        # Verify Qdrant has them
-        qdrant_turns = await qdrant_memory.get_turns(test_conv_id)
-        assert len(qdrant_turns) == 4
-        print(f"✓ Tier 2/3 Qdrant: {len(qdrant_turns)} turns stored")
-
-        # 3. Test semantic search across consolidated data
-        search_results = await qdrant_memory.similarity_search(
-            query="greeting hello",
-            conversation_id=test_conv_id,
-            limit=2
-        )
-        assert len(search_results) > 0
-        print(f"✓ Semantic search: {len(search_results)} matches found")
-        print(f"  Best match: '{search_results[0]['content']}' (score: {search_results[0]['score']:.3f})")
-
-        # 4. Test data consistency
-        buffer_content = [t.content for t in buffer.turns]
-        qdrant_content = [t.content for t in qdrant_turns]
-        assert buffer_content == qdrant_content
-        print(f"✓ Data consistency verified (Buffer ↔ Qdrant)")
-
-        # Cleanup
-        await qdrant_memory.clear_conversation(test_conv_id)
-        await buffer_memory.clear_conversation(test_conv_id)
-        print(f"✓ Cleanup complete")
-
-        print("\n✅ Full Integration Tests: PASSED")
-        return True
-
-    except Exception as e:
-        print(f"\n❌ Full Integration Tests: FAILED")
-        print(f"Error: {e}")
-        import traceback
-        traceback.print_exc()
-
-        # Cleanup on error
-        try:
-            await qdrant_memory.clear_conversation(test_conv_id)
-            await buffer_memory.clear_conversation(test_conv_id)
-        except:
-            pass
-
-        return False
-
-
-def main():
-    """Run all tests"""
-    print("\n" + "="*60)
-    print("PHASE 2 MEMORY SYSTEM - INTEGRATION TESTS")
-    print("="*60)
-    print(f"Start time: {datetime.utcnow().isoformat()}")
-
-    results = []
-
-    # Test 1: Embedding Client
-    results.append(("Embedding Client", test_embedding_client()))
-
-    # Test 2: Qdrant Memory
-    results.append(("Qdrant Memory", asyncio.run(test_qdrant_memory())))
-
-    # Test 3: Full Integration
-    results.append(("Full Integration", asyncio.run(test_full_integration())))
-
-    # Summary
-    print("\n" + "="*60)
-    print("TEST SUMMARY")
-    print("="*60)
-
-    for test_name, passed in results:
-        status = "✅ PASSED" if passed else "❌ FAILED"
-        print(f"{test_name:.<40} {status}")
-
-    total = len(results)
-    passed = sum(1 for _, p in results if p)
-    failed = total - passed
-
-    print(f"\nTotal: {total} | Passed: {passed} | Failed: {failed}")
-    print(f"Success rate: {(passed/total)*100:.1f}%")
-
-    if all(p for _, p in results):
-        print("\n" + "="*60)
-        print("🎉 ALL TESTS PASSED!")
-        print("="*60)
-        print("\nPhase 2 Memory System Status: ✅ FUNCTIONAL")
-        print("- Embedding client working (384d vectors)")
-        print("- Qdrant storage working (chronological + semantic)")
-        print("- Full integration working (Tier 1 ↔ Tier 2/3)")
-        return 0
-    else:
-        print("\n" + "="*60)
-        print("❌ SOME TESTS FAILED")
-        print("="*60)
-        return 1
-
-
-if __name__ == "__main__":
-    exit(main())

tests/test_service_groups.py

@@ -0,0 +1,146 @@
+"""Tests for service_groups module."""
+import pytest
+from src.service_groups import (
+    ALWAYS_ON_SERVICES,
+    SERVICE_GROUPS,
+    is_always_on,
+    get_service_group,
+    get_group_name,
+    list_service_groups,
+    list_stoppable_services,
+    validate_stop_request,
+)
+
+
+class TestAlwaysOnServices:
+    """Test always-on service configuration."""
+
+    def test_always_on_services_includes_infrastructure(self):
+        """Critical infrastructure services should be in always-on list."""
+        assert "portainer" in ALWAYS_ON_SERVICES
+        assert "nginx-proxy-manager" in ALWAYS_ON_SERVICES
+        assert "core-api" in ALWAYS_ON_SERVICES
+
+    def test_uptime_kuma_not_in_always_on(self):
+        """Uptime Kuma was removed from always-on list."""
+        assert "uptime-kuma" not in ALWAYS_ON_SERVICES
+
+
+class TestIsAlwaysOn:
+    """Test is_always_on function."""
+
+    def test_is_always_on_returns_true_for_infrastructure(self):
+        """Infrastructure services should return True."""
+        assert is_always_on("portainer") is True
+        assert is_always_on("nginx-proxy-manager") is True
+        assert is_always_on("core-api") is True
+
+    def test_is_always_on_case_insensitive(self):
+        """Function should be case-insensitive."""
+        assert is_always_on("PORTAINER") is True
+        assert is_always_on("Portainer") is True
+        assert is_always_on("PoRtAiNeR") is True
+
+    def test_is_always_on_returns_false_for_stoppable(self):
+        """Stoppable services should return False."""
+        assert is_always_on("jellyfin") is False
+        assert is_always_on("nextcloud") is False
+        assert is_always_on("unknown-service") is False
+
+
+class TestGetServiceGroup:
+    """Test get_service_group function."""
+
+    def test_returns_group_members_for_grouped_service(self):
+        """Should return all services in the group."""
+        # Assuming jellyfin is defined in SERVICE_GROUPS
+        if "jellyfin" in SERVICE_GROUPS:
+            result = get_service_group("jellyfin")
+            assert "jellyfin" in result
+
+    def test_returns_single_item_for_ungrouped_service(self):
+        """Ungrouped services should return themselves."""
+        result = get_service_group("some-random-service")
+        assert result == ["some-random-service"]
+
+    def test_returns_copy_not_reference(self):
+        """Should return a copy to prevent modification."""
+        if SERVICE_GROUPS:
+            group_name = list(SERVICE_GROUPS.keys())[0]
+            result1 = get_service_group(group_name)
+            result2 = get_service_group(group_name)
+            assert result1 is not result2
+
+
+class TestGetGroupName:
+    """Test get_group_name function."""
+
+    def test_returns_group_name_for_grouped_service(self):
+        """Should return group name for services in groups."""
+        # If ai-stack group exists with ollama
+        if "ai-stack" in SERVICE_GROUPS and "ollama" in SERVICE_GROUPS["ai-stack"]:
+            assert get_group_name("ollama") == "ai-stack"
+
+    def test_returns_service_name_for_ungrouped(self):
+        """Ungrouped services should return their own name."""
+        assert get_group_name("random-service") == "random-service"
+
+
+class TestListServiceGroups:
+    """Test list_service_groups function."""
+
+    def test_returns_all_groups(self):
+        """Should return all defined service groups."""
+        result = list_service_groups()
+        assert isinstance(result, dict)
+        assert result == SERVICE_GROUPS
+
+    def test_returns_copy(self):
+        """Should return a copy to prevent modification."""
+        result = list_service_groups()
+        assert result is not SERVICE_GROUPS
+
+
+class TestListStoppableServices:
+    """Test list_stoppable_services function."""
+
+    def test_returns_list(self):
+        """Should return a list."""
+        result = list_stoppable_services()
+        assert isinstance(result, list)
+
+    def test_excludes_always_on_services(self):
+        """Should not include always-on services."""
+        result = list_stoppable_services()
+        for service in result:
+            assert not is_always_on(service), f"{service} is always-on but in stoppable list"
+
+
+class TestValidateStopRequest:
+    """Test validate_stop_request function."""
+
+    def test_valid_for_stoppable_services(self):
+        """Should return valid for stoppable services."""
+        stoppable = list_stoppable_services()
+        if stoppable:
+            is_valid, error = validate_stop_request([stoppable[0]])
+            assert is_valid is True
+            assert error == ""
+
+    def test_invalid_for_always_on_services(self):
+        """Should return invalid for always-on services."""
+        is_valid, error = validate_stop_request(["portainer"])
+        assert is_valid is False
+        assert "always-on" in error.lower()
+        assert "portainer" in error
+
+    def test_invalid_if_any_service_is_always_on(self):
+        """Should fail if any service in list is always-on."""
+        is_valid, error = validate_stop_request(["jellyfin", "portainer"])
+        assert is_valid is False
+
+    def test_valid_for_empty_list(self):
+        """Empty list should be valid."""
+        is_valid, error = validate_stop_request([])
+        assert is_valid is True
+        assert error == ""