Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """ | |
| Test script to verify ENABLE_TRACING toggle works correctly. | |
| Usage: | |
| # Test with tracing enabled (default) | |
| python test_tracing_toggle.py | |
| # Test with tracing disabled | |
| ENABLE_TRACING=false python test_tracing_toggle.py | |
| """ | |
| import os | |
| import sys | |
| # Add src to path | |
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) | |
| from utils.tracing import is_tracing_enabled, init_tracer | |
| def test_is_tracing_enabled(): | |
| """Test the is_tracing_enabled function with various env values.""" | |
| print("Testing is_tracing_enabled()...") | |
| test_cases = [ | |
| ("true", True), | |
| ("True", True), | |
| ("TRUE", True), | |
| ("1", True), | |
| ("yes", True), | |
| ("YES", True), | |
| ("false", False), | |
| ("False", False), | |
| ("FALSE", False), | |
| ("0", False), | |
| ("no", False), | |
| ("NO", False), | |
| ("", True), # Default is enabled | |
| (None, True), # Default is enabled | |
| ] | |
| for value, expected in test_cases: | |
| # Set or unset environment variable | |
| if value is None: | |
| os.environ.pop("ENABLE_TRACING", None) | |
| else: | |
| os.environ["ENABLE_TRACING"] = value | |
| result = is_tracing_enabled() | |
| status = "β" if result == expected else "β" | |
| print(f" {status} ENABLE_TRACING={value!r} β {result} (expected {expected})") | |
| if result != expected: | |
| return False | |
| return True | |
| def test_init_tracer(): | |
| """Test that init_tracer respects the ENABLE_TRACING flag.""" | |
| print("\nTesting init_tracer()...") | |
| # Test with tracing disabled | |
| os.environ["ENABLE_TRACING"] = "false" | |
| os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://localhost:4318" | |
| print(" Testing with ENABLE_TRACING=false...") | |
| tracer = init_tracer("test-service") | |
| # Should return a tracer (no-op) but not initialize OpenTelemetry | |
| if tracer is None: | |
| print(" β init_tracer returned None (should return no-op tracer)") | |
| return False | |
| print(" β init_tracer returned tracer (no-op)") | |
| # Test with tracing enabled | |
| os.environ["ENABLE_TRACING"] = "true" | |
| print(" Testing with ENABLE_TRACING=true...") | |
| # Note: This will actually try to connect to Jaeger if available | |
| # but won't fail if Jaeger is down | |
| try: | |
| tracer = init_tracer("test-service-enabled") | |
| print(" β init_tracer returned tracer (full initialization)") | |
| except Exception as e: | |
| print(f" β init_tracer failed: {e}") | |
| print(" (This is expected if Jaeger is not running)") | |
| return True | |
| def test_span_creation(): | |
| """Test that spans are created/skipped based on toggle.""" | |
| print("\nTesting span creation...") | |
| from opentelemetry import trace | |
| # Test with tracing disabled | |
| os.environ["ENABLE_TRACING"] = "false" | |
| if is_tracing_enabled(): | |
| print(" β is_tracing_enabled() returned True when ENABLE_TRACING=false") | |
| return False | |
| # This should be a no-op | |
| tracer = trace.get_tracer(__name__) | |
| span = tracer.start_span("test-span") | |
| span.end() | |
| print(" β No-op span created successfully when disabled") | |
| # Test with tracing enabled | |
| os.environ["ENABLE_TRACING"] = "true" | |
| if not is_tracing_enabled(): | |
| print(" β is_tracing_enabled() returned False when ENABLE_TRACING=true") | |
| return False | |
| print(" β Tracing enabled check passed") | |
| return True | |
| def main(): | |
| """Run all tests.""" | |
| print("=" * 60) | |
| print("ENABLE_TRACING Toggle Test Suite") | |
| print("=" * 60) | |
| current_value = os.environ.get("ENABLE_TRACING", "<not set>") | |
| print(f"\nCurrent ENABLE_TRACING: {current_value}") | |
| print() | |
| tests = [ | |
| ("is_tracing_enabled()", test_is_tracing_enabled), | |
| ("init_tracer()", test_init_tracer), | |
| ("span creation", test_span_creation), | |
| ] | |
| results = [] | |
| for name, test_func in tests: | |
| try: | |
| success = test_func() | |
| results.append((name, success)) | |
| except Exception as e: | |
| print(f"\nβ Test '{name}' raised exception: {e}") | |
| results.append((name, False)) | |
| # Print summary | |
| print("\n" + "=" * 60) | |
| print("Test Summary") | |
| print("=" * 60) | |
| passed = sum(1 for _, success in results if success) | |
| total = len(results) | |
| for name, success in results: | |
| status = "β PASS" if success else "β FAIL" | |
| print(f"{status}: {name}") | |
| print(f"\n{passed}/{total} tests passed") | |
| if passed == total: | |
| print("\nβ All tests passed!") | |
| return 0 | |
| else: | |
| print(f"\nβ {total - passed} test(s) failed") | |
| return 1 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |