From 33431e87fc8c41fc6f3fe5e2f080926bd2c1f91e Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Thu, 15 May 2025 00:47:12 +0100 Subject: [PATCH] issue #16 - add model tests, cursor config file and update the tracing.py to setup Jaeger & Prometheus exporters --- .cursor/rules/python.mdc | 118 ++++++++++ amqp/healthcheck.py | 12 - amqp/service/tracing.py | 267 +++++++++++++++-------- tests/adapter/test_trace_info_adapter.py | 254 ++++++++++++++++++++- tests/model/test_model.py | 220 +++++++++++++++++++ tests/requirements-test.txt | 4 + 6 files changed, 775 insertions(+), 100 deletions(-) create mode 100644 .cursor/rules/python.mdc delete mode 100644 amqp/healthcheck.py create mode 100644 tests/model/test_model.py create mode 100644 tests/requirements-test.txt diff --git a/.cursor/rules/python.mdc b/.cursor/rules/python.mdc new file mode 100644 index 0000000..d883676 --- /dev/null +++ b/.cursor/rules/python.mdc @@ -0,0 +1,118 @@ +--- +description: +globs: +alwaysApply: true +--- +--- +description: Python best practices and patterns for modern software development with Flask and SQLite +globs: **/*.py, src/**/*.py, tests/**/*.py +--- + +# Python Best Practices + +## Project Structure +- Use src-layout with `amqp/` +- Place tests in `tests/` directory parallel to `amqp/` +- Keep configuration in `amqp/config/` or as environment variables +- Store requirements in `pyproject.toml` +- Place static files in `static/` directory +- Use `templates/` for Jinja2 templates + +## Code Style +- Follow Black code formatting +- Use isort for import sorting +- Follow PEP 8 naming conventions: + - snake_case for functions and variables + - PascalCase for classes + - UPPER_CASE for constants +- Maximum line length of 100 characters +- Use absolute imports over relative imports + +## Type Hints +- Use type hints for all function parameters and returns +- Import types from `typing` module +- Use `Optional[Type]` instead of `Type | None` +- Use `TypeVar` for generic types +- Define custom types in `types.py` +- Use `Protocol` for duck typing + +## Database +- Use SQLAlchemy ORM +- Implement database migrations with Alembic +- Use proper connection pooling +- Define models in separate modules +- Implement proper relationships +- Use proper indexing strategies + +## Authentication +##- Use Flask-Login for session management +##- Implement Google OAuth using Flask-OAuth +##- Hash passwords with bcrypt +##- Use proper session security +##- Implement CSRF protection +##- Use proper role-based access control + +## API Design +##- Use Flask-RESTful for REST APIs +##- Implement proper request validation +##- Use proper HTTP status codes +##- Handle errors consistently +##- Use proper response formats +##- Implement proper rate limiting + +## Testing +- Use pytest for testing +- Write tests for all routes +- Use pytest-cov for coverage +- Implement proper fixtures +- Use proper mocking with pytest-mock +- Test all error scenarios + +## Security +##- Use HTTPS in production +##- Implement proper CORS +##- Sanitize all user inputs +##- Use proper session configuration +- Implement proper logging +##- Follow OWASP guidelines + +## Performance +##- Use proper caching with Flask-Caching +##- Implement database query optimization +##- Use proper connection pooling +##- Implement proper pagination +##- Use background tasks for heavy operations +- Monitor application performance + +## Error Handling +- Create custom exception classes +- Use proper try-except blocks +- Implement proper logging +- Return proper error responses +- Handle edge cases properly +- Use proper error messages + +## Documentation +- Use Google-style docstrings +- Document all public APIs +- Keep README.md updated +- Use proper inline comments +- Generate API documentation +- Document environment setup + +## Development Workflow +- Use virtual environments (venv) +- Use or Implement pre-commit hooks +- Use proper Git workflow +- Follow semantic versioning +- Use proper CI/CD practices +- Implement proper logging + +## Dependencies +##- Pin dependency versions +##- Use requirements.txt for production +- Separate dev dependencies +##- Use proper package versions +##- Regularly update dependencies +- Check for security vulnerabilitie +]cvx -= diff --git a/amqp/healthcheck.py b/amqp/healthcheck.py deleted file mode 100644 index b1038a0..0000000 --- a/amqp/healthcheck.py +++ /dev/null @@ -1,12 +0,0 @@ -import sys - -import requests - -url = "http://localhost:8080/amq-adapter-healthcheck" -headers = {"Authorization": "Bearer my_access_token"} -params = {"limit": 10, "offset": 20} -response = requests.get(url, headers=headers, params=params) -if response.status_code == 200: - sys.exit(0) -# Failed execution -sys.exit(1) diff --git a/amqp/service/tracing.py b/amqp/service/tracing.py index bb0fac6..630c306 100644 --- a/amqp/service/tracing.py +++ b/amqp/service/tracing.py @@ -1,99 +1,196 @@ +""" +OpenTelemetry configuration for tracing and metrics. + +This module provides functionality to initialize and configure OpenTelemetry +for both tracing (Jaeger via OTLP) and metrics (Prometheus) in the AMQ adapter. +""" + +import logging import os +from dataclasses import dataclass +from typing import Any, Dict, Optional from opentelemetry import metrics, trace from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.exporter.prometheus import PrometheusMetricReader from opentelemetry.sdk.metrics import MeterProvider -from opentelemetry.sdk.resources import SERVICE_NAME, Resource -from opentelemetry.sdk.trace import Tracer, TracerProvider -from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader +from opentelemetry.sdk.resources import SERVICE_NAME, SERVICE_VERSION, Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter +from opentelemetry.trace import Tracer from prometheus_client import start_http_server +logger = logging.getLogger(__name__) -def initialize_trace(app=None, name=None) -> Tracer: - # Resource can be required for some backends, e.g. Jaeger or Prometheus - # If resource wouldn't be set - traces wouldn't appear in Jaeger. - # The 'name' value is the name shown in Jaeger. - resource = Resource( - attributes={ - SERVICE_NAME: (name if name else os.environ.get("APPLICATION_NAME", "Python App")) - } - ) - # - # 1. Tracing (Jaeger) configuration - # ---------------------------------- - # - # Configure the OTLP exporter (or HTTP exporter the same way) - # the recommended configuration is via environment variables. The mandatory ones: - # OTEL_EXPORTER_OTLP_ENDPOINT: "http://jaeger:4317" - # OTEL_EXPORTER_OTLP_INSECURE: "true" - # for more see here: https://opentelemetry-python.readthedocs.io/en/latest/exporter/otlp/otlp.html - # alternatively these values can be provided as input values to OTLPSpanExporter() constructor. - # - # otlp_exporter = OTLPSpanExporter() - # or provide endpoint explicitly, as - otlp_exporter = OTLPSpanExporter(endpoint="http://jaeger:4317") - # - # Initialize the tracer provider. This is abstract API call. - # The actual implementation is given by the included dependency - # The opentelemetry default is the OTLP trace provider. + +@dataclass +class TracingConfig: + """Configuration for OpenTelemetry tracing.""" + + service_name: str + service_version: str + otlp_endpoint: str = "http://localhost:4317" # Default Jaeger OTLP endpoint + insecure: bool = True + debug: bool = False + additional_attributes: Optional[Dict[str, Any]] = None + + +@dataclass +class MetricsConfig: + """Configuration for OpenTelemetry metrics.""" + + prometheus_port: int = 9464 + prometheus_host: str = "localhost" + export_interval_millis: int = 30000 + additional_attributes: Optional[Dict[str, Any]] = None + + +def create_resource(config: TracingConfig) -> Resource: + """Create an OpenTelemetry resource with service information.""" + attributes = { + SERVICE_NAME: config.service_name, + SERVICE_VERSION: config.service_version, + } + + if config.additional_attributes: + attributes.update(config.additional_attributes) + + return Resource.create(attributes) + + +def setup_otlp_exporter(config: TracingConfig) -> OTLPSpanExporter: + """Configure and create an OTLP exporter for Jaeger.""" + return OTLPSpanExporter(endpoint=config.otlp_endpoint, insecure=config.insecure) + + +def setup_trace_provider(config: TracingConfig, resource: Resource) -> TracerProvider: + """Set up the TracerProvider with OTLP exporter.""" provider = TracerProvider(resource=resource) - # Add the exporters to the batch span processor - span_processor = BatchSpanProcessor(otlp_exporter) - provider.add_span_processor(span_processor) - # Sets the global default tracer provider - trace.set_tracer_provider(provider) - # - # logging.info(' * Instrumenting application..... ') - # if Flask, then use: - # FlaskInstrumentor().instrument_app(flask_app) - # if FastAPI, then use - # FastAPIInstrumentor.instrument_app(app) + # Add OTLP exporter for Jaeger + otlp_exporter = setup_otlp_exporter(config) + provider.add_span_processor(BatchSpanProcessor(otlp_exporter)) - # - # Python Flask is now automatically instrumented to export details of each REST endpoint call. - # For Django based app, use 'opentelemetry-instrumentation-django' import instead. see here: - # https://opentelemetry-python.readthedocs.io/en/latest/examples/django/README.html for example - # - # If the used library has no instrumentation support in opentelemetry.io, then the trace - # can be started manually. In this code, - # 'trace' should be global shared variable, initialized once per application (see tracing.py) - # 'tracer' can be shared global or local variable, application can use multiple uniquely - # named tracers, Jaeger will group the generated traces by the name of the tracer - # that was used to start the Span. See this example code: - # - # tracer = trace.get_tracer(name) - # - # @app.route("/server_request") - # def server_request(): - # with tracer.start_as_current_span( - # "server_request", - # context=extract(request.headers), - # kind=trace.SpanKind.SERVER, - # attributes=collect_request_attributes(request.environ), - # ): - # # do the work here, like logging.info something and return response - # logging.info(request.args.get("param")) - # return "served" - # - # Note: - # the otlp code looks for HTTP header named 'traceparent' that, if present, should contain - # the otlp parent span identification, and the span created here will be made a child - # of that parent span, indicating visually the progression of requests via different services - # for as long as the current span is provided in the traceparent header to next service. + # Add console exporter in debug mode + if config.debug: + provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) - # 2. Metrics (Prometheus) configuration - # ------------------------------------- - # - # Start Prometheus client - start_http_server(port=9464, addr="localhost") - # Initialize PrometheusMetricReader which pulls metrics from the SDK - # on-demand to respond to scrape requests (unlike Jaeger, which worls with push model - application - # pushes traces when those become available to Jaeger's collector. Prometheus works with pull model - # where it periodically polls pre-configured and/or discovered endpoints for metrics. - reader = PrometheusMetricReader() - provider = MeterProvider(resource=resource, metric_readers=[reader]) - metrics.set_meter_provider(provider) + return provider - return trace.get_tracer(name) + +def setup_metrics_provider(config: MetricsConfig, resource: Resource) -> MeterProvider: + """Set up the MeterProvider with Prometheus export.""" + try: + # Start Prometheus HTTP server + start_http_server(port=config.prometheus_port, addr=config.prometheus_host) + logger.info( + f"Started Prometheus metrics server on {config.prometheus_host}:{config.prometheus_port}" + ) + + # Create Prometheus reader + prometheus_reader = PrometheusMetricReader() + + # Create periodic reader for custom metrics + periodic_reader = PeriodicExportingMetricReader( + prometheus_reader, export_interval_millis=config.export_interval_millis + ) + + # Create and configure meter provider + return MeterProvider(resource=resource, metric_readers=[prometheus_reader, periodic_reader]) + except Exception as e: + logger.error(f"Failed to set up metrics provider: {str(e)}") + # Return a no-op meter provider + return MeterProvider(resource=resource) + + +def initialize_telemetry( + tracing_config: Optional[TracingConfig] = None, + metrics_config: Optional[MetricsConfig] = None, +) -> Tracer: + """ + Initialize OpenTelemetry with tracing and metrics. + + Args: + tracing_config: Configuration for tracing + metrics_config: Configuration for metrics + + Returns: + Tracer: Configured OpenTelemetry tracer + + Example: + ```python + config = TracingConfig( + service_name="amq-adapter", + service_version="1.0.0", + otlp_endpoint="http://jaeger:4317", + debug=True + ) + + metrics_config = MetricsConfig( + prometheus_port=9464, + prometheus_host="0.0.0.0" + ) + + tracer = initialize_telemetry(config, metrics_config) + ``` + """ + # Use default configs if none provided + if tracing_config is None: + tracing_config = TracingConfig( + service_name=os.getenv("SERVICE_NAME", "amq-adapter"), + service_version=os.getenv("SERVICE_VERSION", "unknown"), + otlp_endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"), + insecure=os.getenv("OTEL_INSECURE_MODE", "true").lower() == "true", + debug=os.getenv("OTEL_DEBUG", "false").lower() == "true", + ) + + if metrics_config is None: + metrics_config = MetricsConfig( + prometheus_port=int(os.getenv("PROMETHEUS_PORT", "9464")), + prometheus_host=os.getenv("PROMETHEUS_HOST", "localhost"), + export_interval_millis=int(os.getenv("METRICS_EXPORT_INTERVAL", "30000")), + ) + + try: + # Create resource + resource = create_resource(tracing_config) + + # Set up tracing + provider = setup_trace_provider(tracing_config, resource) + trace.set_tracer_provider(provider) + logger.info(f"Initialized tracing for service {tracing_config.service_name}") + + # Set up metrics + metrics_provider = setup_metrics_provider(metrics_config, resource) + metrics.set_meter_provider(metrics_provider) + logger.info("Initialized metrics export") + + # Return configured tracer + return trace.get_tracer( + tracing_config.service_name, + tracing_config.service_version, + ) + + except Exception as e: + logger.error(f"Failed to initialize telemetry: {str(e)}") + # Return a no-op tracer in case of failure + return trace.get_tracer("no-op") + + +def create_span(tracer: Tracer, name: str, attributes: Optional[Dict[str, Any]] = None): + """ + Create a new span with the given name and attributes. + + Args: + tracer: The OpenTelemetry tracer + name: Name of the span + attributes: Optional attributes to add to the span + + Returns: + A context manager that creates and manages a span + """ + return tracer.start_as_current_span( + name, + attributes=attributes or {}, + ) diff --git a/tests/adapter/test_trace_info_adapter.py b/tests/adapter/test_trace_info_adapter.py index 8299acf..4460f08 100644 --- a/tests/adapter/test_trace_info_adapter.py +++ b/tests/adapter/test_trace_info_adapter.py @@ -1,3 +1,251 @@ -class TraceInfoAdapterTest: - def test_create_produce_span(self): - assert False +"""Unit tests for the tracing module.""" + +import unittest +from unittest.mock import MagicMock, patch + +from opentelemetry.sdk.resources import Resource +from opentelemetry.trace import Tracer + +from amqp.service.tracing import ( + MetricsConfig, + TracingConfig, + create_resource, + create_span, + initialize_telemetry, + setup_metrics_provider, + setup_otlp_exporter, + setup_trace_provider, +) + + +class TestTracingConfig(unittest.TestCase): + """Test cases for TracingConfig.""" + + def test_default_values(self): + """Test that default values are set correctly.""" + config = TracingConfig(service_name="test-service", service_version="1.0.0") + self.assertEqual(config.service_name, "test-service") + self.assertEqual(config.service_version, "1.0.0") + self.assertEqual(config.otlp_endpoint, "http://localhost:4317") + self.assertTrue(config.insecure) + self.assertFalse(config.debug) + self.assertIsNone(config.additional_attributes) + + def test_custom_values(self): + """Test that custom values are set correctly.""" + config = TracingConfig( + service_name="test-service", + service_version="1.0.0", + otlp_endpoint="http://custom:4317", + insecure=False, + debug=True, + additional_attributes={"custom": "attribute"}, + ) + self.assertEqual(config.otlp_endpoint, "http://custom:4317") + self.assertFalse(config.insecure) + self.assertTrue(config.debug) + self.assertEqual(config.additional_attributes, {"custom": "attribute"}) + + +class TestMetricsConfig(unittest.TestCase): + """Test cases for MetricsConfig.""" + + def test_default_values(self): + """Test that default values are set correctly.""" + config = MetricsConfig() + self.assertEqual(config.prometheus_port, 9464) + self.assertEqual(config.prometheus_host, "localhost") + self.assertEqual(config.export_interval_millis, 30000) + self.assertIsNone(config.additional_attributes) + + def test_custom_values(self): + """Test that custom values are set correctly.""" + config = MetricsConfig( + prometheus_port=8000, + prometheus_host="0.0.0.0", + export_interval_millis=15000, + additional_attributes={"custom": "attribute"}, + ) + self.assertEqual(config.prometheus_port, 8000) + self.assertEqual(config.prometheus_host, "0.0.0.0") + self.assertEqual(config.export_interval_millis, 15000) + self.assertEqual(config.additional_attributes, {"custom": "attribute"}) + + +class TestCreateResource(unittest.TestCase): + """Test cases for create_resource function.""" + + def test_create_resource_basic(self): + """Test creating a resource with basic configuration.""" + config = TracingConfig(service_name="test-service", service_version="1.0.0") + resource = create_resource(config) + self.assertIsInstance(resource, Resource) + self.assertEqual(resource.attributes.get("service.name"), "test-service") + self.assertEqual(resource.attributes.get("service.version"), "1.0.0") + + def test_create_resource_with_attributes(self): + """Test creating a resource with additional attributes.""" + config = TracingConfig( + service_name="test-service", + service_version="1.0.0", + additional_attributes={"custom": "attribute"}, + ) + resource = create_resource(config) + self.assertEqual(resource.attributes.get("custom"), "attribute") + + +class TestSetupExporters(unittest.TestCase): + """Test cases for exporter setup functions.""" + + @patch("amqp.service.tracing.OTLPSpanExporter") + def test_setup_otlp_exporter(self, mock_otlp): + """Test OTLP exporter setup.""" + config = TracingConfig( + service_name="test-service", + service_version="1.0.0", + otlp_endpoint="http://jaeger:4317", + insecure=True, + ) + setup_otlp_exporter(config) + mock_otlp.assert_called_once_with(endpoint="http://jaeger:4317", insecure=True) + + @patch("amqp.service.tracing.OTLPSpanExporter") + def test_setup_otlp_exporter_secure(self, mock_otlp): + """Test OTLP exporter setup with secure connection.""" + config = TracingConfig( + service_name="test-service", + service_version="1.0.0", + otlp_endpoint="https://jaeger:4317", + insecure=False, + ) + setup_otlp_exporter(config) + mock_otlp.assert_called_once_with(endpoint="https://jaeger:4317", insecure=False) + + +class TestSetupTraceProvider(unittest.TestCase): + """Test cases for trace provider setup.""" + + @patch("amqp.service.tracing.TracerProvider") + @patch("amqp.service.tracing.BatchSpanProcessor") + @patch("amqp.service.tracing.setup_otlp_exporter") + def test_setup_trace_provider_basic(self, mock_otlp_exporter, mock_processor, mock_provider): + """Test basic trace provider setup.""" + config = TracingConfig(service_name="test-service", service_version="1.0.0") + resource = create_resource(config) + + mock_provider_instance = MagicMock() + mock_provider.return_value = mock_provider_instance + + setup_trace_provider(config, resource) + + mock_provider.assert_called_once_with(resource=resource) + mock_otlp_exporter.assert_called_once() + self.assertEqual(mock_provider_instance.add_span_processor.call_count, 1) + + @patch("amqp.service.tracing.TracerProvider") + @patch("amqp.service.tracing.BatchSpanProcessor") + @patch("amqp.service.tracing.setup_otlp_exporter") + @patch("amqp.service.tracing.ConsoleSpanExporter") + def test_setup_trace_provider_with_debug( + self, mock_console_exporter, mock_otlp_exporter, mock_processor, mock_provider + ): + """Test trace provider setup with debug mode.""" + config = TracingConfig(service_name="test-service", service_version="1.0.0", debug=True) + resource = create_resource(config) + + mock_provider_instance = MagicMock() + mock_provider.return_value = mock_provider_instance + + setup_trace_provider(config, resource) + + self.assertEqual(mock_provider_instance.add_span_processor.call_count, 2) + mock_console_exporter.assert_called_once() + + +class TestSetupMetricsProvider(unittest.TestCase): + """Test cases for metrics provider setup.""" + + @patch("amqp.service.tracing.start_http_server") + @patch("amqp.service.tracing.PrometheusMetricReader") + @patch("amqp.service.tracing.PeriodicExportingMetricReader") + @patch("amqp.service.tracing.MeterProvider") + def test_setup_metrics_provider_success( + self, mock_meter_provider, mock_periodic_reader, mock_prometheus_reader, mock_start_server + ): + """Test successful metrics provider setup.""" + config = MetricsConfig(prometheus_port=9464, prometheus_host="localhost") + resource = Resource({}) + + setup_metrics_provider(config, resource) + + mock_start_server.assert_called_once_with(port=9464, addr="localhost") + mock_prometheus_reader.assert_called_once() + mock_periodic_reader.assert_called_once() + mock_meter_provider.assert_called_once() + + @patch("amqp.service.tracing.start_http_server") + def test_setup_metrics_provider_failure(self, mock_start_server): + """Test metrics provider setup with failure.""" + mock_start_server.side_effect = Exception("Failed to start server") + + config = MetricsConfig() + resource = Resource({}) + + provider = setup_metrics_provider(config, resource) + self.assertIsNotNone(provider) # Should return no-op provider + + +class TestInitializeTelemetry(unittest.TestCase): + """Test cases for initialize_telemetry function.""" + + @patch("amqp.service.tracing.setup_trace_provider") + @patch("amqp.service.tracing.setup_metrics_provider") + @patch("amqp.service.tracing.trace") + @patch("amqp.service.tracing.metrics") + def test_initialize_telemetry_success( + self, mock_metrics, mock_trace, mock_setup_metrics, mock_setup_trace + ): + """Test successful telemetry initialization.""" + config = TracingConfig(service_name="test-service", service_version="1.0.0") + metrics_config = MetricsConfig() + + mock_tracer = MagicMock(spec=Tracer) + mock_trace.get_tracer.return_value = mock_tracer + + tracer = initialize_telemetry(config, metrics_config) + + self.assertEqual(tracer, mock_tracer) + mock_setup_trace.assert_called_once() + mock_setup_metrics.assert_called_once() + + @patch("amqp.service.tracing.setup_trace_provider") + def test_initialize_telemetry_failure(self, mock_setup_trace): + """Test telemetry initialization with failure.""" + mock_setup_trace.side_effect = Exception("Setup failed") + + tracer = initialize_telemetry() + + self.assertIsNotNone(tracer) # Should return no-op tracer + + +class TestCreateSpan(unittest.TestCase): + """Test cases for create_span function.""" + + def test_create_span_basic(self): + """Test basic span creation.""" + mock_tracer = MagicMock(spec=Tracer) + create_span(mock_tracer, "test-span") + mock_tracer.start_as_current_span.assert_called_once_with("test-span", attributes={}) + + def test_create_span_with_attributes(self): + """Test span creation with attributes.""" + mock_tracer = MagicMock(spec=Tracer) + attributes = {"custom": "attribute"} + create_span(mock_tracer, "test-span", attributes) + mock_tracer.start_as_current_span.assert_called_once_with( + "test-span", attributes=attributes + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/model/test_model.py b/tests/model/test_model.py new file mode 100644 index 0000000..b636c6a --- /dev/null +++ b/tests/model/test_model.py @@ -0,0 +1,220 @@ +import base64 +import json +import unittest + +from amqp.model.model import ( + AMQErrorMessage, + AMQRoute, + CleverMicroMessage, + DataMessage, + DataResponse, + ScalingRequest, + ScalingRequestAlert, +) +from amqp.model.snowflake_id import SnowflakeId + + +class TestCleverMicroMessage(unittest.TestCase): + def setUp(self): + self.test_id = SnowflakeId() + self.test_headers = {"Content-Type": ["application/json"]} + self.test_trace_info = {"trace_id": "123"} + self.test_body = b"test data" + self.test_base64body = base64.b64encode(self.test_body) + + self.message = CleverMicroMessage( + magic="A", + id=self.test_id, + m_field="GET", + d_field="test_domain", + p_field="test_path", + headers=self.test_headers, + trace_info=self.test_trace_info, + base64body=self.test_base64body, + ) + + def test_getters(self): + self.assertEqual(self.message.magic(), "A") + self.assertEqual(self.message.id(), self.test_id) + self.assertEqual(self.message.m_field(), "GET") + self.assertEqual(self.message.d_field(), "test_domain") + self.assertEqual(self.message.p_field(), "test_path") + self.assertEqual(self.message.headers(), self.test_headers) + self.assertEqual(self.message.trace_info(), self.test_trace_info) + self.assertEqual(self.message.base64body(), self.test_base64body) + + def test_body_decoding(self): + self.assertEqual(self.message.body(), self.test_body) + + def test_content_type(self): + self.assertEqual(self.message.content_type(), "application/json") + + # Test default content type + message_no_content_type = CleverMicroMessage( + magic="A", + id=self.test_id, + m_field="GET", + d_field="test_domain", + p_field="test_path", + headers={}, + trace_info={}, + base64body=self.test_base64body, + ) + self.assertEqual( + message_no_content_type.content_type(), CleverMicroMessage.DEFAULT_CONTENT_TYPE[0] + ) + + +class TestDataMessage(unittest.TestCase): + def setUp(self): + self.test_id = SnowflakeId() + self.test_headers = {"Content-Type": ["application/json"]} + self.test_trace_info = {"trace_id": "123"} + self.test_base64body = base64.b64encode(b"test data") + + self.message = DataMessage( + magic="A", + id=self.test_id, + method="GET", + domain="testdomain", + path="testpath", + headers=self.test_headers, + trace_info=self.test_trace_info, + base64body=self.test_base64body, + ) + + def test_getters(self): + self.assertEqual(self.message.method(), "GET") + self.assertEqual(self.message.domain(), "testdomain") + self.assertEqual(self.message.path(), "testpath") + + def test_routing_key(self): + self.assertEqual(self.message.routing_key(), "testdomain.testpath") + + +class TestDataResponse(unittest.TestCase): + def setUp(self): + self.test_id = SnowflakeId() + self.test_trace_info = {"trace_id": "123"} + self.test_base64body = base64.b64encode(b"test data") + + self.response = DataResponse( + magic="R", + id=self.test_id, + response_code="200", + content_type="application/json", + error="", + error_cause="", + headers={}, + trace_info=self.test_trace_info, + base64body=self.test_base64body, + ) + + def test_getters(self): + self.assertEqual(self.response.response_code(), "200") + self.assertEqual(self.response.error(), "") + self.assertEqual(self.response.error_cause(), "") + self.assertEqual(self.response.content_type(), "application/json") + + +class TestAMQRoute(unittest.TestCase): + def setUp(self): + self.route = AMQRoute( + component_name="test_component", + exchange="test_exchange", + queue="test_queue", + key="test_key", + timeout=1000, + valid_until=1735689600000, # Some future timestamp + ) + + def test_str_representation(self): + expected = "test_component:test_exchange:test_queue:test_key:1000:1735689600000" + self.assertEqual(str(self.route), expected) + + def test_equality(self): + same_route = AMQRoute( + component_name="test_component", + exchange="test_exchange", + queue="test_queue", + key="test_key", + timeout=1000, + valid_until=1735689600000, + ) + different_route = AMQRoute( + component_name="other_component", + exchange="test_exchange", + queue="test_queue", + key="test_key", + timeout=1000, + valid_until=1735689600000, + ) + + self.assertEqual(self.route, same_route) + self.assertNotEqual(self.route, different_route) + + +class TestScalingRequest(unittest.TestCase): + def setUp(self): + self.request = ScalingRequest( + service_id="test_service", + task_id="test_task", + max_availability=5, + current_availability=3, + request_type=ScalingRequestAlert.OVERLOAD, + version=1, + ) + + def test_to_json(self): + json_str = self.request.to_json() + data = json.loads(json_str) + + self.assertEqual(data["serviceId"], "test_service") + self.assertEqual(data["taskId"], "test_task") + self.assertEqual(data["maxAvailability"], 5) + self.assertEqual(data["currentAvailability"], 3) + self.assertEqual(data["requestType"], "OVERLOAD") + self.assertEqual(data["version"], 1) + + def test_from_json(self): + json_str = json.dumps( + { + "serviceId": "test_service", + "taskId": "test_task", + "maxAvailability": 5, + "currentAvailability": 3, + "requestType": "OVERLOAD", + "version": 1, + } + ) + + request = ScalingRequest.from_json(json_str) + + self.assertEqual(request.service_id, "test_service") + self.assertEqual(request.task_id, "test_task") + self.assertEqual(request.max_availability, 5) + self.assertEqual(request.current_availability, 3) + self.assertEqual(request.request_type, ScalingRequestAlert.OVERLOAD) + self.assertEqual(request.version, 1) + + +class TestAMQErrorMessage(unittest.TestCase): + def setUp(self): + self.error = AMQErrorMessage( + response_code=400, + error="Bad Request", + detail="Invalid input", + content_type="application/json", + ) + + def test_serialize(self): + serialized = self.error.serialize() + data = json.loads(serialized) + + self.assertEqual(data["response_code"], 400) + self.assertEqual(data["error"], "Bad Request") + self.assertEqual(data["detail"], "Invalid input") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/requirements-test.txt b/tests/requirements-test.txt new file mode 100644 index 0000000..84925e3 --- /dev/null +++ b/tests/requirements-test.txt @@ -0,0 +1,4 @@ +pytest>=7.0.0 +pytest-cov>=4.0.0 +pytest-asyncio>=0.21.0 +aio-pika>=9.0.0