#!/usr/bin/env python3 """ Tests for STORY-001: QwenConfig and Exceptions =============================================== Black Box Tests: - Config values match expected - Environment variable overrides work - Exception hierarchy correct White Box Tests: - URL property construction - Config serialization - Connection test function """ import os import sys import unittest from unittest.mock import patch, MagicMock # Add project root to path sys.path.insert(0, "/mnt/e/genesis-system") from core.qwen.config import QwenConfig, get_default_config, test_connection from core.qwen.exceptions import ( QwenError, QwenConnectionError, QwenTimeoutError, QwenModelNotFoundError, QwenRateLimitError, QwenCircuitOpenError, QwenResponseError, QwenWarmupError, ) class TestQwenConfigBlackBox(unittest.TestCase): """Black box tests - test behavior without internal knowledge.""" def test_default_endpoint_is_correct(self): """CRITICAL: Verify default endpoint is AIVA Ollama, not localhost.""" config = QwenConfig() # Must be AIVA's server, NOT localhost self.assertEqual(config.host, "152.53.201.152") self.assertEqual(config.port, 23405) # NOT 11434! self.assertEqual(config.base_url, "http://152.53.201.152:23405") def test_default_model_is_correct(self): """Verify default model matches AIVA's Ollama.""" config = QwenConfig() self.assertEqual( config.model, "huihui_ai/qwenlong-l1.5-abliterated:30b-a3b" ) def test_environment_variable_override_host(self): """Test that QWEN_HOST env var overrides default.""" with patch.dict(os.environ, {"QWEN_HOST": "192.168.1.100"}): config = QwenConfig() self.assertEqual(config.host, "192.168.1.100") def test_environment_variable_override_port(self): """Test that QWEN_PORT env var overrides default.""" with patch.dict(os.environ, {"QWEN_PORT": "8080"}): config = QwenConfig() self.assertEqual(config.port, 8080) def test_environment_variable_override_model(self): """Test that QWEN_MODEL env var overrides default.""" with patch.dict(os.environ, {"QWEN_MODEL": "llama3:70b"}): config = QwenConfig() self.assertEqual(config.model, "llama3:70b") def test_get_default_config_returns_qwen_config(self): """Test helper function returns correct type.""" config = get_default_config() self.assertIsInstance(config, QwenConfig) def test_timeout_defaults_are_reasonable(self): """Test timeout values are sensible for large model.""" config = QwenConfig() # Connect timeout should be short self.assertLessEqual(config.connect_timeout, 30.0) # Read timeout should be long (for generation) self.assertGreaterEqual(config.read_timeout, 60.0) def test_rate_limits_are_conservative(self): """Test rate limits are set for single Ollama instance.""" config = QwenConfig() # RPM should be conservative for single instance self.assertLessEqual(config.rpm, 20) # AIVA reservation should be meaningful self.assertGreaterEqual(config.aiva_reservation_percent, 10) class TestQwenConfigWhiteBox(unittest.TestCase): """White box tests - test internal implementation.""" def test_url_properties_constructed_correctly(self): """Test all URL properties are built from host/port.""" config = QwenConfig() self.assertEqual(config.generate_url, "http://152.53.201.152:23405/api/generate") self.assertEqual(config.chat_url, "http://152.53.201.152:23405/api/chat") self.assertEqual(config.tags_url, "http://152.53.201.152:23405/api/tags") self.assertEqual(config.ps_url, "http://152.53.201.152:23405/api/ps") self.assertEqual(config.openai_base_url, "http://152.53.201.152:23405/v1") def test_get_connection_params_structure(self): """Test connection params dict has required keys.""" config = QwenConfig() params = config.get_connection_params() self.assertIn("base_url", params) self.assertIn("timeout", params) self.assertIn("model", params) self.assertIsInstance(params["timeout"], tuple) def test_get_generation_options_structure(self): """Test generation options dict has required keys.""" config = QwenConfig() options = config.get_generation_options() self.assertIn("num_ctx", options) self.assertIn("temperature", options) self.assertIn("top_p", options) self.assertIn("top_k", options) self.assertIn("repeat_penalty", options) def test_to_dict_serialization(self): """Test config can be serialized to dict.""" config = QwenConfig() data = config.to_dict() self.assertIn("endpoint", data) self.assertIn("timeouts", data) self.assertIn("rate_limits", data) self.assertEqual(data["endpoint"]["host"], "152.53.201.152") def test_circuit_breaker_config_values(self): """Test circuit breaker config has sensible defaults.""" config = QwenConfig() cb_config = config.get_circuit_breaker_config() self.assertIn("failure_threshold", cb_config) self.assertIn("recovery_timeout", cb_config) self.assertIn("half_open_max_calls", cb_config) # Sensible defaults self.assertGreaterEqual(cb_config["failure_threshold"], 2) self.assertGreaterEqual(cb_config["recovery_timeout"], 30.0) class TestQwenExceptionsBlackBox(unittest.TestCase): """Black box tests for exception classes.""" def test_all_exceptions_inherit_from_qwen_error(self): """Test exception hierarchy is correct.""" exceptions = [ QwenConnectionError, QwenTimeoutError, QwenModelNotFoundError, QwenRateLimitError, QwenCircuitOpenError, QwenResponseError, QwenWarmupError, ] for exc_class in exceptions: self.assertTrue( issubclass(exc_class, QwenError), f"{exc_class.__name__} should inherit from QwenError" ) def test_qwen_error_stores_message(self): """Test base exception stores message.""" error = QwenError("Test error message") self.assertEqual(error.message, "Test error message") self.assertIn("Test error message", str(error)) def test_qwen_error_stores_details(self): """Test base exception stores details dict.""" details = {"code": 500, "endpoint": "http://test"} error = QwenError("Test error", details) self.assertEqual(error.details, details) self.assertIn("500", str(error)) def test_rate_limit_error_has_retry_after(self): """Test rate limit exception has retry_after field.""" error = QwenRateLimitError("Rate limited", retry_after=30.0) self.assertEqual(error.retry_after, 30.0) def test_circuit_open_error_has_recovery_time(self): """Test circuit open exception has recovery_time field.""" error = QwenCircuitOpenError("Circuit open", recovery_time=60.0) self.assertEqual(error.recovery_time, 60.0) class TestQwenExceptionsWhiteBox(unittest.TestCase): """White box tests for exception classes.""" def test_exception_str_with_details(self): """Test string representation includes details.""" error = QwenError("Error", {"key": "value"}) str_repr = str(error) self.assertIn("Error", str_repr) self.assertIn("Details", str_repr) self.assertIn("key", str_repr) def test_exception_str_without_details(self): """Test string representation without details is clean.""" error = QwenError("Simple error") str_repr = str(error) self.assertEqual(str_repr, "Simple error") self.assertNotIn("Details", str_repr) def test_exception_can_be_raised_and_caught(self): """Test exceptions work in try/except.""" with self.assertRaises(QwenConnectionError): raise QwenConnectionError("Cannot connect") # Can catch as base class try: raise QwenTimeoutError("Timeout") except QwenError as e: self.assertIsInstance(e, QwenTimeoutError) class TestConnectionTest(unittest.TestCase): """Tests for test_connection function.""" def test_connection_test_returns_dict(self): """Test function returns proper dict structure.""" # This will likely fail (offline) but should return valid structure result = test_connection() self.assertIn("status", result) self.assertIn("endpoint", result) self.assertIn("model", result) self.assertIn("latency_ms", result) self.assertIn(result["status"], ["online", "offline", "error", "unknown"]) @patch("urllib.request.urlopen") def test_connection_test_online(self, mock_urlopen): """Test connection test when endpoint is online.""" mock_response = MagicMock() mock_response.read.return_value = b'{"models": [{"name": "huihui_ai/qwenlong-l1.5-abliterated:30b-a3b"}]}' mock_response.__enter__ = MagicMock(return_value=mock_response) mock_response.__exit__ = MagicMock(return_value=False) mock_urlopen.return_value = mock_response result = test_connection() self.assertEqual(result["status"], "online") self.assertIsNotNone(result["latency_ms"]) self.assertTrue(result["model_loaded"]) if __name__ == "__main__": # Run tests with verbosity unittest.main(verbosity=2)